diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md index 48400455c..001cc5719 100644 --- a/doc/rfc/stovepipe/request-log.md +++ b/doc/rfc/stovepipe/request-log.md @@ -215,7 +215,7 @@ The `project_facts_recorded` event also projects its retained timestamp into `Re | Ingest | Create accepted Request, then retain accepted. | An existing Request ensures accepted before process publication. | | Process | CAS to superseded or processing, then retain that state. | An existing state is reconstructed from Request context before ack or build publication. | | Build | Create Build after runner acceptance, then retain `build_triggered`. | An identical existing Build ensures the event before buildsignal publication. | -| Buildsignal | Persist terminal Build and retain `build_finished`; CAS the Request outcome and retain its terminal state. | Existing terminal Build and Request outcome each ensure their own entry before record publication. | +| Buildsignal | Persist terminal Build and retain `build_finished`; CAS the Request outcome and retain its terminal state. A validation timeout retains only the terminal state entry. | Existing terminal Build and Request outcome each ensure their own entry before record publication. | | Record | Create or verify the whole-repository fact, then retain `validation_fact_recorded`; after recording project facts, retain one `project_facts_recorded` event. | An identical whole-repository fact and project-fact batch ensure their events before bookmark or promotion work. | | Record DLQ | Retain `record_abandoned`; for failed Requests, create or verify the repository fact and retain `validation_fact_recorded` before publishing its hook. | Existing facts remain authoritative; stable log and hook IDs make replay idempotent without resolving projects or replaying promotion. | | Build analysis DLQ | Retain `build_analysis_abandoned` without rerunning detection or changing validation state. | One stable event per build deduplicates recovery deliveries; only state reads and history retention retry. | diff --git a/doc/rfc/stovepipe/steps/build.md b/doc/rfc/stovepipe/steps/build.md index 7fb415d74..3aa8a36ff 100644 --- a/doc/rfc/stovepipe/steps/build.md +++ b/doc/rfc/stovepipe/steps/build.md @@ -52,7 +52,7 @@ For a delivery carrying request id `R`: either domain — the shape is deferred until then, not decided here. - failure -> return raw; classifier decides (transient runner blip retryable, bad URI not). -6. Persist Build{ID: buildID.ID, RequestID: R.ID, Status: accepted, Version: 1} +6. Persist Build{ID: buildID.ID, RequestID: R.ID, Status: accepted, CreatedAt: now, Version: 1} via BuildStore.Create. - the row carries no scope; it is recoverable from the Request's immutable fields (see the entity table). @@ -301,6 +301,7 @@ Either could be adopted independently: the idempotency token, if a backend that | `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 | | `RequestID` | The `Request` this build validates (`Build`→`Request` navigation) | no | | `Status` | `accepted / running / succeeded / failed / cancelled` | **yes** — `buildsignal` | +| `CreatedAt` | Unix milliseconds when the row was created; zero on rows created before it was recorded | no | | `Version` | `int32` optimistic-locking version | **yes** — with `Status` | 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. diff --git a/doc/rfc/stovepipe/steps/buildsignal.md b/doc/rfc/stovepipe/steps/buildsignal.md index cf0b7f78b..a57522eb1 100644 --- a/doc/rfc/stovepipe/steps/buildsignal.md +++ b/doc/rfc/stovepipe/steps/buildsignal.md @@ -59,6 +59,10 @@ For a delivery carrying build id `B`: - ErrVersionMismatch -> return with its declaration-level retryable classification (a concurrent writer moved the row; reload and re-check). - with the write-once rule, accepted -> running -> {succeeded|failed|cancelled} is monotonic by mechanism, not by assumption about the backend. + - stored status still non-terminal and max_validation_ms elapsed since Build.CreatedAt -> + validation timeout: the Build keeps that status, and step 7 runs without a build_finished + entry or terminal analysis publish, recording R failed with reason validation_timeout and + no terminal build id (see Fail-closed interaction). 7. If the stored status is terminal: a. If R does not already carry an outcome, release the queue's build slot: @@ -165,7 +169,7 @@ The window to guard is between persisting status (step 6) and ack (steps 7–8); 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. -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. +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. When a poll leaves the `Build` non-terminal past the deadline, step 7 runs as for a polled failure, except that the request records the reason `validation_timeout` and no terminal build id, no `build_finished` entry is retained, and no terminal analysis is published, since the build established nothing. The `Build` keeps its last polled status, because the runner reported no failure; nothing lists non-terminal builds, and the terminal request is never reopened. A retry after a failure partway through polls again, finds the build still non-terminal and past the deadline, and takes the same path with the same reason, while step 7a–b no-op once the outcome is recorded. If the build reached a real terminal status in the meantime and the request has no outcome yet, that verdict is recorded instead. 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. ## Entity, storage, and queue additions diff --git a/doc/rfc/stovepipe/steps/process.md b/doc/rfc/stovepipe/steps/process.md index b1089fbe0..39cf7530e 100644 --- a/doc/rfc/stovepipe/steps/process.md +++ b/doc/rfc/stovepipe/steps/process.md @@ -67,7 +67,7 @@ A slot is held from admit until the build goes terminal (`process → build → **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. -**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. +**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. `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). diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 509122734..e2de6e841 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -403,7 +403,7 @@ func registerPrimaryControllers( } count++ - buildSignalController := buildsignal.NewController(logger, scope, store, materializer, brf, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal") + buildSignalController := buildsignal.NewController(logger, scope, store, materializer, queueconfigdefault.NewStore(), brf, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal") if err := c.Register(buildSignalController); err != nil { return count, fmt.Errorf("failed to register buildsignal controller: %w", err) } diff --git a/stovepipe/controller/build/BUILD.bazel b/stovepipe/controller/build/BUILD.bazel index 540880e2a..1422ecd23 100644 --- a/stovepipe/controller/build/BUILD.bazel +++ b/stovepipe/controller/build/BUILD.bazel @@ -48,6 +48,7 @@ go_test( "//stovepipe/extension/buildissue/mock:go_default_library", "//stovepipe/extension/buildrunner:go_default_library", "//stovepipe/extension/buildrunner/mock:go_default_library", + "//stovepipe/extension/queueconfig/mock:go_default_library", "//stovepipe/extension/storage:go_default_library", "//stovepipe/extension/storage/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/stovepipe/controller/build/build.go b/stovepipe/controller/build/build.go index eae495190..7bf78b9bf 100644 --- a/stovepipe/controller/build/build.go +++ b/stovepipe/controller/build/build.go @@ -21,6 +21,7 @@ package build import ( "context" "fmt" + "time" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" @@ -49,6 +50,7 @@ type Controller struct { registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string + now func() time.Time } // _opName is the metric operation name shared by every emit in this file. @@ -74,6 +76,7 @@ func NewController( registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, + now: time.Now, } } @@ -143,6 +146,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er ID: buildID.ID, RequestID: request.ID, Status: entity.BuildStatusAccepted, + CreatedAt: c.now().UnixMilli(), Version: 1, } if err := store.GetBuildStore().Create(ctx, build); err != nil { diff --git a/stovepipe/controller/build/build_analysis_test.go b/stovepipe/controller/build/build_analysis_test.go index 0609168d0..b95fa4cbc 100644 --- a/stovepipe/controller/build/build_analysis_test.go +++ b/stovepipe/controller/build/build_analysis_test.go @@ -30,6 +30,7 @@ import ( "github.com/uber/submitqueue/stovepipe/extension/buildissue" buildissuemock "github.com/uber/submitqueue/stovepipe/extension/buildissue/mock" "github.com/uber/submitqueue/stovepipe/extension/buildrunner" + queueconfigmock "github.com/uber/submitqueue/stovepipe/extension/queueconfig/mock" "github.com/uber/submitqueue/stovepipe/extension/storage" storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" "go.uber.org/mock/gomock" @@ -121,7 +122,7 @@ func TestBuildAnalysisPipeline(t *testing.T) { analysisController := buildanalysis.NewController(zap.NewNop().Sugar(), tally.NoopScope, staticStorageFactory{store: m.store}, detectorFactory, buildController.registry, stovepipemq.TopicKeyBuildAnalysis, "analysis") statusController := buildsignal.NewController(zap.NewNop().Sugar(), tally.NoopScope, - staticStorageFactory{store: m.store}, m.materializer, m.runnerFactory, buildController.registry, + staticStorageFactory{store: m.store}, m.materializer, queueconfigmock.NewMockStore(ctrl), m.runnerFactory, buildController.registry, stovepipemq.TopicKeyBuildSignal, "signal") delivered := func(msg entityqueue.Message) *consumermock.MockDelivery { diff --git a/stovepipe/controller/build/build_test.go b/stovepipe/controller/build/build_test.go index e98089e8e..2cc5e59b5 100644 --- a/stovepipe/controller/build/build_test.go +++ b/stovepipe/controller/build/build_test.go @@ -20,6 +20,7 @@ import ( "database/sql/driver" "errors" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -50,6 +51,7 @@ const ( testHeadURI = "git://repo/monorepo/main/head" testBaseURI = "git://repo/monorepo/main/base" testBuildID = "bk-1" + testNowMs = int64(1700000000000) ) func queueContext() context.Context { @@ -105,6 +107,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMoc require.NoError(t, err) c := NewController(zap.NewNop().Sugar(), scope, staticStorageFactory{store: m.store}, m.materializer, m.runnerFactory, registry, stovepipemq.TopicKeyBuild, "stovepipe-build") + c.now = func() time.Time { return time.UnixMilli(testNowMs) } return c, m } @@ -225,6 +228,7 @@ func TestProcess(t *testing.T) { ID: testBuildID, RequestID: testID, Status: entity.BuildStatusAccepted, + CreatedAt: testNowMs, Version: 1, } createCall := m.buildStore.EXPECT().Create(gomock.Any(), build).Return(nil) @@ -244,6 +248,7 @@ func TestProcess(t *testing.T) { ID: testBuildID, RequestID: testID, Status: entity.BuildStatusAccepted, + CreatedAt: testNowMs, Version: 1, } createCall := m.buildStore.EXPECT().Create(gomock.Any(), build).Return(nil) diff --git a/stovepipe/controller/buildsignal/BUILD.bazel b/stovepipe/controller/buildsignal/BUILD.bazel index 77ac7b005..96e65a7ea 100644 --- a/stovepipe/controller/buildsignal/BUILD.bazel +++ b/stovepipe/controller/buildsignal/BUILD.bazel @@ -16,6 +16,7 @@ go_library( "//stovepipe/core/requestlog:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/buildrunner:go_default_library", + "//stovepipe/extension/queueconfig:go_default_library", "//stovepipe/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", @@ -41,6 +42,7 @@ go_test( "//stovepipe/entity:go_default_library", "//stovepipe/extension/buildrunner:go_default_library", "//stovepipe/extension/buildrunner/mock:go_default_library", + "//stovepipe/extension/queueconfig/mock:go_default_library", "//stovepipe/extension/storage:go_default_library", "//stovepipe/extension/storage/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go index 5adf101f2..5bab3ccd3 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -14,7 +14,8 @@ // Package buildsignal holds the buildsignal-stage queue controller. It consumes // BuildSignal messages (a build id), polls the build-runner until the build -// reaches a terminal status, persists that status, and — once terminal — +// reaches a terminal status, persists that status, and — once terminal, or once +// the build outlives the queue's max_validation_ms, which fails the request — // releases the queue's build slot, projects the outcome onto the request, // advances the successful-build bookmark, and publishes directly to record plus // terminal analysis. See @@ -25,6 +26,7 @@ import ( "context" "errors" "fmt" + "time" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" @@ -37,6 +39,7 @@ import ( "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/buildrunner" + "github.com/uber/submitqueue/stovepipe/extension/queueconfig" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) @@ -65,6 +68,7 @@ type Controller struct { metricsScope tally.Scope stores storage.Factory materializer requestlog.Materializer + queueConfigs queueconfig.Store buildRunners buildrunner.Factory registry consumer.TopicRegistry topicKey consumer.TopicKey @@ -80,6 +84,7 @@ func NewController( scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, + queueConfigs queueconfig.Store, buildRunners buildrunner.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, @@ -90,6 +95,7 @@ func NewController( metricsScope: scope.SubScope("buildsignal_controller"), stores: stores, materializer: materializer, + queueConfigs: queueConfigs, buildRunners: buildRunners, registry: registry, topicKey: topicKey, @@ -172,11 +178,27 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return err } - if effective.IsTerminal() { - if err := c.persistBuildFinishedLog(ctx, store, request, build.ID); err != nil { + // A timed-out build established nothing: it keeps its polled status, is not the request's + // terminal build, and has no final analysis. The request's stored outcome carries the + // decision, and a retry re-derives the same one. + timedOut := false + if !effective.IsTerminal() { + if timedOut, err = c.validationDeadlinePassed(ctx, request, build); err != nil { return err } - if err := c.finishRequest(ctx, store, &request, build.ID, effective); err != nil { + } + + if effective.IsTerminal() || timedOut { + state, reason, terminalBuildID := entity.RequestStateFailed, entity.RequestOutcomeReasonValidationTimeout, "" + if !timedOut { + state = outcomeState(effective) + reason = buildOutcomeReason(state) + terminalBuildID = build.ID + if err := c.persistBuildFinishedLog(ctx, store, request, build.ID); err != nil { + return err + } + } + if err := c.finishRequest(ctx, store, &request, terminalBuildID, state, reason); err != nil { return err } if request.State == entity.RequestStateSucceeded { @@ -193,11 +215,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er recordErr = fmt.Errorf("failed to publish record for request %s: %w", request.ID, recordErr) } - analysisErr := c.publishBuildAnalysis(ctx, build.ID, request.Queue) + var analysisErr error + if !timedOut { + analysisErr = c.publishBuildAnalysis(ctx, build.ID, request.Queue) + } if err := errs.Group(recordErr, analysisErr); err != nil { return err } - c.logger.Infow("build reached terminal status", + c.logger.Infow("recorded request outcome", "build_id", build.ID, "request_id", request.ID, "status", string(effective), @@ -232,8 +257,8 @@ func (c *Controller) persistBuildFinishedLog(ctx context.Context, store storage. return nil } -// finishRequest releases the queue's build slot and projects the build's -// terminal status and ID onto the request, leaving it terminal. It is a no-op +// finishRequest releases the queue's build slot and records state, reason, and +// buildID on the request, leaving it terminal. It is a no-op // when the request already carries an outcome, so a redelivery neither // double-releases the slot nor rewrites the state. // @@ -245,7 +270,7 @@ func (c *Controller) persistBuildFinishedLog(ctx context.Context, store storage. // the request non-terminal, so redelivery re-runs both steps and decrements again // — transiently over-admitting by one until releaseBuildSlot's zero clamp // reconverges, which is the failure mode this pipeline prefers. -func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, request *entity.Request, buildID string, status entity.BuildStatus) error { +func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, request *entity.Request, buildID string, state entity.RequestState, reason entity.RequestOutcomeReason) error { if request.State.HasBuildOutcome() { return nil } @@ -255,14 +280,37 @@ func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, r return err } - state := outcomeState(status) - if err := c.markOutcome(ctx, store, request, buildID, state, buildOutcomeReason(state)); err != nil { + if err := c.markOutcome(ctx, store, request, buildID, state, reason); err != nil { metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) return err } return nil } +// validationDeadlinePassed reports whether build has outlived its queue's max_validation_ms. +// A zero limit or a zero creation time means the build has no deadline. +func (c *Controller) validationDeadlinePassed(ctx context.Context, request entity.Request, build entity.Build) (bool, error) { + if build.CreatedAt == 0 { + return false, nil + } + cfg, err := c.queueConfigs.Get(ctx, request.Queue) + if err != nil { + return false, fmt.Errorf("failed to load queue config for %s: %w", request.Queue, err) + } + if cfg.MaxValidationMs <= 0 || time.Now().UnixMilli()-build.CreatedAt < cfg.MaxValidationMs { + return false, nil + } + + metrics.NamedCounter(c.metricsScope, _opName, "validation_timeouts", 1, metrics.TagsFromContext(ctx)...) + c.logger.Warnw("failing build that exceeded the queue's max validation time", + "build_id", build.ID, + "request_id", request.ID, + "created_at", build.CreatedAt, + "max_validation_ms", cfg.MaxValidationMs, + ) + return true, nil +} + func (c *Controller) persistOutcomeLog(ctx context.Context, store storage.Storage, request entity.Request) error { if !request.State.HasBuildOutcome() { return fmt.Errorf("request %s has no build outcome to record", request.ID) diff --git a/stovepipe/controller/buildsignal/buildsignal_test.go b/stovepipe/controller/buildsignal/buildsignal_test.go index 597cab9ee..49c9e476f 100644 --- a/stovepipe/controller/buildsignal/buildsignal_test.go +++ b/stovepipe/controller/buildsignal/buildsignal_test.go @@ -39,6 +39,7 @@ import ( "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/buildrunner" buildrunnermock "github.com/uber/submitqueue/stovepipe/extension/buildrunner/mock" + queueconfigmock "github.com/uber/submitqueue/stovepipe/extension/queueconfig/mock" "github.com/uber/submitqueue/stovepipe/extension/storage" storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" "go.uber.org/mock/gomock" @@ -64,6 +65,7 @@ type buildsignalMocks struct { queueStore *storagemock.MockQueueStore store *storagemock.MockStorage materializer *requestlogmock.MockMaterializer + queueConfigs *queueconfigmock.MockStore runnerFactory *buildrunnermock.MockFactory runner *buildrunnermock.MockBuildRunner publisher *mqmock.MockPublisher @@ -86,6 +88,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildsig queueStore: storagemock.NewMockQueueStore(ctrl), store: storagemock.NewMockStorage(ctrl), materializer: requestlogmock.NewMockMaterializer(ctrl), + queueConfigs: queueconfigmock.NewMockStore(ctrl), runnerFactory: buildrunnermock.NewMockFactory(ctrl), runner: buildrunnermock.NewMockBuildRunner(ctrl), publisher: mqmock.NewMockPublisher(ctrl), @@ -106,7 +109,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildsig }) require.NoError(t, err) - c := NewController(zap.NewNop().Sugar(), scope, staticStorageFactory{store: m.store}, m.materializer, m.runnerFactory, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal") + c := NewController(zap.NewNop().Sugar(), scope, staticStorageFactory{store: m.store}, m.materializer, m.queueConfigs, m.runnerFactory, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal") return c, m } @@ -226,14 +229,15 @@ func expectBuildFinished(m buildsignalMocks) *gomock.Call { } func expectFinish(m buildsignalMocks, state entity.RequestState) *gomock.Call { - return expectOutcomeLog(m, state, 2).After(expectFinishWrites(m, state)) + return expectOutcomeLog(m, requestWithState(state), 2).After(expectFinishWrites(m, state)) } -func expectOutcomeLog(m buildsignalMocks, state entity.RequestState, version int32) *gomock.Call { - request := requestWithState(state) +func expectOutcomeLog(m buildsignalMocks, request entity.Request, version int32) *gomock.Call { request.Version = version log := requestlog.NewRequestStateLog(request, request.OutcomeReason) - log.Metadata[requestlog.MetadataKeyBuildID] = request.TerminalBuildID + if request.TerminalBuildID != "" { + log.Metadata[requestlog.MetadataKeyBuildID] = request.TerminalBuildID + } return m.materializer.EXPECT().PersistLog( gomock.Any(), m.store, @@ -346,6 +350,39 @@ func TestProcess(t *testing.T) { m.buildStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil) }, }, + { + name: "nonterminal poll holds when the queue disables the validation deadline", + wantHoldMs: PollDelayRunningMs, + setup: func(m buildsignalMocks) { + running := build(entity.BuildStatusRunning, 2) + running.CreatedAt = 1 + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(running, nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusRunning, nil, nil) + m.queueConfigs.EXPECT().Get(gomock.Any(), testQueue).Return(entity.QueueConfig{Name: testQueue}, nil) + }, + }, + { + name: "nonterminal poll past the validation deadline fails the request without holding, failing the build or analyzing it", + setup: func(m buildsignalMocks) { + running := build(entity.BuildStatusRunning, 2) + running.CreatedAt = 1 + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(running, nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusRunning, nil, nil) + m.queueConfigs.EXPECT().Get(gomock.Any(), testQueue).Return(entity.QueueConfig{Name: testQueue, MaxValidationMs: 1}, nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow(1, 4), nil) + m.queueStore.EXPECT().Update(gomock.Any(), queueRow(0, 4), int32(4), int32(5)).Return(nil) + outcome := requestWithState(entity.RequestStateFailed) + outcome.TerminalBuildID = "" + outcome.OutcomeReason = entity.RequestOutcomeReasonValidationTimeout + outcomeCall := m.reqStore.EXPECT().Update(gomock.Any(), outcome, int32(1), int32(2)).Return(nil) + logCall := expectOutcomeLog(m, outcome, 2).After(outcomeCall) + m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil).After(logCall) + }, + }, { name: "succeeded build frees the slot, marks the request and publishes to record", setup: func(m buildsignalMocks) { @@ -401,7 +438,7 @@ func TestProcess(t *testing.T) { queue.LastGreenRequestID = testID queue.LastGreenURI = request.URI getCall := m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queue, nil).After(eventCall) - logCall := expectOutcomeLog(m, entity.RequestStateSucceeded, 2).After(getCall) + logCall := expectOutcomeLog(m, requestWithState(entity.RequestStateSucceeded), 2).After(getCall) m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil).After(logCall) m.publisher.EXPECT().Publish(gomock.Any(), "buildanalysis", gomock.Any()).Return(nil).After(logCall) }, diff --git a/stovepipe/entity/build.go b/stovepipe/entity/build.go index e278bf904..807bfe1b7 100644 --- a/stovepipe/entity/build.go +++ b/stovepipe/entity/build.go @@ -67,6 +67,9 @@ type Build struct { RequestID string `json:"request_id"` // Status is the build's lifecycle state. Status BuildStatus `json:"status"` + // CreatedAt is the Unix time in milliseconds at which the build was created. + // It is zero on builds created before creation times were recorded. + CreatedAt int64 `json:"created_at"` // Version is used for optimistic locking. Versioning starts at 1 and // is incremented for each change to the object. Version int32 `json:"version"` diff --git a/stovepipe/entity/queue_config.go b/stovepipe/entity/queue_config.go index b6681cd83..a3a4360ff 100644 --- a/stovepipe/entity/queue_config.go +++ b/stovepipe/entity/queue_config.go @@ -26,4 +26,8 @@ type QueueConfig struct { MaxConcurrent int32 `json:"max_concurrent" yaml:"max_concurrent"` // GateWaitDelayMs is the redelivery delay while the latest head waits for a slot. GateWaitDelayMs int64 `json:"gate_wait_delay_ms" yaml:"gate_wait_delay_ms"` + // MaxValidationMs is the maximum time in milliseconds from build creation to a + // terminal build status, past which the validation fails by timeout. Zero + // disables the deadline. + MaxValidationMs int64 `json:"max_validation_ms" yaml:"max_validation_ms"` } diff --git a/stovepipe/extension/queueconfig/README.md b/stovepipe/extension/queueconfig/README.md index 40f88558d..d75bfba5d 100644 --- a/stovepipe/extension/queueconfig/README.md +++ b/stovepipe/extension/queueconfig/README.md @@ -10,7 +10,7 @@ Pipeline stages read mutable runtime state from storage and read knobs such as ` ## Entities -Queue configuration entity lives in `stovepipe/entity/queue_config.go` and carries deployment knobs (`max_concurrent`, `gate_wait_delay_ms`) separate from the mutable `Queue` row. +Queue configuration entity lives in `stovepipe/entity/queue_config.go` and carries deployment knobs (`max_concurrent`, `gate_wait_delay_ms`, `max_validation_ms`) separate from the mutable `Queue` row. ## Implementations diff --git a/stovepipe/extension/queueconfig/default/default.go b/stovepipe/extension/queueconfig/default/default.go index 5fb696710..2db12d22f 100644 --- a/stovepipe/extension/queueconfig/default/default.go +++ b/stovepipe/extension/queueconfig/default/default.go @@ -19,6 +19,7 @@ package defaultconfig import ( "context" + "time" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/queueconfig" @@ -27,6 +28,8 @@ import ( const ( _defaultMaxConcurrent = 1 _defaultGateWaitDelayMs = 5000 + // Bounds a hung build without cutting a long full-repository validation short. + _defaultMaxValidationMs = int64(2 * time.Hour / time.Millisecond) ) // Store is a queueconfig.Store that returns the same defaults for every queue. @@ -46,6 +49,7 @@ func (Store) Get(_ context.Context, name string) (entity.QueueConfig, error) { Name: name, MaxConcurrent: _defaultMaxConcurrent, GateWaitDelayMs: _defaultGateWaitDelayMs, + MaxValidationMs: _defaultMaxValidationMs, }, nil } diff --git a/stovepipe/extension/queueconfig/default/default_test.go b/stovepipe/extension/queueconfig/default/default_test.go index 490ff35a7..8e649810c 100644 --- a/stovepipe/extension/queueconfig/default/default_test.go +++ b/stovepipe/extension/queueconfig/default/default_test.go @@ -32,6 +32,7 @@ func TestStore_Get(t *testing.T) { assert.Equal(t, "monorepo/main", cfg.Name) assert.Equal(t, int32(1), cfg.MaxConcurrent) assert.Equal(t, int64(5000), cfg.GateWaitDelayMs) + assert.Positive(t, cfg.MaxValidationMs) }) t.Run("empty name is not found", func(t *testing.T) { diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index d79b2ec33..d97efc0bf 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -30,6 +30,10 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte Existing MySQL deployments must add `outcome_reason VARCHAR(64) NOT NULL DEFAULT ''` to `request` before deploying the new readers and writers. +## Builds + +Existing MySQL deployments must add `created_at BIGINT NOT NULL DEFAULT 0` to `build` before deploying the new readers and writers. + ## Build issue observations `BuildIssueStore` accumulates observations identified by build, task, and unit within the bound queue. The build ID is supplied separately from the detected issue. Repeated observations return `ErrAlreadyExists`, which callers may treat as successful persistence. Build and task IDs must be nonempty; empty unit IDs represent task-wide observations. diff --git a/stovepipe/extension/storage/mysql/build_store.go b/stovepipe/extension/storage/mysql/build_store.go index ab079a04a..6461f8379 100644 --- a/stovepipe/extension/storage/mysql/build_store.go +++ b/stovepipe/extension/storage/mysql/build_store.go @@ -46,12 +46,13 @@ func (b *buildStore) Create(ctx context.Context, build entity.Build) (retErr err defer func() { op.Complete(retErr) }() _, err := b.db.ExecContext(ctx, - `INSERT INTO build (queue, id, request_id, status, version) - VALUES (?, ?, ?, ?, ?)`, + `INSERT INTO build (queue, id, request_id, status, created_at, version) + VALUES (?, ?, ?, ?, ?, ?)`, b.queue, build.ID, build.RequestID, build.Status, + build.CreatedAt, build.Version, ) if err != nil { @@ -71,13 +72,14 @@ func (b *buildStore) Get(ctx context.Context, id string) (ret entity.Build, retE var build entity.Build err := b.db.QueryRowContext(ctx, - `SELECT id, request_id, status, version + `SELECT id, request_id, status, created_at, version FROM build WHERE queue = ? AND id = ?`, b.queue, id, ).Scan( &build.ID, &build.RequestID, &build.Status, + &build.CreatedAt, &build.Version, ) diff --git a/stovepipe/extension/storage/mysql/build_store_test.go b/stovepipe/extension/storage/mysql/build_store_test.go index c2af2c000..220f14a8c 100644 --- a/stovepipe/extension/storage/mysql/build_store_test.go +++ b/stovepipe/extension/storage/mysql/build_store_test.go @@ -45,6 +45,7 @@ func TestBuildStore_Create(t *testing.T) { ID: "bk-1001", RequestID: "1", Status: entity.BuildStatusAccepted, + CreatedAt: 1700000000000, Version: 1, } @@ -58,7 +59,7 @@ func TestBuildStore_Create(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO build"). - WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version). + WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.CreatedAt, build.Version). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -66,7 +67,7 @@ func TestBuildStore_Create(t *testing.T) { name: "duplicate id returns ErrAlreadyExists", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO build"). - WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version). + WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.CreatedAt, build.Version). WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) }, wantErr: true, @@ -76,7 +77,7 @@ func TestBuildStore_Create(t *testing.T) { name: "other exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO build"). - WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.Version). + WithArgs("monorepo/main", build.ID, build.RequestID, build.Status, build.CreatedAt, build.Version). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -109,6 +110,7 @@ func TestBuildStore_Get(t *testing.T) { ID: "bk-1001", RequestID: "1", Status: entity.BuildStatusRunning, + CreatedAt: 1700000000000, Version: 2, } @@ -124,9 +126,9 @@ func TestBuildStore_Get(t *testing.T) { name: "found", id: want.ID, setup: func(mock sqlmock.Sqlmock) { - rows := sqlmock.NewRows([]string{"id", "request_id", "status", "version"}). - AddRow(want.ID, want.RequestID, string(want.Status), want.Version) - mock.ExpectQuery("SELECT id, request_id, status, version"). + rows := sqlmock.NewRows([]string{"id", "request_id", "status", "created_at", "version"}). + AddRow(want.ID, want.RequestID, string(want.Status), want.CreatedAt, want.Version) + mock.ExpectQuery("SELECT id, request_id, status, created_at, version"). WithArgs("monorepo/main", want.ID). WillReturnRows(rows) }, @@ -136,7 +138,7 @@ func TestBuildStore_Get(t *testing.T) { name: "not found", id: "missing", setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT id, request_id, status, version"). + mock.ExpectQuery("SELECT id, request_id, status, created_at, version"). WithArgs("monorepo/main", "missing"). WillReturnError(sql.ErrNoRows) }, @@ -147,7 +149,7 @@ func TestBuildStore_Get(t *testing.T) { name: "query error", id: "bad", setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT id, request_id, status, version"). + mock.ExpectQuery("SELECT id, request_id, status, created_at, version"). WithArgs("monorepo/main", "bad"). WillReturnError(fmt.Errorf("connection reset")) }, diff --git a/stovepipe/extension/storage/mysql/schema/build.sql b/stovepipe/extension/storage/mysql/schema/build.sql index 4c60f17c1..b070681b3 100644 --- a/stovepipe/extension/storage/mysql/schema/build.sql +++ b/stovepipe/extension/storage/mysql/schema/build.sql @@ -5,6 +5,7 @@ CREATE TABLE IF NOT EXISTS build ( id VARCHAR(255) NOT NULL, request_id VARCHAR(255) NOT NULL, status VARCHAR(64) NOT NULL, + created_at BIGINT NOT NULL DEFAULT 0, version INT NOT NULL, PRIMARY KEY (queue, id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/test/integration/stovepipe/extension/storage/suite.go b/test/integration/stovepipe/extension/storage/suite.go index fec3a3d8b..606bbfa6c 100644 --- a/test/integration/stovepipe/extension/storage/suite.go +++ b/test/integration/stovepipe/extension/storage/suite.go @@ -347,6 +347,7 @@ func (s *BuildStoreContractSuite) TestBuildStore_CreateAndGet() { ID: id, RequestID: "1", Status: entity.BuildStatusAccepted, + CreatedAt: 1700000000000, Version: 1, } require.NoError(t, s.buildStore.Create(s.ctx, build))