From 30413ada0b85bf4831e88ebcc92c4455e7d85035 Mon Sep 17 00:00:00 2001 From: "chenghan.ying" Date: Thu, 16 Jul 2026 21:26:33 +0000 Subject: [PATCH 1/5] add buildsignal controller --- service/stovepipe/server/BUILD.bazel | 1 + service/stovepipe/server/main.go | 30 +- stovepipe/controller/buildsignal/BUILD.bazel | 43 +++ .../controller/buildsignal/buildsignal.go | 285 +++++++++++++++ .../buildsignal/buildsignal_test.go | 328 ++++++++++++++++++ stovepipe/core/messagequeue/messagequeue.go | 4 + .../core/messagequeue/messagequeue_test.go | 14 +- stovepipe/core/messagequeue/proto/BUILD.bazel | 1 + .../core/messagequeue/proto/record.proto | 37 ++ .../core/messagequeue/protopb/BUILD.bazel | 1 + .../core/messagequeue/protopb/record.pb.go | 147 ++++++++ stovepipe/core/messagequeue/topics.go | 6 + tool/proto/BUILD.bazel | 1 + 13 files changed, 894 insertions(+), 4 deletions(-) create mode 100644 stovepipe/controller/buildsignal/BUILD.bazel create mode 100644 stovepipe/controller/buildsignal/buildsignal.go create mode 100644 stovepipe/controller/buildsignal/buildsignal_test.go create mode 100644 stovepipe/core/messagequeue/proto/record.proto create mode 100644 stovepipe/core/messagequeue/protopb/record.pb.go diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index 49158b2c..faf7b91e 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -16,6 +16,7 @@ go_library( "//service/stovepipe/server/mapper:go_default_library", "//stovepipe/controller:go_default_library", "//stovepipe/controller/build:go_default_library", + "//stovepipe/controller/buildsignal:go_default_library", "//stovepipe/controller/dlq:go_default_library", "//stovepipe/controller/process:go_default_library", "//stovepipe/core/messagequeue:go_default_library", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 8e40c51f..3bce151f 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -38,6 +38,7 @@ import ( "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/buildsignal" "github.com/uber/submitqueue/stovepipe/controller/dlq" "github.com/uber/submitqueue/stovepipe/controller/process" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" @@ -254,6 +255,16 @@ func run() error { } logger.Info("controllers registered", zap.Int("primary", primaryCount), zap.Int("dlq", dlqCount)) + buildController := build.NewController(logger.Sugar(), scope, store, fakeBuildRunnerFactory{}, registry, stovepipemq.TopicKeyBuild, "stovepipe-build") + if err := primaryConsumer.Register(buildController); err != nil { + return fmt.Errorf("failed to register build controller: %w", err) + } + + buildSignalController := buildsignal.NewController(logger.Sugar(), scope, store, fakeBuildRunnerFactory{}, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal") + if err := primaryConsumer.Register(buildSignalController); err != nil { + return fmt.Errorf("failed to register buildsignal controller: %w", err) + } + // Start consumers. DLQ first because Start begins processing messages // immediately; if the primary consumer then fails to start, the half we // already started is the DLQ side, whose work is idempotent reconciliation @@ -407,8 +418,10 @@ func registerDLQControllers( // newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest // 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. +// to the build topic and the build consumer subscribes to it; build publishes to the buildsignal +// topic and the buildsignal consumer subscribes to it, and also republishes to itself while +// polling. buildsignal publishes to the record topic once a build reaches a terminal status; it +// has no Subscription yet since no consumer for it exists until the record stage lands. func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) { return consumer.NewTopicRegistry([]consumer.TopicConfig{ { @@ -426,11 +439,22 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Subscription: extqueue.DefaultSubscriptionConfig( subscriberName, "stovepipe-build", ), + + }, + { + Key: stovepipemq.TopicKeyBuildSignal, + Subscription: extqueue.DefaultSubscriptionConfig( + subscriberName, "stovepipe-buildsignal", + ), + }, + { + Key: stovepipemq.TopicKeyRecord, + Name: "record", + Queue: q, }, { Key: dlq.TopicKey(stovepipemq.TopicKeyProcess), Name: "process_dlq", - Queue: q, Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-process-dlq"), }, }) diff --git a/stovepipe/controller/buildsignal/BUILD.bazel b/stovepipe/controller/buildsignal/BUILD.bazel new file mode 100644 index 00000000..cd9f5a38 --- /dev/null +++ b/stovepipe/controller/buildsignal/BUILD.bazel @@ -0,0 +1,43 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["buildsignal.go"], + importpath = "github.com/uber/submitqueue/stovepipe/controller/buildsignal", + 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 = ["buildsignal_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/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go new file mode 100644 index 00000000..bbe8e780 --- /dev/null +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -0,0 +1,285 @@ +// 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 buildsignal holds the buildsignal-stage queue controller. It consumes +// BuildSignal messages (a build id), polls the build-runner until the build +// reaches a terminal status, persists that status, and — once terminal — +// publishes the build id to record. See +// doc/rfc/stovepipe/steps/buildsignal.md. +package buildsignal + +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" +) + +// Poll delays for non-terminal statuses. Vars (not consts) so tests can +// shorten them; the server always uses the defaults. +// +// TODO: move these behind a queueconfig-style extension so operators can +// tune poll cadence per queue without a code change. +var ( + // PollDelayAcceptedMs is the delay before the next Status call while the + // build is queued by the runner but has not started executing. + PollDelayAcceptedMs int64 = 5000 + // PollDelayRunningMs is the delay before the next Status call while the + // build is executing. + PollDelayRunningMs int64 = 2000 +) + +// Controller consumes BuildSignal messages, polls the build-runner toward a +// terminal status, persists the result, and either reschedules itself or +// publishes the build id to record. 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 = "buildsignal" + +// NewController creates a new buildsignal 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("buildsignal_controller"), + metricsScope: scope.SubScope("buildsignal_controller"), + store: store, + buildRunners: buildRunners, + registry: registry, + topicKey: topicKey, + consumerGroup: consumerGroup, + } +} + +// Process reloads the build referenced by the delivery, polls its runner for +// the latest status, persists a real transition, and either reschedules a +// poll or, once terminal, publishes the build id to record. 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() + + sig := &stovepipemq.BuildSignal{} + if err := stovepipemq.Unmarshal(msg.Payload, sig); 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 signal: %w", err) + } + + build, err := c.loadBuild(ctx, sig.Id) + if err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1) + return err + } + + request, err := c.loadRequest(ctx, build.RequestID) + if err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1) + return err + } + + // The request is done (record already ran, or the head was superseded); + // stop polling. + 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("BuildSignalController failed to resolve build runner for queue %s: %w", request.Queue, err) + } + + status, _, err := buildRunner.Status(ctx, entity.BuildID{ID: build.ID}) + if err != nil { + return fmt.Errorf("BuildSignalController failed to poll status for build %s: %w", build.ID, err) + } + + effective, err := c.reconcile(ctx, build, status) + if err != nil { + return err + } + + if effective.IsTerminal() { + if err := c.publishRecord(ctx, build.ID, request.ID); err != nil { + return fmt.Errorf("BuildSignalController failed to publish record for build %s: %w", build.ID, err) + } + c.logger.Infow("build reached terminal status", + "build_id", build.ID, + "request_id", request.ID, + "status", string(effective), + ) + return nil + } + + delayMs := pollDelay(effective) + if err := c.publishBuildSignal(ctx, build.ID, delayMs); err != nil { + return errs.NewRetryableError(fmt.Errorf("BuildSignalController failed to reschedule poll for build %s: %w", build.ID, err)) + } + c.logger.Debugw("rescheduled build status poll", + "build_id", build.ID, + "status", string(effective), + "delay_ms", delayMs, + ) + return nil +} + +// reconcile persists a real status transition and returns the status that +// should drive the rest of Process: the polled status when persisted (or +// already unchanged), or the stored status when a stored terminal status is +// write-once-protected against a differing poll. +func (c *Controller) reconcile(ctx context.Context, build entity.Build, status entity.BuildStatus) (entity.BuildStatus, error) { + if status == build.Status { + return build.Status, nil + } + + // Terminal is write-once: a later poll of a flaky backend must never + // overwrite an already-committed terminal status. + if build.Status.IsTerminal() { + return build.Status, nil + } + + newVersion := build.Version + 1 + updated := build + updated.Status = status + if err := c.store.GetBuildStore().Update(ctx, updated, build.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + return "", errs.NewRetryableError(fmt.Errorf("build %s version conflict: %w", build.ID, err)) + } + return "", fmt.Errorf("BuildSignalController failed to persist status for build %s: %w", build.ID, err) + } + return status, nil +} + +// loadBuild returns the build for id. A not-yet-visible row is retryable. +func (c *Controller) loadBuild(ctx context.Context, id string) (entity.Build, error) { + got, err := c.store.GetBuildStore().Get(ctx, id) + if err == nil { + return got, nil + } + if errors.Is(err, storage.ErrNotFound) { + return entity.Build{}, errs.NewRetryableError(fmt.Errorf("build %s not found yet: %w", id, err)) + } + return entity.Build{}, fmt.Errorf("BuildSignalController failed to load build %s: %w", id, err) +} + +// loadRequest returns the request for id. A miss here is almost certainly a +// lagging read behind the Build that referenced it, so it is retryable; a +// genuinely orphaned Build still dead-letters at MaxAttempts. +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)) + } + return entity.Request{}, fmt.Errorf("BuildSignalController failed to load request %s: %w", id, err) +} + +// pollDelay returns the delay before the next Status call for a non-terminal status. +func pollDelay(status entity.BuildStatus) int64 { + switch status { + case entity.BuildStatusRunning: + return PollDelayRunningMs + default: + // Accepted and any unknown non-terminal status. + return PollDelayAcceptedMs + } +} + +// publishRecord publishes buildID to the record stage, partitioned by +// requestID. +func (c *Controller) publishRecord(ctx context.Context, buildID, requestID string) error { + payload, err := stovepipemq.Marshal(&stovepipemq.Record{Id: buildID}) + if err != nil { + return fmt.Errorf("failed to serialize record: %w", err) + } + msg := entityqueue.NewMessage(buildID, payload, requestID, nil) + return c.publish(ctx, stovepipemq.TopicKeyRecord, msg, 0) +} + +// publishBuildSignal re-publishes buildID to buildsignal after delayMs, +// partitioned by build id so each build's poll loop runs in its own +// partition. A fresh message, not a nack — polling is not failure. +func (c *Controller) publishBuildSignal(ctx context.Context, buildID string, delayMs int64) 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) + return c.publish(ctx, stovepipemq.TopicKeyBuildSignal, msg, delayMs) +} + +// publish sends msg to the queue registered for key, using PublishAfter when +// delayMs > 0 and Publish otherwise. +func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, msg entityqueue.Message, delayMs int64) error { + q, ok := c.registry.Queue(key) + if !ok { + return fmt.Errorf("no queue registered for topic key %s", key) + } + topicName, ok := c.registry.TopicName(key) + if !ok { + return fmt.Errorf("no topic name registered for topic key %s", key) + } + if delayMs > 0 { + return q.Publisher().PublishAfter(ctx, topicName, msg, delayMs) + } + return q.Publisher().Publish(ctx, topicName, msg) +} + +// Name returns the controller name for logging and metrics. +func (c *Controller) Name() string { + return "buildsignal" +} + +// 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/buildsignal/buildsignal_test.go b/stovepipe/controller/buildsignal/buildsignal_test.go new file mode 100644 index 00000000..33aa02f0 --- /dev/null +++ b/stovepipe/controller/buildsignal/buildsignal_test.go @@ -0,0 +1,328 @@ +// 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 buildsignal + +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" + testBuildID = "bk-1" +) + +// buildsignalMocks bundles the mocks a buildsignal controller test case wires +// expectations on. +type buildsignalMocks 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, buildsignalMocks) { + t.Helper() + + m := buildsignalMocks{ + 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}, + {Key: stovepipemq.TopicKeyRecord, Name: "record", Queue: queue}, + }) + require.NoError(t, err) + + c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, m.runnerFactory, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal") + 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(testBuildID, payload, testBuildID, nil)).AnyTimes() + d.EXPECT().Attempt().Return(1).AnyTimes() + return d +} + +func buildSignalPayload(t *testing.T, id string) []byte { + t.Helper() + b, err := stovepipemq.Marshal(&stovepipemq.BuildSignal{Id: id}) + require.NoError(t, err) + return b +} + +// requestWithState returns a Request past process's admit, in the given state. +func requestWithState(state entity.RequestState) entity.Request { + return entity.Request{ + ID: testID, + Queue: testQueue, + State: state, + Version: 1, + } +} + +// build returns a Build with the given status/version, tied to testID. +func build(status entity.BuildStatus, version int32) entity.Build { + return entity.Build{ + ID: testBuildID, + RequestID: testID, + Status: status, + Version: version, + } +} + +func TestProcess(t *testing.T) { + tests := []struct { + name string + payload []byte + setup func(m buildsignalMocks) + wantErr bool + wantRetry bool + }{ + { + name: "build not found is retryable", + wantErr: true, + wantRetry: true, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{}, storage.ErrNotFound) + }, + }, + { + name: "build storage error is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{}, errors.New("db down")) + }, + }, + { + name: "request not found is retryable", + wantErr: true, + wantRetry: true, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusAccepted, 1), nil) + 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 buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusAccepted, 1), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, errors.New("db down")) + }, + }, + { + name: "superseded request is a no-op", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateSuperseded), nil) + }, + }, + { + name: "recorded green request is a no-op", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateRecordedGreen), nil) + }, + }, + { + name: "recorded not green request is a no-op", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateRecordedNotGreen), nil) + }, + }, + { + name: "factory lookup failure is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusAccepted, 1), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(nil, errors.New("no runner")) + }, + }, + { + name: "status call error is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusAccepted, 1), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatus(""), nil, errors.New("runner unavailable")) + }, + }, + { + name: "unchanged status skips write and reschedules", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusRunning, nil, nil) + m.publisher.EXPECT().PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs).Return(nil) + }, + }, + { + name: "status transition persists and reschedules", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusAccepted, 1), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusRunning, nil, nil) + updated := build(entity.BuildStatusRunning, 1) + m.buildStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil) + m.publisher.EXPECT().PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs).Return(nil) + }, + }, + { + name: "transition to terminal persists and publishes to record", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) + updated := build(entity.BuildStatusSucceeded, 2) + m.buildStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) + m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil) + }, + }, + { + name: "write-once: stored terminal status is never overwritten", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusSucceeded, 5), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusFailed, nil, nil) + // No Update call expected: a stored terminal status is write-once. + m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil) + }, + }, + { + name: "version conflict is retryable", + wantErr: true, + wantRetry: true, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusAccepted, 1), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusRunning, nil, nil) + m.buildStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(storage.ErrVersionMismatch) + }, + }, + { + name: "build store update error is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusAccepted, 1), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusRunning, nil, nil) + m.buildStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(errors.New("db down")) + }, + }, + { + name: "publish to record failure is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) + m.buildStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(2), int32(3)).Return(nil) + m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(errors.New("queue down")) + }, + }, + { + name: "reschedule publish failure is retryable", + wantErr: true, + wantRetry: true, + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusAccepted, 1), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusAccepted, nil, nil) + m.publisher.EXPECT().PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayAcceptedMs).Return(errors.New("queue down")) + }, + }, + { + name: "malformed payload is not retryable", + payload: []byte("not-json"), + wantErr: true, + wantRetry: false, + setup: func(m buildsignalMocks) {}, + }, + } + + 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 = buildSignalPayload(t, testBuildID) + } + + 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/core/messagequeue/messagequeue.go b/stovepipe/core/messagequeue/messagequeue.go index 0beb8f2a..867eb3aa 100644 --- a/stovepipe/core/messagequeue/messagequeue.go +++ b/stovepipe/core/messagequeue/messagequeue.go @@ -49,6 +49,10 @@ type ( // BuildSignal is the payload build publishes to the buildsignal stage, and // buildsignal re-publishes to itself while polling: the build id to poll. BuildSignal = protopb.BuildSignal + + // Record is the payload buildsignal publishes to the record stage once a + // build reaches a terminal status: the build id to record. + Record = protopb.Record ) // marshalOpts keeps the JSON field names identical to the proto field names diff --git a/stovepipe/core/messagequeue/messagequeue_test.go b/stovepipe/core/messagequeue/messagequeue_test.go index 9d1635db..141dddda 100644 --- a/stovepipe/core/messagequeue/messagequeue_test.go +++ b/stovepipe/core/messagequeue/messagequeue_test.go @@ -55,6 +55,17 @@ func TestBuildSignalRoundTrip(t *testing.T) { assert.True(t, proto.Equal(sig, got), "round-tripped BuildSignal should equal the original") } +func TestRecordRoundTrip(t *testing.T) { + rec := &Record{Id: "bk-1001"} + + data, err := Marshal(rec) + require.NoError(t, err) + + got := &Record{} + require.NoError(t, Unmarshal(data, got)) + assert.True(t, proto.Equal(rec, got), "round-tripped Record should equal the original") +} + // TestWireFormat locks the protojson encoding decision the contract relies on: // snake_case field names (UseProtoNames). func TestWireFormat(t *testing.T) { @@ -69,7 +80,7 @@ func TestWireFormat(t *testing.T) { // no topic_keys option names an unknown key. func TestTopicKeysBindEveryTopicKey(t *testing.T) { bound := map[string]int{} - for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}} { + for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &Record{}} { keys := TopicKeys(m) require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option") for _, key := range keys { @@ -81,6 +92,7 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) { TopicKeyProcess, TopicKeyBuild, TopicKeyBuildSignal, + TopicKeyRecord, } valid := map[string]bool{} diff --git a/stovepipe/core/messagequeue/proto/BUILD.bazel b/stovepipe/core/messagequeue/proto/BUILD.bazel index f625ba47..5f2390a8 100644 --- a/stovepipe/core/messagequeue/proto/BUILD.bazel +++ b/stovepipe/core/messagequeue/proto/BUILD.bazel @@ -3,6 +3,7 @@ exports_files( "build.proto", "buildsignal.proto", "process.proto", + "record.proto", ], visibility = ["//tool/proto:__pkg__"], ) diff --git a/stovepipe/core/messagequeue/proto/record.proto b/stovepipe/core/messagequeue/proto/record.proto new file mode 100644 index 00000000..0ca18e58 --- /dev/null +++ b/stovepipe/core/messagequeue/proto/record.proto @@ -0,0 +1,37 @@ +// 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. + +syntax = "proto3"; + +package uber.stovepipe.messagequeue; + +import "api/base/messagequeue/proto/messagequeue.proto"; + +option go_package = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb"; +option java_multiple_files = true; +option java_outer_classname = "RecordProto"; +option java_package = "com.uber.submitqueue.stovepipe.messagequeue"; + +// Record is the payload buildsignal publishes once a build reaches a terminal +// status: only the build id travels on the queue. record reloads the full +// Build from storage by this id (producer and consumer share the store, so +// the id is enough and redelivery stays idempotent). See +// doc/rfc/stovepipe/steps/buildsignal.md. +message Record { + option (uber.base.messagequeue.topic_keys) = "record"; + + // id is the terminal build's id: the runner-assigned id minted by + // BuildRunner.Trigger (entity.Build.ID / entity.BuildID.ID). + string id = 1; +} diff --git a/stovepipe/core/messagequeue/protopb/BUILD.bazel b/stovepipe/core/messagequeue/protopb/BUILD.bazel index 5c59a722..7f61fb89 100644 --- a/stovepipe/core/messagequeue/protopb/BUILD.bazel +++ b/stovepipe/core/messagequeue/protopb/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "build.pb.go", "buildsignal.pb.go", "process.pb.go", + "record.pb.go", ], importpath = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb", visibility = ["//visibility:public"], diff --git a/stovepipe/core/messagequeue/protopb/record.pb.go b/stovepipe/core/messagequeue/protopb/record.pb.go new file mode 100644 index 00000000..ca79d9cc --- /dev/null +++ b/stovepipe/core/messagequeue/protopb/record.pb.go @@ -0,0 +1,147 @@ +// 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. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.10 +// protoc v5.29.3 +// source: record.proto + +package protopb + +import ( + reflect "reflect" + sync "sync" + unsafe "unsafe" + + _ "github.com/uber/submitqueue/api/base/messagequeue/protopb" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// Record is the payload buildsignal publishes once a build reaches a terminal +// status: only the build id travels on the queue. record reloads the full +// Build from storage by this id (producer and consumer share the store, so +// the id is enough and redelivery stays idempotent). See +// doc/rfc/stovepipe/steps/buildsignal.md. +type Record struct { + state protoimpl.MessageState `protogen:"open.v1"` + // id is the terminal build's id: the runner-assigned id minted by + // BuildRunner.Trigger (entity.Build.ID / entity.BuildID.ID). + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Record) Reset() { + *x = Record{} + mi := &file_record_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Record) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Record) ProtoMessage() {} + +func (x *Record) ProtoReflect() protoreflect.Message { + mi := &file_record_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Record.ProtoReflect.Descriptor instead. +func (*Record) Descriptor() ([]byte, []int) { + return file_record_proto_rawDescGZIP(), []int{0} +} + +func (x *Record) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +var File_record_proto protoreflect.FileDescriptor + +const file_record_proto_rawDesc = "" + + "\n" + + "\frecord.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"$\n" + + "\x06Record\x12\x0e\n" + + "\x02id\x18\x01 \x01(\tR\x02id:\n" + + "\x8a\xb5\x18\x06recordB}\n" + + "+com.uber.submitqueue.stovepipe.messagequeueB\vRecordProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3" + +var ( + file_record_proto_rawDescOnce sync.Once + file_record_proto_rawDescData []byte +) + +func file_record_proto_rawDescGZIP() []byte { + file_record_proto_rawDescOnce.Do(func() { + file_record_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_record_proto_rawDesc), len(file_record_proto_rawDesc))) + }) + return file_record_proto_rawDescData +} + +var file_record_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_record_proto_goTypes = []any{ + (*Record)(nil), // 0: uber.stovepipe.messagequeue.Record +} +var file_record_proto_depIdxs = []int32{ + 0, // [0:0] is the sub-list for method output_type + 0, // [0:0] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_record_proto_init() } +func file_record_proto_init() { + if File_record_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_record_proto_rawDesc), len(file_record_proto_rawDesc)), + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_record_proto_goTypes, + DependencyIndexes: file_record_proto_depIdxs, + MessageInfos: file_record_proto_msgTypes, + }.Build() + File_record_proto = out.File + file_record_proto_goTypes = nil + file_record_proto_depIdxs = nil +} diff --git a/stovepipe/core/messagequeue/topics.go b/stovepipe/core/messagequeue/topics.go index cbc12387..0dac733d 100644 --- a/stovepipe/core/messagequeue/topics.go +++ b/stovepipe/core/messagequeue/topics.go @@ -41,4 +41,10 @@ const ( // Partitioned by build id, so each build's poll loop is an independent // partition. TopicKeyBuildSignal TopicKey = "buildsignal" + + // TopicKeyRecord carries a build's terminal status from buildsignal to the + // record stage. The buildsignal controller publishes a Record (the build + // id) here once, and only once, a build reaches a terminal status; + // non-terminal polls never publish here. Partitioned by request id. + TopicKeyRecord TopicKey = "record" ) diff --git a/tool/proto/BUILD.bazel b/tool/proto/BUILD.bazel index 9e3f55d7..5b9ebca1 100644 --- a/tool/proto/BUILD.bazel +++ b/tool/proto/BUILD.bazel @@ -70,6 +70,7 @@ go_proto_generated_files( "//stovepipe/core/messagequeue/proto:build.proto", "//stovepipe/core/messagequeue/proto:buildsignal.proto", "//stovepipe/core/messagequeue/proto:process.proto", + "//stovepipe/core/messagequeue/proto:record.proto", ], gen_services = False, imports = [ From bf03045d425f9d8028e304136c9cedf537d36807 Mon Sep 17 00:00:00 2001 From: "chenghan.ying" Date: Thu, 16 Jul 2026 23:18:04 +0000 Subject: [PATCH 2/5] wire buildsignal controller into stovepipe service --- service/stovepipe/server/main.go | 20 +++++++++----------- 1 file changed, 9 insertions(+), 11 deletions(-) diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 3bce151f..8df878af 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -255,16 +255,6 @@ func run() error { } logger.Info("controllers registered", zap.Int("primary", primaryCount), zap.Int("dlq", dlqCount)) - buildController := build.NewController(logger.Sugar(), scope, store, fakeBuildRunnerFactory{}, registry, stovepipemq.TopicKeyBuild, "stovepipe-build") - if err := primaryConsumer.Register(buildController); err != nil { - return fmt.Errorf("failed to register build controller: %w", err) - } - - buildSignalController := buildsignal.NewController(logger.Sugar(), scope, store, fakeBuildRunnerFactory{}, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal") - if err := primaryConsumer.Register(buildSignalController); err != nil { - return fmt.Errorf("failed to register buildsignal controller: %w", err) - } - // Start consumers. DLQ first because Start begins processing messages // immediately; if the primary consumer then fails to start, the half we // already started is the DLQ side, whose work is idempotent reconciliation @@ -393,6 +383,12 @@ func registerPrimaryControllers( } count++ + buildSignalController := buildsignal.NewController(logger, scope, store, brf, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal") + if err := c.Register(buildSignalController); err != nil { + return count, fmt.Errorf("failed to register buildsignal controller: %w", err) + } + count++ + return count, nil } @@ -439,10 +435,11 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Subscription: extqueue.DefaultSubscriptionConfig( subscriberName, "stovepipe-build", ), - }, { Key: stovepipemq.TopicKeyBuildSignal, + Name: "buildsignal", + Queue: q, Subscription: extqueue.DefaultSubscriptionConfig( subscriberName, "stovepipe-buildsignal", ), @@ -455,6 +452,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe { Key: dlq.TopicKey(stovepipemq.TopicKeyProcess), Name: "process_dlq", + Queue: q, Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-process-dlq"), }, }) From a9b7ef4686879d6c1e491950dd225c4b0d126071 Mon Sep 17 00:00:00 2001 From: "chenghan.ying" Date: Tue, 21 Jul 2026 21:55:19 +0000 Subject: [PATCH 3/5] fix(stovepipe): treat build/request not-found in buildsignal as non-retryable --- .../controller/buildsignal/buildsignal.go | 24 +++++++------------ .../buildsignal/buildsignal_test.go | 8 +++---- 2 files changed, 12 insertions(+), 20 deletions(-) diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go index bbe8e780..ab86cf68 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -192,30 +192,22 @@ func (c *Controller) reconcile(ctx context.Context, build entity.Build, status e return status, nil } -// loadBuild returns the build for id. A not-yet-visible row is retryable. +// loadBuild returns the build for id. func (c *Controller) loadBuild(ctx context.Context, id string) (entity.Build, error) { got, err := c.store.GetBuildStore().Get(ctx, id) - if err == nil { - return got, nil - } - if errors.Is(err, storage.ErrNotFound) { - return entity.Build{}, errs.NewRetryableError(fmt.Errorf("build %s not found yet: %w", id, err)) + if err != nil { + return entity.Build{}, fmt.Errorf("BuildSignalController failed to load build %s: %w", id, err) } - return entity.Build{}, fmt.Errorf("BuildSignalController failed to load build %s: %w", id, err) + return got, nil } -// loadRequest returns the request for id. A miss here is almost certainly a -// lagging read behind the Build that referenced it, so it is retryable; a -// genuinely orphaned Build still dead-letters at MaxAttempts. +// 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("BuildSignalController failed to load request %s: %w", id, err) } - return entity.Request{}, fmt.Errorf("BuildSignalController failed to load request %s: %w", id, err) + return got, nil } // pollDelay returns the delay before the next Status call for a non-terminal status. diff --git a/stovepipe/controller/buildsignal/buildsignal_test.go b/stovepipe/controller/buildsignal/buildsignal_test.go index 33aa02f0..4b14f02f 100644 --- a/stovepipe/controller/buildsignal/buildsignal_test.go +++ b/stovepipe/controller/buildsignal/buildsignal_test.go @@ -124,9 +124,9 @@ func TestProcess(t *testing.T) { wantRetry bool }{ { - name: "build not found is retryable", + name: "build not found is not retryable", wantErr: true, - wantRetry: true, + wantRetry: false, setup: func(m buildsignalMocks) { m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{}, storage.ErrNotFound) }, @@ -140,9 +140,9 @@ 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 buildsignalMocks) { m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusAccepted, 1), nil) m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, storage.ErrNotFound) From 073a6ddc457d01dfbe99831f50e8225c981db596 Mon Sep 17 00:00:00 2001 From: "chenghan.ying" Date: Tue, 21 Jul 2026 21:55:34 +0000 Subject: [PATCH 4/5] docs(stovepipe): update buildsignal RFC to match non-retryable not-found behavior --- doc/rfc/stovepipe/steps/buildsignal.md | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/doc/rfc/stovepipe/steps/buildsignal.md b/doc/rfc/stovepipe/steps/buildsignal.md index 8959d94c..74271a8b 100644 --- a/doc/rfc/stovepipe/steps/buildsignal.md +++ b/doc/rfc/stovepipe/steps/buildsignal.md @@ -20,14 +20,12 @@ For a delivery carrying build id `B`: ``` 1. Load Build B from the build store. - - ErrNotFound -> retryable (build's Create 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. Load Request R = store.Get(Build.RequestID) — needed for R.Queue to resolve the build-runner. - - ErrNotFound -> retryable, like step 1: the Build's existence proves the Request write is older, - so a miss here is almost certainly a lagging read; redelivery converges. A genuinely orphaned - Build (integrity fault) still dead-letters at MaxAttempts — the same terminal outcome, without - rejecting straight to DLQ on a stale read. + - ErrNotFound -> return raw; non-retryable, same as step 1 — the Build's existence proves the + Request write is already committed, so a miss here is a storage defect, not a lagging read. 3. If R.State is terminal (superseded / recorded-green / recorded-not-green): ack and return. - the Request is done (record already ran, or the head was superseded); stop polling. @@ -102,19 +100,19 @@ Per `platform/errs`'s non-retryable-by-default rule (see [platform/errs/README.m | Failure | Disposition | Why | |---|---|---| -| `Build` not found | retryable (`errs.NewRetryableError`) | `build`'s `Create` not visible yet; redelivery converges. | -| `Request` not found | retryable (`errs.NewRetryableError`) | The Build's existence proves the Request write is older, so a miss is a stale read; a genuine orphan still dead-letters at `MaxAttempts`. | | `Status` call | raw error; classifier decides | Deliberately left open rather than fixed either way — runner timeout/connection is transient, "runner not deployed for this queue" is not, and only a backend classifier can tell them apart. | | `Update` CAS conflict (`ErrVersionMismatch`) | retryable | A concurrent (redelivered) writer moved the row; reload and re-check converges. | | `PublishAfter` re-poll | retryable | The poll heartbeat; it runs only after status/persist/record all succeeded, so a transient enqueue blip is worth replaying to `MaxAttempts` before dead-lettering. | +`Build`/`Request` not found (`storage.ErrNotFound`) are **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, an `Update` store error other than a CAS conflict, and the publish to `record` — is returned raw with no override, because the default is already correct: a queue with no registered runner is a config error, and storage/queue failures dead-letter and let DLQ reconciliation recover. ## Idempotency Every branch is safe under at-least-once redelivery: -- **Build not found** — retryable; converges as the row becomes visible. +- **Build 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. - **Status already persisted** — a redelivery re-runs the whole algorithm from step 1, including a redundant `Status` poll (harmless — the runner reports the same thing); step 6 no-ops on the unchanged status, and the delivery proceeds to re-schedule the poll (non-terminal) or republish to `record` (terminal, idempotent). No corruption. - **Terminal already published** — a redelivery reloads, re-polls, no-ops at step 6, republishes the same terminal signal to `record` (idempotent), and acks. Harmless. - **`PublishAfter` failed, then retried** — the nacked delivery re-runs from step 1; there is no way to resume mid-algorithm, so it re-polls the runner too, but the row already carries the non-terminal status and step 6 no-ops. Only the final enqueue does new work. From d187bbc15082fbd57f72dadd517efee9cb71be3b Mon Sep 17 00:00:00 2001 From: "chenghan.ying" Date: Tue, 21 Jul 2026 22:09:58 +0000 Subject: [PATCH 5/5] refactor(stovepipe): share the load-entity-by-id pattern via stovepipe/core/loader --- stovepipe/controller/build/BUILD.bazel | 1 + stovepipe/controller/build/build.go | 7 +-- stovepipe/controller/buildsignal/BUILD.bazel | 1 + .../controller/buildsignal/buildsignal.go | 13 ++--- stovepipe/controller/process/BUILD.bazel | 1 + stovepipe/controller/process/process.go | 13 ++--- stovepipe/core/loader/BUILD.bazel | 18 +++++++ stovepipe/core/loader/loader.go | 46 ++++++++++++++++ stovepipe/core/loader/loader_test.go | 54 +++++++++++++++++++ 9 files changed, 129 insertions(+), 25 deletions(-) create mode 100644 stovepipe/core/loader/BUILD.bazel create mode 100644 stovepipe/core/loader/loader.go create mode 100644 stovepipe/core/loader/loader_test.go diff --git a/stovepipe/controller/build/BUILD.bazel b/stovepipe/controller/build/BUILD.bazel index be0c4736..bb5dd113 100644 --- a/stovepipe/controller/build/BUILD.bazel +++ b/stovepipe/controller/build/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/metrics:go_default_library", + "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/buildrunner:go_default_library", diff --git a/stovepipe/controller/build/build.go b/stovepipe/controller/build/build.go index 885bf9a4..9043b774 100644 --- a/stovepipe/controller/build/build.go +++ b/stovepipe/controller/build/build.go @@ -28,6 +28,7 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/buildrunner" @@ -149,11 +150,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r // 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 + return loader.ByID(ctx, id, c.store.GetRequestStore().Get, "BuildController", "request") } // publishBuildSignal publishes buildID to the buildsignal stage, partitioned by diff --git a/stovepipe/controller/buildsignal/BUILD.bazel b/stovepipe/controller/buildsignal/BUILD.bazel index cd9f5a38..9b33d891 100644 --- a/stovepipe/controller/buildsignal/BUILD.bazel +++ b/stovepipe/controller/buildsignal/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/metrics:go_default_library", + "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/buildrunner:go_default_library", diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go index ab86cf68..06d3ea3b 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -29,6 +29,7 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/buildrunner" @@ -194,20 +195,12 @@ func (c *Controller) reconcile(ctx context.Context, build entity.Build, status e // loadBuild returns the build for id. func (c *Controller) loadBuild(ctx context.Context, id string) (entity.Build, error) { - got, err := c.store.GetBuildStore().Get(ctx, id) - if err != nil { - return entity.Build{}, fmt.Errorf("BuildSignalController failed to load build %s: %w", id, err) - } - return got, nil + return loader.ByID(ctx, id, c.store.GetBuildStore().Get, "BuildSignalController", "build") } // 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("BuildSignalController failed to load request %s: %w", id, err) - } - return got, nil + return loader.ByID(ctx, id, c.store.GetRequestStore().Get, "BuildSignalController", "request") } // pollDelay returns the delay before the next Status call for a non-terminal status. diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index 2ff96fdd..907112ba 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/stovepipe/controller/process/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/metrics:go_default_library", + "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/queueconfig:go_default_library", diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 9cadb884..3fea751f 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -29,6 +29,7 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/queueconfig" @@ -455,20 +456,12 @@ func (c *Controller) rescheduleProcess(ctx context.Context, request entity.Reque // 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("ProcessController failed to load request %s: %w", id, err) - } - return got, nil + return loader.ByID(ctx, id, c.store.GetRequestStore().Get, "ProcessController", "request") } // 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 entity.Queue{}, fmt.Errorf("ProcessController failed to load queue %s: %w", name, err) - } - return got, nil + return loader.ByID(ctx, name, c.store.GetQueueStore().Get, "ProcessController", "queue") } // publishBuild publishes the admitted request ID to the build stage. The build diff --git a/stovepipe/core/loader/BUILD.bazel b/stovepipe/core/loader/BUILD.bazel new file mode 100644 index 00000000..4ab9ebfc --- /dev/null +++ b/stovepipe/core/loader/BUILD.bazel @@ -0,0 +1,18 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["loader.go"], + importpath = "github.com/uber/submitqueue/stovepipe/core/loader", + visibility = ["//visibility:public"], +) + +go_test( + name = "go_default_test", + srcs = ["loader_test.go"], + embed = [":go_default_library"], + deps = [ + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/stovepipe/core/loader/loader.go b/stovepipe/core/loader/loader.go new file mode 100644 index 00000000..d04fcfdd --- /dev/null +++ b/stovepipe/core/loader/loader.go @@ -0,0 +1,46 @@ +// 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 loader holds the "load one entity by id from a store" pattern +// shared by Stovepipe's queue controllers (build, process, buildsignal). +package loader + +import ( + "context" + "fmt" +) + +// ByID loads one entity by id via get, returning it unwrapped on success. On +// failure it wraps the error as " failed to load +// : " so every stovepipe controller reports load failures in the +// same shape. +// +// get is typically a store's Get method value passed directly (e.g. +// c.store.GetRequestStore().Get), which fixes T through inference so callers +// never spell out the type parameter. +// +// Not-found is never special-cased into a retryable error here: per the +// storage read-after-write-consistency contract (see +// stovepipe/extension/storage/README.md), a Get miss on a row that a +// causally-prior write should already have produced is a storage +// implementation defect, not a lag condition worth retrying through. It +// surfaces as a plain error, non-retryable by platform/errs's default. +func ByID[T any](ctx context.Context, id string, get func(context.Context, string) (T, error), controllerName, entityName string) (T, error) { + got, err := get(ctx, id) + if err != nil { + var zero T + return zero, fmt.Errorf("%s failed to load %s %s: %w", controllerName, entityName, id, err) + } + return got, nil +} diff --git a/stovepipe/core/loader/loader_test.go b/stovepipe/core/loader/loader_test.go new file mode 100644 index 00000000..00fe4fcb --- /dev/null +++ b/stovepipe/core/loader/loader_test.go @@ -0,0 +1,54 @@ +// 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 loader + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type widget struct { + Name string +} + +func TestByID(t *testing.T) { + t.Run("success returns the loaded value unwrapped", func(t *testing.T) { + get := func(_ context.Context, id string) (widget, error) { + return widget{Name: id}, nil + } + + got, err := ByID(context.Background(), "abc", get, "TestController", "widget") + + require.NoError(t, err) + assert.Equal(t, widget{Name: "abc"}, got) + }) + + t.Run("failure returns the zero value and a wrapped error", func(t *testing.T) { + sentinel := errors.New("boom") + get := func(_ context.Context, _ string) (widget, error) { + return widget{}, sentinel + } + + got, err := ByID(context.Background(), "abc", get, "TestController", "widget") + + require.Error(t, err) + assert.True(t, errors.Is(err, sentinel)) + assert.Equal(t, widget{}, got) + }) +}