Skip to content

Commit 655b844

Browse files
sbalabanov-zzclaude
andcommitted
fix(stovepipe): fail validations that exceed max_validation_ms
buildsignal held a nonterminal build unconditionally, and holds do not consume the retry budget, so a runner that kept answering accepted or running was re-polled forever. The request never dead-lettered and, at MaxConcurrent 1, kept the queue's only build slot. Record Build.CreatedAt and add QueueConfig.MaxValidationMs (two hours in the reference config; zero disables it). A nonterminal poll of a nonterminal build past the deadline persists the build as failed through the write-once reconcile and runs the normal terminal path, marking the request failed with reason validation_timeout. A redelivery finds the stored terminal build and re-logs the stored reason. Existing MySQL deployments must add created_at to build before deploying. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1 parent d42b7fd commit 655b844

21 files changed

Lines changed: 153 additions & 33 deletions

File tree

‎doc/rfc/stovepipe/steps/build.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ For a delivery carrying request id `R`:
5252
either domain — the shape is deferred until then, not decided here.
5353
- failure -> return raw; classifier decides (transient runner blip retryable, bad URI not).
5454
55-
6. Persist Build{ID: buildID.ID, RequestID: R.ID, Status: accepted, Version: 1}
55+
6. Persist Build{ID: buildID.ID, RequestID: R.ID, Status: accepted, CreatedAt: now, Version: 1}
5656
via BuildStore.Create.
5757
- the row carries no scope; it is recoverable from the Request's immutable fields
5858
(see the entity table).
@@ -301,6 +301,7 @@ Either could be adopted independently: the idempotency token, if a backend that
301301
| `ID` | The build's own key — the runner-assigned id returned by `Trigger` (a Buildkite build number, a CI-gateway job id); opaque, never parsed or derived | no |
302302
| `RequestID` | The `Request` this build validates (`Build`→`Request` navigation) | no |
303303
| `Status` | `accepted / running / succeeded / failed / cancelled` | **yes** — `buildsignal` |
304+
| `CreatedAt` | Unix milliseconds when the row was created; zero on rows created before it was recorded | no |
304305
| `Version` | `int32` optimistic-locking version | **yes** — with `Status` |
305306

306307
The row deliberately carries no scope: `R.URI`, `R.BaseURI`, and `R.BuildStrategy` — immutable and reachable through `RequestID` — fully determine what a build ran against.

‎doc/rfc/stovepipe/steps/buildsignal.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,8 @@ For a delivery carrying build id `B`:
5151
see "Status" below for what it's for and how a future check could use it.
5252
5353
6. Reconcile the polled status against the stored row:
54+
- status and Build.Status both non-terminal, and max_validation_ms elapsed since
55+
Build.CreatedAt -> reconcile failed in place of status (see Fail-closed interaction).
5456
- status == Build.Status -> no-op poll: skip the write, continue with the stored status.
5557
- Build.Status already terminal, status differs -> terminal is WRITE-ONCE: do not overwrite;
5658
continue with the STORED status as authoritative (see Edge cases).
@@ -165,7 +167,7 @@ The window to guard is between persisting status (step 6) and ack (steps 7–8);
165167

