diff --git a/doc/rfc/stovepipe/steps/build.md b/doc/rfc/stovepipe/steps/build.md index df673235..13040efd 100644 --- a/doc/rfc/stovepipe/steps/build.md +++ b/doc/rfc/stovepipe/steps/build.md @@ -22,7 +22,7 @@ For a delivery carrying request id `R`: ``` 1. Load Request R from the request store. - - ErrNotFound -> retryable (process/analyze write not visible yet; redelivery converges). + - ErrNotFound -> return raw; non-retryable (storage is read-after-write consistent; see [storage README](stovepipe/extension/storage/README.md)). - other store error -> return raw; classifier decides. 2. If R.State is terminal (superseded / recorded-green / recorded-not-green): ack and return. @@ -75,7 +75,8 @@ For a delivery carrying request id `R`: Every branch is safe under at-least-once redelivery — with SubmitQueue's posture on duplicates adopted wholesale: `build` has no pre-trigger dedup check (there is no caller-derivable key to check by; see [Alternatives considered](#alternatives-considered-for-the-build-identity)), so a redelivery that reaches step 5 starts a second, independent build, and safety comes from downstream idempotency rather than from preventing the duplicate: -- **Request not found / strategy not yet visible** — retryable; the producing stage's write is not visible on this reader yet. +- **Request not found** — non-retryable; storage's read-after-write guarantee means a miss here is a storage defect, not a lag condition to retry through. +- **Strategy not yet visible** — retryable; the producing stage's write is not visible on this reader yet. - **Request already terminal** (step 2) — ack, no build. A redelivery after `record` finished, or after `process` superseded the head, never starts a stale build. - **Redelivery while the Request is still in flight** (crash or failure anywhere in steps 5–8) — the redelivery re-runs from step 1, `Trigger` mints a fresh id, `Create` persists a second `Build` row, and a second poll loop starts. Harmless, in three layers: both builds target the identical `(headURI, baseURI)` scope; each `Build` polls in its own partition and `buildsignal` short-circuits the moment the Request goes terminal (its step 3); and `record`'s terminal transition is CAS-guarded, so the second verdict is a no-op. A build triggered but never persisted (crash between steps 5 and 6) is the same story minus the row: an orphan the runner finishes and nobody ever reads. Wasted CI compute, not a correctness risk — the same accepted trade as SubmitQueue. - **Trigger / publish / other store failure** — nothing durable is left half-written that a redelivery can't reconcile; the error rejects to DLQ, and the fail-closed reconciler drives the Request terminal (see [workflow.md](doc/rfc/stovepipe/workflow.md#fail-closed-on-unprocessable-work)). @@ -101,9 +102,11 @@ Per `platform/errs`'s non-retryable-by-default rule (see [platform/errs/README.m | Failure | Disposition | Why | |---|---|---| -| `Request` not found (`storage.ErrNotFound`) | retryable (`errs.NewRetryableError`) | On a primary-only read this shouldn't happen in practice — `process` commits before it publishes — but the check costs nothing when the row is already visible and gives free convergence on the rare chance it isn't, same posture as `process.go`. | +| `BuildStrategy` not yet visible (step 4) | retryable (`errs.NewRetryableError`) | The producing stage's write (`process`'s CAS) may not be visible on this reader yet; redelivery converges. | | `Trigger` | raw error; classifier decides | Deliberately left open rather than fixed either way — a runner timeout/connection is transient, a bad URI is permanent, and only a backend classifier can tell them apart. | +`Request` not found (`storage.ErrNotFound`) is **not** in this table: storage is required to be read-after-write consistent (see [storage README](stovepipe/extension/storage/README.md)), so a miss here is already the correct default (non-retryable, straight to DLQ) rather than a departure worth overriding. + Everything else — factory lookup, a malformed message, a build-store error other than `ErrAlreadyExists`, and the publish to `buildsignal` — is returned raw with no override, because the default is already correct: none of them are worth automatically replaying (a queue with no registered builder is a config error, a broken payload will never parse, and storage/queue and publish failures dead-letter and let DLQ reconciliation recover). ## Orchestrator vs. Stovepipe build diff --git a/doc/rfc/stovepipe/steps/process.md b/doc/rfc/stovepipe/steps/process.md index e45b97c0..76508f53 100644 --- a/doc/rfc/stovepipe/steps/process.md +++ b/doc/rfc/stovepipe/steps/process.md @@ -17,7 +17,7 @@ For a delivery carrying request id `R`: ``` 1. Load Request R from the request store. - - not found yet -> retryable error (ingest write not visible; redelivery converges) + - not found -> non-retryable (storage is read-after-write consistent; see [storage README](../../../../stovepipe/extension/storage/README.md)). 2. If R.State is terminal (superseded / recorded-green / recorded-not-green): - ack and return (idempotent no-op). 3. If R.State is processing (strategy already recorded): @@ -163,7 +163,7 @@ On a crash between admit and `record`, the Request stays non-terminal; visibilit - **Re-ingest of a superseded URI.** Ingest dedups on `(Queue, URI)` and returns the existing (now terminal `superseded`) id; `process` acks it as a no-op (step 2). Correct: a URI is only superseded for a *strictly newer* head, so re-validating it is never wanted. - **Gate closed, no newer head.** The single latest head waits for a slot until the in-flight validation completes — the steady state, not an error. - **Head equals last-green.** `IsAncestor(lastGreen, R.URI)` with `R.URI == lastGreen` is degenerate; treat as already-green, or (simpler) run an incremental build with an empty delta. Left to `build`. -- **Queue row missing.** First head for a Queue: ingest get-or-creates the row with defaults (`in_flight_count = 0`, empty `last_green_uri`). `process` treats a missing row as retryable (ingest write not yet visible). +- **Queue row missing.** First head for a Queue: ingest get-or-creates the row with defaults (`in_flight_count = 0`, empty `last_green_uri`). `process` treats a missing row as non-retryable — storage's read-after-write guarantee means ingest's write is already visible by the time `process` reads it, so a miss is a storage defect, not lag. ## Entity model diff --git a/stovepipe/controller/build/BUILD.bazel b/stovepipe/controller/build/BUILD.bazel new file mode 100644 index 00000000..be0c4736 --- /dev/null +++ b/stovepipe/controller/build/BUILD.bazel @@ -0,0 +1,43 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["build.go"], + importpath = "github.com/uber/submitqueue/stovepipe/controller/build", + visibility = ["//visibility:public"], + deps = [ + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/errs:go_default_library", + "//platform/metrics:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/buildrunner:go_default_library", + "//stovepipe/extension/storage:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["build_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/errs:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", + "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/buildrunner:go_default_library", + "//stovepipe/extension/buildrunner/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", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) diff --git a/stovepipe/controller/build/build.go b/stovepipe/controller/build/build.go new file mode 100644 index 00000000..885bf9a4 --- /dev/null +++ b/stovepipe/controller/build/build.go @@ -0,0 +1,193 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package build holds the build-stage queue controller. It consumes BuildRequest +// messages (a request id), reloads the Request, triggers the build-runner for +// the scope process already decided, persists a Build row, and publishes the +// build id to buildsignal. +package build + +import ( + "context" + "errors" + "fmt" + + "github.com/uber-go/tally" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/platform/metrics" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/buildrunner" + "github.com/uber/submitqueue/stovepipe/extension/storage" + "go.uber.org/zap" +) + +// Controller consumes BuildRequest messages, reloads the referenced Request, +// triggers a build for its already-decided scope, and publishes the resulting +// build id to buildsignal. Implements consumer.Controller. +type Controller struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + store storage.Storage + buildRunners buildrunner.Factory + registry consumer.TopicRegistry + topicKey consumer.TopicKey + consumerGroup string +} + +// Verify Controller implements consumer.Controller interface at compile time. +var _ consumer.Controller = (*Controller)(nil) + +// _opName is the metric operation name shared by every emit in this file. +const _opName = "build" + +// NewController creates a new build controller. +func NewController( + logger *zap.SugaredLogger, + scope tally.Scope, + store storage.Storage, + buildRunners buildrunner.Factory, + registry consumer.TopicRegistry, + topicKey consumer.TopicKey, + consumerGroup string, +) *Controller { + return &Controller{ + logger: logger.Named("build_controller"), + metricsScope: scope.SubScope("build_controller"), + store: store, + buildRunners: buildRunners, + registry: registry, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process reloads the request referenced by the delivery, triggers a build for +// its decided scope, and publishes the build id to buildsignal. Returns nil to +// ack (success) or an error to nack (retry) / reject (DLQ). +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { + op := metrics.Begin(c.metricsScope, _opName) + defer func() { op.Complete(retErr) }() + + msg := delivery.Message() + + br := &stovepipemq.BuildRequest{} + if err := stovepipemq.Unmarshal(msg.Payload, br); err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "deserialize_errors", 1) + // Non-retryable: a malformed message will never succeed regardless of retries. + return fmt.Errorf("failed to deserialize build request: %w", err) + } + + request, err := c.loadRequest(ctx, br.Id) + if err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1) + return err + } + + // A redelivery after record already finished, or after process superseded + // the head, must not start a fresh build. + if request.State.IsTerminal() { + return nil + } + + buildRunner, err := c.buildRunners.For(buildrunner.Config{QueueName: request.Queue}) + if err != nil { + // A queue with no registered builder is a config error. + return fmt.Errorf("BuildController failed to resolve build runner for queue %s: %w", request.Queue, err) + } + + // process decided the scope; build never re-derives incremental-vs-full. + if request.BuildStrategy == entity.BuildStrategyUnknown { + metrics.NamedCounter(c.metricsScope, _opName, "strategy_not_visible", 1) + return errs.NewRetryableError(fmt.Errorf("request %s has no build strategy yet", request.ID)) + } + baseURI := "" + if request.BuildStrategy == entity.BuildStrategyIncrementalSinceGreen { + baseURI = request.BaseURI + } + + buildID, err := buildRunner.Trigger(ctx, baseURI, request.URI, nil) + if err != nil { + return fmt.Errorf("BuildController failed to trigger build for request %s: %w", request.ID, err) + } + + build := entity.Build{ + ID: buildID.ID, + RequestID: request.ID, + Status: entity.BuildStatusAccepted, + Version: 1, + } + if err := c.store.GetBuildStore().Create(ctx, build); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("BuildController failed to persist build %s: %w", build.ID, err) + } + + if err := c.publishBuildSignal(ctx, build.ID); err != nil { + return fmt.Errorf("BuildController failed to publish build signal for %s: %w", build.ID, err) + } + + c.logger.Debugw("triggered build", + "request_id", request.ID, + "build_id", build.ID, + "queue", request.Queue, + "base_uri", baseURI, + ) + return nil +} + +// loadRequest returns the request for id. +func (c *Controller) loadRequest(ctx context.Context, id string) (entity.Request, error) { + got, err := c.store.GetRequestStore().Get(ctx, id) + if err != nil { + return entity.Request{}, fmt.Errorf("BuildController failed to load request %s: %w", id, err) + } + return got, nil +} + +// publishBuildSignal publishes buildID to the buildsignal stage, partitioned by +// build id so each build's poll loop runs in its own partition. +func (c *Controller) publishBuildSignal(ctx context.Context, buildID string) error { + payload, err := stovepipemq.Marshal(&stovepipemq.BuildSignal{Id: buildID}) + if err != nil { + return fmt.Errorf("failed to serialize build signal: %w", err) + } + + msg := entityqueue.NewMessage(buildID, payload, buildID, nil) + + q, ok := c.registry.Queue(stovepipemq.TopicKeyBuildSignal) + if !ok { + return fmt.Errorf("no queue registered for topic key %s", stovepipemq.TopicKeyBuildSignal) + } + topicName, ok := c.registry.TopicName(stovepipemq.TopicKeyBuildSignal) + if !ok { + return fmt.Errorf("no topic name registered for topic key %s", stovepipemq.TopicKeyBuildSignal) + } + return q.Publisher().Publish(ctx, topicName, msg) +} + +// Name returns the controller name for logging and metrics. +func (c *Controller) Name() string { + return "build" +} + +// TopicKey returns the topic key this controller subscribes to. +func (c *Controller) TopicKey() consumer.TopicKey { + return c.topicKey +} + +// ConsumerGroup returns the consumer group for offset tracking. +func (c *Controller) ConsumerGroup() string { + return c.consumerGroup +} diff --git a/stovepipe/controller/build/build_test.go b/stovepipe/controller/build/build_test.go new file mode 100644 index 00000000..da176089 --- /dev/null +++ b/stovepipe/controller/build/build_test.go @@ -0,0 +1,293 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package build + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" + mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/buildrunner" + buildrunnermock "github.com/uber/submitqueue/stovepipe/extension/buildrunner/mock" + "github.com/uber/submitqueue/stovepipe/extension/storage" + storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +const ( + testQueue = "monorepo/main" + testID = "request/monorepo/main/7" + testHeadURI = "git://repo/monorepo/main/head" + testBaseURI = "git://repo/monorepo/main/base" + testBuildID = "bk-1" +) + +// buildMocks bundles the mocks a build controller test case wires expectations on. +type buildMocks struct { + reqStore *storagemock.MockRequestStore + buildStore *storagemock.MockBuildStore + runnerFactory *buildrunnermock.MockFactory + runner *buildrunnermock.MockBuildRunner + publisher *mqmock.MockPublisher +} + +func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMocks) { + t.Helper() + + m := buildMocks{ + reqStore: storagemock.NewMockRequestStore(ctrl), + buildStore: storagemock.NewMockBuildStore(ctrl), + runnerFactory: buildrunnermock.NewMockFactory(ctrl), + runner: buildrunnermock.NewMockBuildRunner(ctrl), + publisher: mqmock.NewMockPublisher(ctrl), + } + + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() + store.EXPECT().GetBuildStore().Return(m.buildStore).AnyTimes() + + queue := mqmock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(m.publisher).AnyTimes() + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: stovepipemq.TopicKeyBuildSignal, Name: "buildsignal", Queue: queue}, + }) + require.NoError(t, err) + + c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, m.runnerFactory, registry, stovepipemq.TopicKeyBuild, "stovepipe-build") + return c, m +} + +func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.Delivery { + t.Helper() + d := mqmock.NewMockDelivery(ctrl) + d.EXPECT().Message().Return(entityqueue.NewMessage(testID, payload, testID, nil)).AnyTimes() + d.EXPECT().Attempt().Return(1).AnyTimes() + return d +} + +func buildPayload(t *testing.T, id string) []byte { + t.Helper() + b, err := stovepipemq.Marshal(&stovepipemq.BuildRequest{Id: id}) + require.NoError(t, err) + return b +} + +// processingRequest returns a Request past process's admit, with the given +// already-decided scope. +func processingRequest(strategy entity.BuildStrategy, baseURI string) entity.Request { + return entity.Request{ + ID: testID, + Queue: testQueue, + URI: testHeadURI, + BuildStrategy: strategy, + BaseURI: baseURI, + State: entity.RequestStateProcessing, + Version: 1, + } +} + +func TestProcess(t *testing.T) { + tests := []struct { + name string + payload []byte + setup func(m buildMocks) + wantErr bool + wantRetry bool + }{ + { + name: "incremental build triggers and publishes", + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyIncrementalSinceGreen, testBaseURI) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Trigger(gomock.Any(), testBaseURI, testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil) + build := entity.Build{ + ID: testBuildID, + RequestID: testID, + Status: entity.BuildStatusAccepted, + Version: 1, + } + m.buildStore.EXPECT().Create(gomock.Any(), build).Return(nil) + m.publisher.EXPECT().Publish(gomock.Any(), "buildsignal", gomock.Any()).Return(nil) + }, + }, + { + name: "full build ignores stale base uri", + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyFull, testBaseURI) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil) + build := entity.Build{ + ID: testBuildID, + RequestID: testID, + Status: entity.BuildStatusAccepted, + Version: 1, + } + m.buildStore.EXPECT().Create(gomock.Any(), build).Return(nil) + m.publisher.EXPECT().Publish(gomock.Any(), "buildsignal", gomock.Any()).Return(nil) + }, + }, + { + name: "superseded is a no-op", + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyFull, "") + req.State = entity.RequestStateSuperseded + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + }, + }, + { + name: "recorded green is a no-op", + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyFull, "") + req.State = entity.RequestStateRecordedGreen + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + }, + }, + { + name: "recorded not green is a no-op", + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyFull, "") + req.State = entity.RequestStateRecordedNotGreen + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + }, + }, + { + name: "build strategy not yet visible is retryable", + wantErr: true, + wantRetry: true, + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyUnknown, "") + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + }, + }, + { + name: "request not found is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, storage.ErrNotFound) + }, + }, + { + name: "request storage error is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, errors.New("db down")) + }, + }, + { + name: "factory lookup failure is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyFull, "") + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(nil, errors.New("no runner")) + }, + }, + { + name: "trigger failure is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyFull, "") + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{}, errors.New("runner down")) + }, + }, + { + name: "already exists on create is swallowed and publish still happens", + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyFull, "") + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil) + m.buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists) + m.publisher.EXPECT().Publish(gomock.Any(), "buildsignal", gomock.Any()).Return(nil) + }, + }, + { + name: "build store error is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyFull, "") + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil) + m.buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(errors.New("db down")) + }, + }, + { + name: "publish failure is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildMocks) { + req := processingRequest(entity.BuildStrategyFull, "") + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(req, nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Trigger(gomock.Any(), "", testHeadURI, entity.BuildMetadata(nil)).Return(entity.BuildID{ID: testBuildID}, nil) + m.buildStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil) + m.publisher.EXPECT().Publish(gomock.Any(), "buildsignal", gomock.Any()).Return(errors.New("queue down")) + }, + }, + { + name: "malformed payload is not retryable", + payload: []byte("not-json"), + wantErr: true, + wantRetry: false, + setup: func(m buildMocks) {}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + if tt.setup != nil { + tt.setup(m) + } + + payload := tt.payload + if payload == nil { + payload = buildPayload(t, testID) + } + + err := c.Process(context.Background(), delivery(t, ctrl, payload)) + + if tt.wantErr { + require.Error(t, err) + assert.Equal(t, tt.wantRetry, errs.IsRetryable(err)) + return + } + require.NoError(t, err) + }) + } +} diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 21b82e83..9cadb884 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -453,28 +453,22 @@ func (c *Controller) rescheduleProcess(ctx context.Context, request entity.Reque return nil } -// loadRequest returns the request for id. A not-yet-visible row is retryable. +// loadRequest returns the request for id. func (c *Controller) loadRequest(ctx context.Context, id string) (entity.Request, error) { got, err := c.store.GetRequestStore().Get(ctx, id) - if err == nil { - return got, nil - } - if errors.Is(err, storage.ErrNotFound) { - return entity.Request{}, errs.NewRetryableError(fmt.Errorf("request %s not found yet: %w", id, err)) + if err != nil { + return entity.Request{}, fmt.Errorf("ProcessController failed to load request %s: %w", id, err) } - return entity.Request{}, fmt.Errorf("ProcessController failed to load request %s: %w", id, err) + return got, nil } -// loadQueue returns the queue row for name. A not-yet-visible row is retryable. +// loadQueue returns the queue row for name. func (c *Controller) loadQueue(ctx context.Context, name string) (entity.Queue, error) { got, err := c.store.GetQueueStore().Get(ctx, name) - if err == nil { - return got, nil - } - if errors.Is(err, storage.ErrNotFound) { - return entity.Queue{}, errs.NewRetryableError(fmt.Errorf("queue %s not found yet: %w", name, err)) + if err != nil { + return entity.Queue{}, fmt.Errorf("ProcessController failed to load queue %s: %w", name, err) } - return entity.Queue{}, fmt.Errorf("ProcessController failed to load queue %s: %w", name, err) + return got, nil } // publishBuild publishes the admitted request ID to the build stage. The build diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 80fc1cf8..35a0bbf8 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -743,17 +743,17 @@ func TestProcess(t *testing.T) { }, }, { - name: "request not found is retryable", + name: "request not found is not retryable", wantErr: true, - wantRetry: true, + wantRetry: false, setup: func(m processMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, storage.ErrNotFound) }, }, { - name: "queue not found is retryable", + name: "queue not found is not retryable", wantErr: true, - wantRetry: true, + wantRetry: false, setup: func(m processMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{}, storage.ErrNotFound) diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md new file mode 100644 index 00000000..992ba907 --- /dev/null +++ b/stovepipe/extension/storage/README.md @@ -0,0 +1,17 @@ +# Storage + +Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`). Implementations live under `extension/storage//`. This is a separate contract from `submitqueue/extension/storage` — same shape and conventions by design, but its own interfaces and its own `ErrNotFound`/`ErrAlreadyExists`/`ErrVersionMismatch` sentinels, since Stovepipe and SubmitQueue are independent domains. + +## Optimistic locking contract + +Entities that support concurrent mutation (`Request`, `Build`) carry an `int32 Version` field. `Update` methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write) — the store performs a pure conditional write and never computes `oldVersion + 1` itself. Version arithmetic is owned by the controller: it computes `newVersion`, calls `Update`, and only assigns `entity.Version = newVersion` after the call succeeds. See [CLAUDE.md](../../../CLAUDE.md) and the [submitqueue storage README](../../../submitqueue/extension/storage/README.md#optimistic-locking-contract) for the full rationale and the caller pattern — the convention is identical here. + +## Read-after-write consistency + +A `Get` immediately following a successful write (`Create`/`Update`) — by the same caller, or a causally-dependent one such as a queue consumer processing a message published after the write committed — must return that write. This is a requirement on every storage implementation, not a condition callers negotiate around. + +**Controllers must not treat `ErrNotFound` as "not visible yet, retry."** The store interface is intentionally general enough to run over any backend, so a controller has no way to know whether a missing row will appear shortly or does not exist at all — retrying on that assumption just reintroduces, in business logic, the consistency gap the storage contract exists to close. If a `Get` misses a row that a causally-prior write should already have produced (e.g. `build` loading the `Request` that `process` published its message for, or `process` loading the `Queue` row `ingest` get-or-created before publishing), that is a storage implementation defect: let the error surface as a normal (non-retryable, per [`platform/errs`](../../../platform/errs/README.md)'s default) failure rather than absorbing it with a retryable wrapper. See [build.md](../../../doc/rfc/stovepipe/steps/build.md#error-classification) and [process.md](../../../doc/rfc/stovepipe/steps/process.md) for the pipeline stages this applies to. + +## Key-value contract + +Same design space as [`submitqueue/extension/storage`](../../../submitqueue/extension/storage/README.md#key-value-contract): every method must be satisfiable by a plain key-value backend as cheaply as by MySQL — get/put/conditional-update by primary key only, no query-by-attribute or server-side filtering. `RequestURIStore` is the reverse-lookup example here (which request owns a given commit URI), kept as its own store rather than a secondary index on `RequestStore`. diff --git a/submitqueue/extension/storage/README.md b/submitqueue/extension/storage/README.md index c82e875b..51e17e71 100644 --- a/submitqueue/extension/storage/README.md +++ b/submitqueue/extension/storage/README.md @@ -26,6 +26,12 @@ entity.Version = newVersion // only after the write succeeded The post-success assignment matters whenever the entity is read again later in the same flow. Pre-incrementing in memory before the call is a bug pattern: if the call fails and the caller swallows the error, the in-memory version is now ahead of the database and subsequent updates will fail with `ErrVersionMismatch` for non-obvious reasons. +## Read-after-write consistency + +A `Get` immediately following a successful write (`Create`/`Update`) — by the same caller, or a causally-dependent one such as a queue consumer processing a message published after the write committed — must return that write. This is a requirement on every storage implementation, not a condition callers negotiate around: a MySQL primary (including after a promotion) satisfies it, and any other backend (KV, document, etc.) must too. + +**Controllers must not treat `ErrNotFound` as "not visible yet, retry."** The store interface is intentionally general enough to run over any backend, so a controller has no way to know whether a missing row will appear shortly or does not exist at all — retrying on that assumption just reintroduces, in business logic, the consistency gap the storage contract exists to close. If a `Get` misses a row that a causally-prior write should already have produced, that is a storage implementation defect: let the error surface as a normal (non-retryable, per `platform/errs`'s default) failure rather than absorbing it with a retryable wrapper. + ## Key-value contract Store interfaces are designed for the storage technology *space*, not for SQL (see the Extensions section of the repo `CLAUDE.md`): every method must be satisfiable by a plain key-value backend (DynamoDB, Bigtable, an in-memory map) as cheaply as by MySQL. Concretely, a store exposes only get/put/conditional-update **by primary key**. No lookups by other attributes, no listings filtered server-side, no joins.