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/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index fa4612ff..49158b2c 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -15,12 +15,16 @@ go_library( "//platform/extension/messagequeue/mysql:go_default_library", "//service/stovepipe/server/mapper:go_default_library", "//stovepipe/controller:go_default_library", + "//stovepipe/controller/build:go_default_library", "//stovepipe/controller/dlq:go_default_library", "//stovepipe/controller/process:go_default_library", "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/extension/buildrunner:go_default_library", + "//stovepipe/extension/buildrunner/fake:go_default_library", "//stovepipe/extension/queueconfig/default:go_default_library", "//stovepipe/extension/sourcecontrol:go_default_library", "//stovepipe/extension/sourcecontrol/fake:go_default_library", + "//stovepipe/extension/storage:go_default_library", "//stovepipe/extension/storage/mysql:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_uber_go_tally//:go_default_library", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 7b3aae31..8e40c51f 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -37,12 +37,16 @@ import ( queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" "github.com/uber/submitqueue/service/stovepipe/server/mapper" "github.com/uber/submitqueue/stovepipe/controller" + "github.com/uber/submitqueue/stovepipe/controller/build" "github.com/uber/submitqueue/stovepipe/controller/dlq" "github.com/uber/submitqueue/stovepipe/controller/process" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/extension/buildrunner" + buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake" queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default" "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" sourcecontrolfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake" + "github.com/uber/submitqueue/stovepipe/extension/storage" storageMySQL "github.com/uber/submitqueue/stovepipe/extension/storage/mysql" "go.uber.org/zap" "google.golang.org/grpc" @@ -100,6 +104,15 @@ func (fakeSourceControlFactory) For(cfg sourcecontrol.Config) (sourcecontrol.Sou return sourcecontrolfake.New([]string{fmt.Sprintf("git://%s/HEAD", cfg.QueueName)}), nil } +// fakeBuildRunnerFactory is the example BuildRunner factory: every queue shares the same +// stateless fake runner, which succeeds unless a caller embeds a failure marker in the head +// URI. A real deployment supplies a backend-specific factory (e.g. Buildkite, per queue). +type fakeBuildRunnerFactory struct{} + +func (fakeBuildRunnerFactory) For(_ buildrunner.Config) (buildrunner.BuildRunner, error) { + return buildrunnerfake.New(), nil +} + func main() { code := 0 if err := run(); err != nil { @@ -225,24 +238,21 @@ func run() error { errs.AlwaysRetryableProcessor, ) - processController := process.NewController( - logger.Sugar(), - scope, - store, - queueconfigdefault.NewStore(), - fakeSourceControlFactory{}, - registry, - stovepipemq.TopicKeyProcess, - "stovepipe-process", - ) - if err := primaryConsumer.Register(processController); err != nil { - return fmt.Errorf("failed to register process controller: %w", err) - } + // Each factory is constructed once and threaded through every consumer of + // it, so a real (stateful) backend introduced later is shared rather than + // silently duplicated across controllers. + scf := fakeSourceControlFactory{} + brf := fakeBuildRunnerFactory{} - processDLQController := dlq.NewController(logger.Sugar(), scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq") - if err := dlqConsumer.Register(processDLQController); err != nil { - return fmt.Errorf("failed to register process dlq controller: %w", err) + primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, store, registry, scf, brf) + if err != nil { + return err } + dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, store, registry) + if err != nil { + return err + } + logger.Info("controllers registered", zap.Int("primary", primaryCount), zap.Int("dlq", dlqCount)) // Start consumers. DLQ first because Start begins processing messages // immediately; if the primary consumer then fails to start, the half we @@ -266,7 +276,7 @@ func run() error { logger.Sugar(), scope, newInMemoryCounter(), - fakeSourceControlFactory{}, + scf, store, registry, ) @@ -338,10 +348,67 @@ func run() error { return err } +// registerPrimaryControllers creates the primary-pipeline queue controllers and +// registers them with c, returning how many were registered. +func registerPrimaryControllers( + c consumer.Consumer, + logger *zap.SugaredLogger, + scope tally.Scope, + store storage.Storage, + registry consumer.TopicRegistry, + scf sourcecontrol.Factory, + brf buildrunner.Factory, +) (int, error) { + var count int + + processController := process.NewController( + logger, + scope, + store, + queueconfigdefault.NewStore(), + scf, + registry, + stovepipemq.TopicKeyProcess, + "stovepipe-process", + ) + if err := c.Register(processController); err != nil { + return count, fmt.Errorf("failed to register process controller: %w", err) + } + count++ + + buildController := build.NewController(logger, scope, store, brf, registry, stovepipemq.TopicKeyBuild, "stovepipe-build") + if err := c.Register(buildController); err != nil { + return count, fmt.Errorf("failed to register build controller: %w", err) + } + count++ + + return count, nil +} + +// registerDLQControllers creates one DLQ reconciler per primary stage and +// registers them with c, returning how many were registered. +func registerDLQControllers( + c consumer.Consumer, + logger *zap.SugaredLogger, + scope tally.Scope, + store storage.Storage, + registry consumer.TopicRegistry, +) (int, error) { + var count int + + processDLQController := dlq.NewController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq") + if err := c.Register(processDLQController); err != nil { + return count, fmt.Errorf("failed to register process dlq controller: %w", err) + } + count++ + + return count, nil +} + // newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest -// publishes to process; process publishes admitted requests to the publish-only build topic. -// The process_dlq topic is the dead-letter destination the queue backend routes to (per -// DefaultSubscriptionConfig's DLQ.TopicSuffix) when the process controller exhausts retries. +// publishes to the process topic and the process consumer subscribes to it; process publishes +// to the build topic and the build consumer subscribes to it. The buildsignal topic is added +// once the buildsignal controller lands to consume it. func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) { return consumer.NewTopicRegistry([]consumer.TopicConfig{ { @@ -356,6 +423,9 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Key: stovepipemq.TopicKeyBuild, Name: "build", Queue: q, + Subscription: extqueue.DefaultSubscriptionConfig( + subscriberName, "stovepipe-build", + ), }, { Key: dlq.TopicKey(stovepipemq.TopicKeyProcess), 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/buildrunner/BUILD.bazel b/stovepipe/extension/buildrunner/BUILD.bazel new file mode 100644 index 00000000..0f5451e6 --- /dev/null +++ b/stovepipe/extension/buildrunner/BUILD.bazel @@ -0,0 +1,9 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["buildrunner.go"], + importpath = "github.com/uber/submitqueue/stovepipe/extension/buildrunner", + visibility = ["//visibility:public"], + deps = ["//stovepipe/entity:go_default_library"], +) diff --git a/stovepipe/extension/buildrunner/README.md b/stovepipe/extension/buildrunner/README.md new file mode 100644 index 00000000..3abb05c0 --- /dev/null +++ b/stovepipe/extension/buildrunner/README.md @@ -0,0 +1,11 @@ +# BuildRunner + +Vendor-agnostic interface through which Stovepipe triggers and polls builds against an external build system. + +- **Trigger** starts a new build against `headURI`, optionally relative to an incremental `baseURI`, and returns the runner-minted build id. There is no caller-supplied dedup input — every call starts a fresh build; downstream idempotency absorbs any duplicate from a redelivery. Trigger must return promptly; the build itself runs asynchronously. +- **Status** polls the current status and any provider metadata for a build id `Trigger` returned. Unlike `Trigger`, it may round-trip to the backend and block. +- **Cancel** requests cancellation for a build id, returning once the request reaches the runner rather than once the build actually stops. Unused today; kept for contract parity with SubmitQueue. + +Implementations return plain, unclassified errors — the calling controller decides retryable-vs-not and user-vs-infra, per `platform/errs`. + +See [doc/rfc/stovepipe/steps/build.md](../../../doc/rfc/stovepipe/steps/build.md#why-separate-contracts) for why this is a separate contract from SubmitQueue's own `buildrunner` rather than a shared one. To add a backend, create `buildrunner/{backend}/`, implement `BuildRunner`, and return it from a `New(...)` constructor. diff --git a/stovepipe/extension/buildrunner/buildrunner.go b/stovepipe/extension/buildrunner/buildrunner.go new file mode 100644 index 00000000..4afa39be --- /dev/null +++ b/stovepipe/extension/buildrunner/buildrunner.go @@ -0,0 +1,86 @@ +// 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 buildrunner defines the contract through which Stovepipe triggers and +// polls builds against an external build system. It is shaped the same as +// SubmitQueue's own buildrunner extension — same Trigger/Status/Cancel verbs, +// same async contract, same id model — but is a separate interface rather than a +// shared one: Stovepipe validates a single commit against a baseline (or from +// scratch), not a stack of dependency batches, so Trigger takes URI identity +// instead of batch identity. See doc/rfc/stovepipe/steps/build.md's "Why separate +// contracts" for the full rationale. +package buildrunner + +//go:generate mockgen -source=buildrunner.go -destination=mock/buildrunner_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// BuildRunner triggers builds against an external build system, polls their +// status, and cancels them. Implementations are long-lived singletons and must: +// - be safe for concurrent use by multiple goroutines; +// - recover from transient connectivity failures internally, returning plain +// errors during the recovery window rather than blocking the caller +// indefinitely; +// - keep only transient local state (caches, pools) — the durable link between +// a Request and its Build lives in BuildStore, never in the runner; +// - return plain, unclassified errors and leave user-vs-infra and +// retryable-vs-not classification to the calling controller, per +// platform/errs. +type BuildRunner interface { + // Trigger starts a new build every call and mints the build's identity — + // there is no caller-supplied dedup input. baseURI is the incremental + // baseline, empty for a full build; headURI is the commit under + // validation; both are opaque tokens owned by SourceControl. metadata is + // caller-supplied annotation the runner may echo back via Status but must + // not depend on. Trigger is async: it must return promptly with the + // runner-assigned id, not an outcome — callers learn progress via Status. + Trigger(ctx context.Context, baseURI, headURI string, metadata entity.BuildMetadata) (entity.BuildID, error) + + // Status returns the build's current status and any provider metadata for + // the id Trigger returned. Unlike Trigger, Status may round-trip to the + // backend and block. The returned BuildMetadata is caller-supplied, + // provider-echoed — the runner must not depend on it, but a caller may + // read it for its own purposes. + Status(ctx context.Context, buildID entity.BuildID) (entity.BuildStatus, entity.BuildMetadata, error) + + // Cancel requests cancellation of the build for the id Trigger returned, + // returning once the request reaches the runner, not once the build + // actually stops. A no-op on an already-terminal build. No controller + // calls Cancel today — see doc/rfc/stovepipe/steps/build.md's + // "Cancellation: defined, not yet called" — it exists for contract parity + // and future use. + Cancel(ctx context.Context, buildID entity.BuildID) error +} + +// Config carries the per-queue identity handed to a Factory. It is the only +// identity the system hands a Factory; everything else a concrete BuildRunner +// needs (endpoint, credentials, pipeline mapping) is injected by the integrator +// at construction. +type Config struct { + // QueueName identifies which Queue's build-runner backend to resolve. + QueueName string +} + +// Factory resolves the BuildRunner for a Config. Implementations and the +// per-queue routing that picks a backend for a Config.QueueName live in the +// wiring layer (service/stovepipe/.../server/main.go), not here — see +// CLAUDE.md's extension rules. +type Factory interface { + // For returns the BuildRunner for the given queue. + For(cfg Config) (BuildRunner, error) +} diff --git a/stovepipe/extension/buildrunner/fake/BUILD.bazel b/stovepipe/extension/buildrunner/fake/BUILD.bazel new file mode 100644 index 00000000..3157397b --- /dev/null +++ b/stovepipe/extension/buildrunner/fake/BUILD.bazel @@ -0,0 +1,24 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["fake.go"], + importpath = "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake", + visibility = ["//visibility:public"], + deps = [ + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/buildrunner:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["fake_test.go"], + embed = [":go_default_library"], + deps = [ + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/buildrunner:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/stovepipe/extension/buildrunner/fake/fake.go b/stovepipe/extension/buildrunner/fake/fake.go new file mode 100644 index 00000000..f32b0ed7 --- /dev/null +++ b/stovepipe/extension/buildrunner/fake/fake.go @@ -0,0 +1,137 @@ +// 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 fake provides a buildrunner.BuildRunner whose outcome is driven by the +// triggered head URI. With no marker every build immediately succeeds, behaving +// as a best-case stub for local-stack/e2e wiring. Failures are injected by +// embedding a marker token in headURI of the form "buildrunner-fake=": +// +// buildrunner-fake=trigger-error -> Trigger returns a non-nil error +// buildrunner-fake=build-fail -> Status reports BuildStatusFailed +// buildrunner-fake=build-error -> Status returns a non-nil error +// +// The runner is stateless: Trigger encodes the desired terminal outcome into the +// returned BuildID, and Status decides the result purely from the BuildID it is +// given — no per-build bookkeeping. This means any runner instance can answer +// Status for an id minted by any other (Trigger and Status can even live in +// different processes), and a single running stack can exercise the negative +// paths purely by varying request payloads. It is intended for examples and +// tests only, never production. +package fake + +import ( + "context" + "crypto/rand" + "encoding/hex" + "fmt" + "strings" + + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/buildrunner" +) + +// markerPrefix introduces a marker token in headURI: "buildrunner-fake=". +const markerPrefix = "buildrunner-fake=" + +// Recognized marker tokens. See the package doc for the convention. +const ( + tokenTriggerError = "trigger-error" + tokenFail = "build-fail" + tokenError = "build-error" +) + +// outcomeOK is the BuildID outcome segment for a build that should succeed. +const outcomeOK = "ok" + +// runner is a buildrunner.BuildRunner that reports every build as succeeded +// unless a marker token in headURI requests otherwise. It holds no per-build +// state: the outcome is encoded in the BuildID at Trigger and read back out at +// Status. Uniqueness comes from a random suffix per id, so it needs no shared +// counter and never collides across instances or processes. +type runner struct{} + +// New returns a buildrunner.BuildRunner that defaults to succeeding and honors +// marker tokens embedded in the triggered headURI. +func New() buildrunner.BuildRunner { + return runner{} +} + +// Trigger fails when headURI carries the trigger-error marker; otherwise it +// returns a unique BuildID that encodes the terminal outcome the build should +// report at Status time (decided from the headURI marker). baseURI and metadata +// are ignored. +func (r runner) Trigger(_ context.Context, _, headURI string, _ entity.BuildMetadata) (entity.BuildID, error) { + outcome := outcomeOK + switch marker(headURI) { + case tokenTriggerError: + return entity.BuildID{}, fmt.Errorf("fake: marked trigger error") + case tokenFail: + outcome = tokenFail + case tokenError: + outcome = tokenError + } + + // Encode the outcome in the id (e.g. "fake-build-fail-a1b2c3d4") so Status is + // stateless. The random suffix keeps ids globally unique across instances and + // processes without any shared state. + suffix, err := randomSuffix() + if err != nil { + return entity.BuildID{}, fmt.Errorf("fake: generating build id: %w", err) + } + return entity.BuildID{ID: fmt.Sprintf("fake-%s-%s", outcome, suffix)}, nil +} + +// Status decides the result purely from the BuildID's encoded outcome. Ids that +// carry no recognized outcome (including those not minted by this fake) default +// to succeeded, keeping the runner best-case. +func (r runner) Status(_ context.Context, buildID entity.BuildID) (entity.BuildStatus, entity.BuildMetadata, error) { + switch { + case strings.Contains(buildID.ID, tokenError): + return entity.BuildStatusUnknown, nil, fmt.Errorf("fake: marked build error") + case strings.Contains(buildID.ID, tokenFail): + return entity.BuildStatusFailed, nil, nil + default: + return entity.BuildStatusSucceeded, nil, nil + } +} + +// Cancel is a no-op and always succeeds. +func (r runner) Cancel(_ context.Context, _ entity.BuildID) error { + return nil +} + +// marker returns the marker token embedded in uri, or "" if none is present. +// The token ends at the first "&" or "#" delimiter, so a marker may sit among +// other query parameters or a fragment. +func marker(uri string) string { + _, rest, found := strings.Cut(uri, markerPrefix) + if !found { + return "" + } + if i := strings.IndexAny(rest, "&#"); i >= 0 { + rest = rest[:i] + } + return rest +} + +// randomSuffix returns a short random hex string used to keep fake BuildIDs +// globally unique. Hex digits never spell the outcome marker tokens, so the +// suffix cannot interfere with Status decoding the outcome via substring match. +func randomSuffix() (string, error) { + var b [4]byte + if _, err := rand.Read(b[:]); err != nil { + return "", err + } + return hex.EncodeToString(b[:]), nil +} diff --git a/stovepipe/extension/buildrunner/fake/fake_test.go b/stovepipe/extension/buildrunner/fake/fake_test.go new file mode 100644 index 00000000..1e2ffe28 --- /dev/null +++ b/stovepipe/extension/buildrunner/fake/fake_test.go @@ -0,0 +1,112 @@ +// 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 fake + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/buildrunner" +) + +func TestNew_ImplementsInterface(t *testing.T) { + var _ buildrunner.BuildRunner = New() +} + +func TestTrigger(t *testing.T) { + tests := []struct { + name string + headURI string + wantErr bool + }{ + {name: "no marker succeeds", headURI: "git://repo/ref/deadbeef"}, + {name: "unrelated query params succeed", headURI: "git://repo/ref/deadbeef?attempt=2"}, + {name: "trigger-error marker fails", headURI: "git://repo/ref/deadbeef?buildrunner-fake=trigger-error", wantErr: true}, + {name: "build-fail marker still triggers", headURI: "git://repo/ref/deadbeef?buildrunner-fake=build-fail"}, + {name: "build-error marker still triggers", headURI: "git://repo/ref/deadbeef?buildrunner-fake=build-error"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + id, err := New().Trigger(context.Background(), "", tt.headURI, nil) + if tt.wantErr { + require.Error(t, err) + assert.Empty(t, id.ID) + return + } + require.NoError(t, err) + assert.NotEmpty(t, id.ID) + }) + } +} + +func TestTrigger_UniqueIDs(t *testing.T) { + a, err := New().Trigger(context.Background(), "", "git://repo/ref/deadbeef", nil) + require.NoError(t, err) + b, err := New().Trigger(context.Background(), "", "git://repo/ref/deadbeef", nil) + require.NoError(t, err) + assert.NotEqual(t, a.ID, b.ID) +} + +func TestStatus(t *testing.T) { + tests := []struct { + name string + headURI string + wantStatus entity.BuildStatus + wantErr bool + }{ + {name: "no marker succeeds", headURI: "git://repo/ref/deadbeef", wantStatus: entity.BuildStatusSucceeded}, + {name: "build-fail marker fails", headURI: "git://repo/ref/deadbeef?buildrunner-fake=build-fail", wantStatus: entity.BuildStatusFailed}, + {name: "build-error marker errors", headURI: "git://repo/ref/deadbeef?buildrunner-fake=build-error", wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + id, err := New().Trigger(context.Background(), "", tt.headURI, nil) + require.NoError(t, err) + + status, metadata, err := New().Status(context.Background(), id) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.wantStatus, status) + assert.Nil(t, metadata) + }) + } +} + +func TestStatus_UnrecognizedIDSucceeds(t *testing.T) { + status, metadata, err := New().Status(context.Background(), entity.BuildID{ID: "not-minted-by-this-fake"}) + require.NoError(t, err) + assert.Equal(t, entity.BuildStatusSucceeded, status) + assert.Nil(t, metadata) +} + +func TestStatus_StatelessAcrossInstances(t *testing.T) { + id, err := New().Trigger(context.Background(), "", "git://repo/ref/deadbeef?buildrunner-fake=build-fail", nil) + require.NoError(t, err) + + status, _, err := New().Status(context.Background(), id) + require.NoError(t, err) + assert.Equal(t, entity.BuildStatusFailed, status) +} + +func TestCancel_NoOp(t *testing.T) { + err := New().Cancel(context.Background(), entity.BuildID{ID: "anything"}) + assert.NoError(t, err) +} diff --git a/stovepipe/extension/buildrunner/mock/BUILD.bazel b/stovepipe/extension/buildrunner/mock/BUILD.bazel new file mode 100644 index 00000000..5e4d317b --- /dev/null +++ b/stovepipe/extension/buildrunner/mock/BUILD.bazel @@ -0,0 +1,13 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["buildrunner_mock.go"], + importpath = "github.com/uber/submitqueue/stovepipe/extension/buildrunner/mock", + visibility = ["//visibility:public"], + deps = [ + "//stovepipe/entity:go_default_library", + "//stovepipe/extension/buildrunner:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/stovepipe/extension/buildrunner/mock/buildrunner_mock.go b/stovepipe/extension/buildrunner/mock/buildrunner_mock.go new file mode 100644 index 00000000..61f1c9d0 --- /dev/null +++ b/stovepipe/extension/buildrunner/mock/buildrunner_mock.go @@ -0,0 +1,127 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: buildrunner.go +// +// Generated by this command: +// +// mockgen -source=buildrunner.go -destination=mock/buildrunner_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + buildrunner "github.com/uber/submitqueue/stovepipe/extension/buildrunner" + gomock "go.uber.org/mock/gomock" +) + +// MockBuildRunner is a mock of BuildRunner interface. +type MockBuildRunner struct { + ctrl *gomock.Controller + recorder *MockBuildRunnerMockRecorder + isgomock struct{} +} + +// MockBuildRunnerMockRecorder is the mock recorder for MockBuildRunner. +type MockBuildRunnerMockRecorder struct { + mock *MockBuildRunner +} + +// NewMockBuildRunner creates a new mock instance. +func NewMockBuildRunner(ctrl *gomock.Controller) *MockBuildRunner { + mock := &MockBuildRunner{ctrl: ctrl} + mock.recorder = &MockBuildRunnerMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockBuildRunner) EXPECT() *MockBuildRunnerMockRecorder { + return m.recorder +} + +// Cancel mocks base method. +func (m *MockBuildRunner) Cancel(ctx context.Context, buildID entity.BuildID) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Cancel", ctx, buildID) + ret0, _ := ret[0].(error) + return ret0 +} + +// Cancel indicates an expected call of Cancel. +func (mr *MockBuildRunnerMockRecorder) Cancel(ctx, buildID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Cancel", reflect.TypeOf((*MockBuildRunner)(nil).Cancel), ctx, buildID) +} + +// Status mocks base method. +func (m *MockBuildRunner) Status(ctx context.Context, buildID entity.BuildID) (entity.BuildStatus, entity.BuildMetadata, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Status", ctx, buildID) + ret0, _ := ret[0].(entity.BuildStatus) + ret1, _ := ret[1].(entity.BuildMetadata) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 +} + +// Status indicates an expected call of Status. +func (mr *MockBuildRunnerMockRecorder) Status(ctx, buildID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Status", reflect.TypeOf((*MockBuildRunner)(nil).Status), ctx, buildID) +} + +// Trigger mocks base method. +func (m *MockBuildRunner) Trigger(ctx context.Context, baseURI, headURI string, metadata entity.BuildMetadata) (entity.BuildID, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Trigger", ctx, baseURI, headURI, metadata) + ret0, _ := ret[0].(entity.BuildID) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Trigger indicates an expected call of Trigger. +func (mr *MockBuildRunnerMockRecorder) Trigger(ctx, baseURI, headURI, metadata any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trigger", reflect.TypeOf((*MockBuildRunner)(nil).Trigger), ctx, baseURI, headURI, metadata) +} + +// MockFactory is a mock of Factory interface. +type MockFactory struct { + ctrl *gomock.Controller + recorder *MockFactoryMockRecorder + isgomock struct{} +} + +// MockFactoryMockRecorder is the mock recorder for MockFactory. +type MockFactoryMockRecorder struct { + mock *MockFactory +} + +// NewMockFactory creates a new mock instance. +func NewMockFactory(ctrl *gomock.Controller) *MockFactory { + mock := &MockFactory{ctrl: ctrl} + mock.recorder = &MockFactoryMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockFactory) EXPECT() *MockFactoryMockRecorder { + return m.recorder +} + +// For mocks base method. +func (m *MockFactory) For(cfg buildrunner.Config) (buildrunner.BuildRunner, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "For", cfg) + ret0, _ := ret[0].(buildrunner.BuildRunner) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// For indicates an expected call of For. +func (mr *MockFactoryMockRecorder) For(cfg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockFactory)(nil).For), cfg) +} 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.