166168
A build that never reaches terminal `Status` — runner outage, a build the runner lost — must not wedge its `Request` forever, since callers gate deployments on greenness reaching a recorded terminal state. `buildsignal` does not implement the forcing function: per [workflow.md](../workflow.md#fail-closed-on-unprocessable-work) and the `in_flight_count` slot lifecycle in [process.md](process.md#concurrency-lifecycle), a `Request` stuck at `buildsignal` past `MaxAttempts` dead-letters, and the DLQ reconciler forces a conservative terminal `failed` and releases the Queue's slot. This is the same posture SubmitQueue's build/buildsignal pair relies on: terminal status is what releases the slot and lets validation progress.
167169

168-
One boundary of that posture is worth stating: the `MaxAttempts` path fires only when polls *fail*. A runner that keeps answering a healthy non-terminal status forever — a hung build on a backend with no timeout of its own — never errors, so the hold chain (whose redeliveries deliberately do not count toward the retry limit) re-polls indefinitely and nothing dead-letters; SubmitQueue's poll loop shares this property. Bounding it requires a poll deadline — a `max_validation_ms` past which `buildsignal` treats the build as failed and lets the normal terminal path run — which pairs naturally with the lease idea [process.md](process.md#per-queue-concurrency-gate) floats for `in_flight_count`. Deferred with it; until then a too-old non-terminal `Build` is an operational alert, not a self-healing path.
170+
The `MaxAttempts` path fires only when polls *fail*. A runner that keeps answering a healthy non-terminal status — a hung build on a backend with no timeout of its own — never errors, and hold redeliveries deliberately do not count toward the retry limit, so a deadline bounds that case instead: the per-queue `max_validation_ms` (`QueueConfig.MaxValidationMs`) caps the time from `Build.CreatedAt` to a terminal status. A non-terminal poll of a non-terminal `Build` past the deadline treats the build as `failed`. Step 6 persists `failed` under the write-once rule, so a later poll cannot reopen it and a redelivery resumes from the stored status. Step 7 then runs exactly as for a polled failure, except that the request records the reason `validation_timeout`. Zero disables the deadline, as does a zero `CreatedAt` on a `Build` created before creation times were recorded. The runner's build is not cancelled (see [build.md](build.md#cancellation-defined-not-yet-called)) and may keep running on its backend. SubmitQueue's poll loop has no equivalent deadline.
169171

170172
## Entity, storage, and queue additions
171173

‎doc/rfc/stovepipe/steps/process.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ A slot is held from admit until the build goes terminal (`process → build →
6767

6868
**Liveness — a stuck Request wedges the whole Queue.** The slot is shared per-Queue, so a Request admitted but never driven terminal holds the only slot and stalls the whole Queue. The fail-closed path in [workflow.md](../workflow.md#fail-closed-on-unprocessable-work) prevents this: every admitted Request terminates — `build`/`buildsignal` errors retry to `MaxAttempts`, then dead-letter, and the DLQ reconciler forces a conservative terminal `failed`, freeing the slot. Residual risk is operational: a **poison DLQ message** the reconciler can't process loops forever (it runs always-retryable), wedging that Queue until an operator removes it. The gate makes the blast radius the whole Queue, not one Request — monitor and alert on the DLQ.
6969

70-
**Future alternative — time-bounded leases.** To self-heal instead of relying on operator cleanup, `in_flight_count` could become a set of `{owner, expires_at}` leases: the gate counts only unexpired leases, terminal transitions drop the owner's lease, and a leaked lease is reclaimed on expiry — bounding any stall to a `max_validation_ms`. A *list* of leases would also generalize to `max_concurrent > 1`. Deferred; `in_flight_count` plus the fail-closed path is enough for MVP.
70+
**Future alternative — time-bounded leases.** To self-heal instead of relying on operator cleanup, `in_flight_count` could become a set of `{owner, expires_at}` leases: the gate counts only unexpired leases, terminal transitions drop the owner's lease, and a leaked lease is reclaimed on expiry — bounding a stall whose request never reaches a polled build, which `buildsignal`'s `max_validation_ms` deadline cannot (see [buildsignal.md](buildsignal.md#fail-closed-interaction)). A *list* of leases would also generalize to `max_concurrent > 1`. Deferred; `in_flight_count` plus the fail-closed path is enough for MVP.
7171

7272
`max_concurrent` is a scalar, and `= 1` is an MVP simplicity choice, **not** a correctness requirement — see [Raising `max_concurrent`](#raising-max_concurrent-speculative-validation).
7373

‎service/stovepipe/server/main.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -403,7 +403,7 @@ func registerPrimaryControllers(
403403
}
404404
count++
405405

406-
buildSignalController := buildsignal.NewController(logger, scope, store, materializer, brf, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal")
406+
buildSignalController := buildsignal.NewController(logger, scope, store, materializer, queueconfigdefault.NewStore(), brf, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal")
407407
if err := c.Register(buildSignalController); err != nil {
408408
return count, fmt.Errorf("failed to register buildsignal controller: %w", err)
409409
}

‎stovepipe/controller/build/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ go_test(
4848
"//stovepipe/extension/buildissue/mock:go_default_library",
4949
"//stovepipe/extension/buildrunner:go_default_library",
5050
"//stovepipe/extension/buildrunner/mock:go_default_library",
51+
"//stovepipe/extension/queueconfig/mock:go_default_library",
5152
"//stovepipe/extension/storage:go_default_library",
5253
"//stovepipe/extension/storage/mock:go_default_library",
5354
"@com_github_stretchr_testify//assert:go_default_library",

‎stovepipe/controller/build/build.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ package build
2121
import (
2222
"context"
2323
"fmt"
24+
"time"
2425

2526
"github.com/uber-go/tally"
2627
entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
@@ -49,6 +50,7 @@ type Controller struct {
4950
registry consumer.TopicRegistry
5051
topicKey consumer.TopicKey
5152
consumerGroup string
53+
now func() time.Time
5254
}
5355

5456
// _opName is the metric operation name shared by every emit in this file.
@@ -74,6 +76,7 @@ func NewController(
7476
registry: registry,
7577
topicKey: topicKey,
7678
consumerGroup: consumerGroup,
79+
now: time.Now,
7780
}
7881
}
7982

@@ -143,6 +146,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
143146
ID: buildID.ID,
144147
RequestID: request.ID,
145148
Status: entity.BuildStatusAccepted,
149+
CreatedAt: c.now().UnixMilli(),
146150
Version: 1,
147151
}
148152
if err := store.GetBuildStore().Create(ctx, build); err != nil {

‎stovepipe/controller/build/build_analysis_test.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import (
3030
"github.com/uber/submitqueue/stovepipe/extension/buildissue"
3131
buildissuemock "github.com/uber/submitqueue/stovepipe/extension/buildissue/mock"
3232
"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
33+
queueconfigmock "github.com/uber/submitqueue/stovepipe/extension/queueconfig/mock"
3334
"github.com/uber/submitqueue/stovepipe/extension/storage"
3435
storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock"
3536
"go.uber.org/mock/gomock"
@@ -121,7 +122,7 @@ func TestBuildAnalysisPipeline(t *testing.T) {
121122
analysisController := buildanalysis.NewController(zap.NewNop().Sugar(), tally.NoopScope,
122123
staticStorageFactory{store: m.store}, detectorFactory, buildController.registry, stovepipemq.TopicKeyBuildAnalysis, "analysis")
123124
statusController := buildsignal.NewController(zap.NewNop().Sugar(), tally.NoopScope,
124-
staticStorageFactory{store: m.store}, m.materializer, m.runnerFactory, buildController.registry,
125+
staticStorageFactory{store: m.store}, m.materializer, queueconfigmock.NewMockStore(ctrl), m.runnerFactory, buildController.registry,
125126
stovepipemq.TopicKeyBuildSignal, "signal")
126127

127128
delivered := func(msg entityqueue.Message) *consumermock.MockDelivery {

‎stovepipe/controller/build/build_test.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"database/sql/driver"
2121
"errors"
2222
"testing"
23+
"time"
2324

2425
"github.com/stretchr/testify/assert"
2526
"github.com/stretchr/testify/require"
@@ -50,6 +51,7 @@ const (
5051
testHeadURI = "git://repo/monorepo/main/head"
5152
testBaseURI = "git://repo/monorepo/main/base"
5253
testBuildID = "bk-1"
54+
testNowMs = int64(1700000000000)
5355
)
5456

5557
func queueContext() context.Context {
@@ -105,6 +107,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMoc
105107
require.NoError(t, err)
106108

107109
c := NewController(zap.NewNop().Sugar(), scope, staticStorageFactory{store: m.store}, m.materializer, m.runnerFactory, registry, stovepipemq.TopicKeyBuild, "stovepipe-build")
110+
c.now = func() time.Time { return time.UnixMilli(testNowMs) }
108111
return c, m
109112
}
110113

@@ -225,6 +228,7 @@ func TestProcess(t *testing.T) {
225228
ID: testBuildID,
226229
RequestID: testID,
227230
Status: entity.BuildStatusAccepted,
231+
CreatedAt: testNowMs,
228232
Version: 1,
229233
}
230234
createCall := m.buildStore.EXPECT().Create(gomock.Any(), build).Return(nil)
@@ -244,6 +248,7 @@ func TestProcess(t *testing.T) {
244248
ID: testBuildID,
245249
RequestID: testID,
246250
Status: entity.BuildStatusAccepted,
251+
CreatedAt: testNowMs,
247252
Version: 1,
248253
}
249254
createCall := m.buildStore.EXPECT().Create(gomock.Any(), build).Return(nil)

‎stovepipe/controller/buildsignal/BUILD.bazel‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ go_library(
1616
"//stovepipe/core/requestlog:go_default_library",
1717
"//stovepipe/entity:go_default_library",
1818
"//stovepipe/extension/buildrunner:go_default_library",
19+
"//stovepipe/extension/queueconfig:go_default_library",
1920
"//stovepipe/extension/storage:go_default_library",
2021
"@com_github_uber_go_tally//:go_default_library",
2122
"@org_uber_go_zap//:go_default_library",
@@ -41,6 +42,7 @@ go_test(
4142
"//stovepipe/entity:go_default_library",
4243
"//stovepipe/extension/buildrunner:go_default_library",
4344
"//stovepipe/extension/buildrunner/mock:go_default_library",
45+
"//stovepipe/extension/queueconfig/mock:go_default_library",
4446
"//stovepipe/extension/storage:go_default_library",
4547
"//stovepipe/extension/storage/mock:go_default_library",
4648
"@com_github_stretchr_testify//assert:go_default_library",

‎stovepipe/controller/buildsignal/buildsignal.go‎

Lines changed: 53 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,8 @@
1414

1515
// Package buildsignal holds the buildsignal-stage queue controller. It consumes
1616
// BuildSignal messages (a build id), polls the build-runner until the build
17-
// reaches a terminal status, persists that status, and — once terminal —
17+
// reaches a terminal status or outlives the queue's max_validation_ms (which
18+
// fails it), persists that status, and — once terminal —
1819
// releases the queue's build slot, projects the outcome onto the request,
1920
// advances the successful-build bookmark, and publishes directly to record plus
2021
// terminal analysis. See
@@ -25,6 +26,7 @@ import (
2526
"context"
2627
"errors"
2728
"fmt"
29+
"time"
2830

2931
"github.com/uber-go/tally"
3032
entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
@@ -37,6 +39,7 @@ import (
3739
"github.com/uber/submitqueue/stovepipe/core/requestlog"
3840
"github.com/uber/submitqueue/stovepipe/entity"
3941
"github.com/uber/submitqueue/stovepipe/extension/buildrunner"
42+
"github.com/uber/submitqueue/stovepipe/extension/queueconfig"
4043
"github.com/uber/submitqueue/stovepipe/extension/storage"
4144
"go.uber.org/zap"
4245
)
@@ -65,6 +68,7 @@ type Controller struct {
6568
metricsScope tally.Scope
6669
stores storage.Factory
6770
materializer requestlog.Materializer
71+
queueConfigs queueconfig.Store
6872
buildRunners buildrunner.Factory
6973
registry consumer.TopicRegistry
7074
topicKey consumer.TopicKey
@@ -80,6 +84,7 @@ func NewController(
8084
scope tally.Scope,
8185
stores storage.Factory,
8286
materializer requestlog.Materializer,
87+
queueConfigs queueconfig.Store,
8388
buildRunners buildrunner.Factory,
8489
registry consumer.TopicRegistry,
8590
topicKey consumer.TopicKey,
@@ -90,6 +95,7 @@ func NewController(
9095
metricsScope: scope.SubScope("buildsignal_controller"),
9196
stores: stores,
9297
materializer: materializer,
98+
queueConfigs: queueConfigs,
9399
buildRunners: buildRunners,
94100
registry: registry,
95101
topicKey: topicKey,
@@ -167,16 +173,33 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
167173
return fmt.Errorf("failed to poll status for build %s: %w", build.ID, err)
168174
}
169175

176+
// Checked before reconcile so a timeout persists as the build's write-once terminal
177+
// status the same way a polled failure does.
178+
timedOut := false
179+
if !status.IsTerminal() && !build.Status.IsTerminal() {
180+
if timedOut, err = c.validationDeadlinePassed(ctx, request, build); err != nil {
181+
return err
182+
}
183+
if timedOut {
184+
status = entity.BuildStatusFailed
185+
}
186+
}
187+
170188
effective, err := c.reconcile(ctx, store, build, status)
171189
if err != nil {
172190
return err
173191
}
174192

175193
if effective.IsTerminal() {
194+
state := outcomeState(effective)
195+
reason := buildOutcomeReason(state)
196+
if timedOut {
197+
reason = entity.RequestOutcomeReasonValidationTimeout
198+
}
176199
if err := c.persistBuildFinishedLog(ctx, store, request, build.ID); err != nil {
177200
return err
178201
}
179-
if err := c.finishRequest(ctx, store, &request, build.ID, effective); err != nil {
202+
if err := c.finishRequest(ctx, store, &request, build.ID, state, reason); err != nil {
180203
return err
181204
}
182205
if request.State == entity.RequestStateSucceeded {
@@ -232,8 +255,8 @@ func (c *Controller) persistBuildFinishedLog(ctx context.Context, store storage.
232255
return nil
233256
}
234257

235-
// finishRequest releases the queue's build slot and projects the build's
236-
// terminal status and ID onto the request, leaving it terminal. It is a no-op
258+
// finishRequest releases the queue's build slot and records state, reason, and
259+
// the build's ID on the request, leaving it terminal. It is a no-op
237260
// when the request already carries an outcome, so a redelivery neither
238261
// double-releases the slot nor rewrites the state.
239262
//
@@ -245,7 +268,7 @@ func (c *Controller) persistBuildFinishedLog(ctx context.Context, store storage.
245268
// the request non-terminal, so redelivery re-runs both steps and decrements again
246269
// — transiently over-admitting by one until releaseBuildSlot's zero clamp
247270
// reconverges, which is the failure mode this pipeline prefers.
248-
func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, request *entity.Request, buildID string, status entity.BuildStatus) error {
271+
func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, request *entity.Request, buildID string, state entity.RequestState, reason entity.RequestOutcomeReason) error {
249272
if request.State.HasBuildOutcome() {
250273
return nil
251274
}
@@ -255,14 +278,37 @@ func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, r
255278
return err
256279
}
257280

258-
state := outcomeState(status)
259-
if err := c.markOutcome(ctx, store, request, buildID, state, buildOutcomeReason(state)); err != nil {
281+
if err := c.markOutcome(ctx, store, request, buildID, state, reason); err != nil {
260282
metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...)
261283
return err
262284
}
263285
return nil
264286
}
265287

288+
// validationDeadlinePassed reports whether build has outlived its queue's max_validation_ms.
289+
// A zero limit or a zero creation time means the build has no deadline.
290+
func (c *Controller) validationDeadlinePassed(ctx context.Context, request entity.Request, build entity.Build) (bool, error) {
291+
if build.CreatedAt == 0 {
292+
return false, nil
293+
}
294+
cfg, err := c.queueConfigs.Get(ctx, request.Queue)
295+
if err != nil {
296+
return false, fmt.Errorf("failed to load queue config for %s: %w", request.Queue, err)
297+
}
298+
if cfg.MaxValidationMs <= 0 || time.Now().UnixMilli()-build.CreatedAt < cfg.MaxValidationMs {
299+
return false, nil
300+
}
301+
302+
metrics.NamedCounter(c.metricsScope, _opName, "validation_timeouts", 1, metrics.TagsFromContext(ctx)...)
303+
c.logger.Warnw("failing build that exceeded the queue's max validation time",
304+
"build_id", build.ID,
305+
"request_id", request.ID,
306+
"created_at", build.CreatedAt,
307+
"max_validation_ms", cfg.MaxValidationMs,
308+
)
309+
return true, nil
310+
}
311+
266312
func (c *Controller) persistOutcomeLog(ctx context.Context, store storage.Storage, request entity.Request) error {
267313
if !request.State.HasBuildOutcome() {
268314
return fmt.Errorf("request %s has no build outcome to record", request.ID)

0 commit comments

Comments
 (0)