From 93f5ace414830b3b6906b0d00629eff4826cae31 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Sat, 1 Aug 2026 10:51:24 -0700 Subject: [PATCH] feat(orchestrator): migrate the buildsignal poll loop to the hold primitive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The orchestrator's buildsignal stage carried the same ack-and-republish poll loop the stovepipe stages just migrated off (previous commit): each non-terminal poll acked the delivery and PublishAfter'd the build id back to its own topic. Notably its republish reused the build id as the message id — the exact dedup collision that stalled stovepipe's loop (#465) was latent here — and a transient publish failure was wrapped retryable as the loop's only liveness, the awkward classification #469 removed on the stovepipe side. ### What? A non-terminal status now records a hold for the per-status poll delay and returns success; the framework postpones the delivery, which redelivers without counting toward the retry limit. publishBuild is deleted (the re-poll was its only caller); the speculate publish and halted-batch short-circuit are unchanged, and the DLQ reconciler is unaffected. The Process retryability comment is rewritten: the loop's continuation is framework-owned, so no publish needs a retryable wrap. build-runner.md's "Polling primitive" section is updated — hold supersedes the PublishAfter design it argued for, keeping the same retry_count semantics with no publisher, no minted ids, and no per-tick rows. ## Test Plan ✅ `bazel test //submitqueue/...` — NonTerminal cases assert Hold(per-status delay); Terminal/StatusError/UpdateStatusError/Halted cases fail on any Hold; the RepublishError test is deleted (a hold cannot fail). ✅ `make fmt`. --- doc/rfc/submitqueue/build-runner.md | 25 +++---- submitqueue/core/topickey/topickey.go | 4 +- .../orchestrator/controller/build/build.go | 4 +- .../controller/buildsignal/buildsignal.go | 59 ++++------------- .../buildsignal/buildsignal_test.go | 65 ++++++------------- 5 files changed, 45 insertions(+), 112 deletions(-) diff --git a/doc/rfc/submitqueue/build-runner.md b/doc/rfc/submitqueue/build-runner.md index d33b29bf..73e33b80 100644 --- a/doc/rfc/submitqueue/build-runner.md +++ b/doc/rfc/submitqueue/build-runner.md @@ -10,7 +10,7 @@ The build stage needs a vendor-agnostic abstraction for talking to a Build Runne ## Flow -`build` triggers the runner and hands the `buildID` to the `buildsignal` poll loop. The loop calls `Status` on its own partition per build until the build is terminal: terminal results wake the batch state machine via `speculate`; non-terminal results re-enqueue the same `buildID` after a delay (`PublishAfter`). A webhook-capable backend can publish a status message into the same queue — the consumer cannot tell a push from a poll. +`build` triggers the runner and hands the `buildID` to the `buildsignal` poll loop. The loop calls `Status` on its own partition per build until the build is terminal: terminal results wake the batch state machine via `speculate`; non-terminal results hold the delivery, so the same message redelivers after a delay. A webhook-capable backend can publish a status message into the same queue — the consumer cannot tell a push from a poll. ``` ┌────────────────────────────────────────────────────┐ @@ -29,8 +29,8 @@ The build stage needs a vendor-agnostic abstraction for talking to a Build Runne ▼ ▼ ┌───────────────┐ ┌──────────────────────────────────┐ │ terminal │ │ non-terminal │ - │ → speculate │ │ → PublishAfter(buildID, delay) │ - │ re-evaluate │ │ re-enqueues to buildsignal │ + │ → speculate │ │ → hold(delay) │ + │ re-evaluate │ │ same message redelivers │ └───────────────┘ └──────────────────────────────────┘ ``` @@ -88,20 +88,13 @@ This makes polling behave like everything else in the orchestrator: - **Independent partitions** — slow polls on one build don't block others. - **Restart-safe** — pending polls live in the queue, not in memory. - **Retry-native** — a `Status` call that errors out is `Nack`'d and redelivered with the queue's normal backoff, separate from polling cadence. -- **Tunable cadence** — re-publish delay can vary by status (longer for `Accepted`, shorter for `Running`). +- **Tunable cadence** — the hold delay can vary by status (longer for `Accepted`, shorter for `Running`). -### Polling primitive: `PublishAfter`, not `Nack` +### Polling primitive: hold, not `Nack` -Postponing the next poll needs a "publish-with-delay" verb. Two candidates exist in or near the queue extension: +Postponing the next poll needs a "check back later" verb, and the consumer framework provides one: the controller records a hold on its delivery and returns success, and the framework postpones the message — it redelivers after the delay, and the redelivery is exempt from `retry_count` accounting (see [consumer-hold.md](../consumer-hold.md)). `Nack` remains the primitive for genuine `Status` failures, with its normal bounded-retry-then-DLQ behaviour. The two signals stay separate: `retry_count` means "consecutive failures," never "polls so far." -- **`Publisher.PublishAfter(topic, msg, delayMs)`** — a new primitive. A fresh message, made visible only after `delayMs`. The SQL-backed queue already has the column needed (`invisible_until`); `PublishAfter` is `Publish` with a non-zero delay. -- **`Delivery.Nack(requeueAfterMs)`** — the existing primitive. Re-uses the same message, sets it invisible until `now + delay`, increments `retry_count`. - -Both deliver the same surface behaviour: one message per build at a time, redelivered after the chosen delay. The difference is what `retry_count` means. - -`Nack` is "this delivery failed, try again," and `retry_count` feeds `MaxAttempts` and DLQ. Using it for "build not yet done" overloads that counter — every poll bumps a number that is supposed to flag problems. - -`PublishAfter` is "postpone this work." Each poll cycle is a fresh message with `retry_count = 0`. `Nack` stays available for true `Status` failures with its normal bounded-retry-then-DLQ behaviour. The two signals stay separate. +An earlier revision of this design reached the same separation with `Publisher.PublishAfter` — ack the delivery, publish a fresh copy of the same message with delayed visibility. Hold supersedes it: no publisher in the poll loop, no fresh message ids to mint around the queue's publish dedup, no new log row per tick, and the loop's continuation no longer depends on an enqueue succeeding (a failed postpone write lapses into a normal visibility-timeout redelivery). **Why not `Nack` with `MaxAttempts = ∞`** (one message per build, just keep cycling)? The mechanism works. Three things break: @@ -109,9 +102,7 @@ Both deliver the same surface behaviour: one message per build at a time, redeli - **Conflated metric.** `retry_count` is the obvious dashboard signal for "this consumer is having trouble." With infinite-retry polling, a `retry_count` of 500 might mean "build has been running 30 minutes" *or* "Status has errored 500 times" — operationally indistinguishable. - **Visibility-timeout coupling.** If the consumer crashes mid-poll before its `Nack`, the queue's visibility timeout redelivers the message and bumps `retry_count`. One number ends up counting legitimate polls, real errors, *and* consumer crashes — three signals fused. -`PublishAfter` costs one new queue primitive. It buys back the queue's diagnostic semantics. - -Trade-off acknowledged: `PublishAfter` writes more — Ack deletes the old message, PublishAfter inserts a new one — vs `Nack` updating one row in place. At minute cadence the difference is noise; at second cadence it is real but small. +Hold keeps all three signals separate for free: a deliberate postpone resets the failure streak, so `retry_count` counts only consecutive genuine failures. ### Push, when a backend supports it diff --git a/submitqueue/core/topickey/topickey.go b/submitqueue/core/topickey/topickey.go index 8255c2cf..28ff0f5b 100644 --- a/submitqueue/core/topickey/topickey.go +++ b/submitqueue/core/topickey/topickey.go @@ -36,8 +36,8 @@ const ( // TopicKeyBuildSignal is the polling stage for triggered builds. Each // message carries a Build; the consumer calls BuildRunner.Status, // persists the latest status, publishes the batch ID to TopicKeySpeculate - // so the state machine re-evaluates, and re-publishes itself via - // PublishAfter when the build has not yet reached a terminal state. + // so the state machine re-evaluates, and holds the delivery for the next + // poll when the build has not yet reached a terminal state. TopicKeyBuildSignal TopicKey = "buildsignal" // TopicKeyMerge is the pipeline stage where speculated batches are published for merging. TopicKeyMerge TopicKey = "submitqueue-merge" diff --git a/submitqueue/orchestrator/controller/build/build.go b/submitqueue/orchestrator/controller/build/build.go index a2fc396e..6b1b30d0 100644 --- a/submitqueue/orchestrator/controller/build/build.go +++ b/submitqueue/orchestrator/controller/build/build.go @@ -149,8 +149,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er } // Hand off to the buildsignal poll loop; it calls Status, updates the - // persisted Build, publishes to speculate, and re-publishes itself via - // PublishAfter until terminal. + // persisted Build, publishes to speculate, and holds its delivery + // between polls until terminal. if err := c.publish(ctx, topickey.TopicKeyBuildSignal, build); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish to buildsignal: %w", err) diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go index f2064164..778c8863 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go @@ -15,11 +15,11 @@ // Package buildsignal implements the build poll loop. Each message carries // a Build; the controller calls BuildRunner.Status, writes the latest // status to the BuildStore, publishes the batch ID to TopicKeySpeculate -// so the state machine re-evaluates, and re-publishes itself via -// PublishAfter when the build has not yet reached a terminal state. Each -// buildID partitions independently, so slow polls on one build do not -// block others. A webhook-capable backend can publish into this same -// topic — the controller cannot tell a poll-driven message from a push. +// so the state machine re-evaluates, and holds the delivery for the next +// poll when the build has not yet reached a terminal state. Each message +// partitions by batch ID, so slow polls on one batch's build do not block +// others. A webhook-capable backend can publish into this same topic — the +// controller cannot tell a poll-driven message from a push. package buildsignal import ( @@ -89,18 +89,16 @@ func NewController( } // Process polls the build's current status, persists it, publishes the -// batch ID to speculate so the state machine re-evaluates, and re-publishes -// a delayed message back to this topic when the build is still in flight. +// batch ID to speculate so the state machine re-evaluates, and holds the +// delivery for the next poll when the build is still in flight. // Returns nil to ack (success), or error to nack/reject. // // Error classification: deserialize, Status, UpdateStatus, and the speculate // publish stay non-retryable — they reject straight to DLQ on the first // failure, where the operational republish path is the recovery mechanism. -// Only the PublishAfter self-reschedule is retryable: it is the poll loop's -// heartbeat and runs only after status/persist/speculate have all succeeded, -// so a transient enqueue blip nacks and replays (up to MaxAttempts) rather -// than silently stalling the build, then still falls through to DLQ if it -// persists. +// The poll loop's continuation is a hold, not a publish: the framework +// postpones the delivery, and a failed postpone write lapses into a normal +// visibility-timeout redelivery, so the loop cannot stall on an enqueue. func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" @@ -186,14 +184,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er return nil } + // Not terminal yet: hold the delivery so this same message redelivers + // after the poll delay, without counting toward the retry limit. delayMs := pollDelay(status) metrics.NamedCounter(c.metricsScope, opName, "rescheduled", 1, metrics.NewTag("status", string(status))) - if err := c.publishBuild(ctx, c.topicKey, build, delayMs); err != nil { - metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) - return fmt.Errorf("failed to re-publish to buildsignal: %w", err) - } + delivery.Hold(delayMs) - c.logger.Debugw("rescheduled build status poll", + c.logger.Debugw("holding for next build status poll", "build_id", build.ID, "status", string(status), "delay_ms", delayMs, @@ -212,34 +209,6 @@ func pollDelay(status entity.BuildStatus) int64 { } } -// publishBuild publishes a build's ID to the topic identified by key. delayMs > 0 -// uses PublishAfter; otherwise it uses Publish. Only the identifier travels on -// the queue — the consumer reloads the full Build from storage. -func (c *Controller) publishBuild(ctx context.Context, key consumer.TopicKey, build entity.Build, delayMs int64) error { - payload, err := entity.BuildID{ID: build.ID}.ToBytes() - if err != nil { - return fmt.Errorf("failed to serialize build ID: %w", err) - } - - msg := entityqueue.NewMessage(build.ID, payload, build.BatchID, nil) - - 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) - } - - publisher := q.Publisher() - if delayMs > 0 { - return publisher.PublishAfter(ctx, topicName, msg, delayMs) - } - return publisher.Publish(ctx, topicName, msg) -} - // publishBatchID publishes a batch ID to the topic identified by key. func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error { bid := entity.BatchID{ID: batchID} diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go index b1a144e0..1dc5628e 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal_test.go @@ -36,8 +36,8 @@ import ( ) // testHarness wires a Controller against mock queues for two topic keys -// (buildsignal and speculate) so individual tests can assert which -// Publish / PublishAfter happens. +// (buildsignal and speculate) so individual tests can assert which publish +// or hold happens. type testHarness struct { controller *Controller br *buildrunnermock.MockBuildRunner @@ -93,8 +93,9 @@ func newTestHarness(t *testing.T, ctrl *gomock.Controller) *testHarness { // buildDelivery builds a delivery whose payload is the build's ID, matching // the on-queue contract: only the identifier travels, the consumer loads the -// full Build from storage. -func buildDelivery(t *testing.T, ctrl *gomock.Controller, b entity.Build) consumer.Delivery { +// full Build from storage. Tests that expect a hold add the expectation on +// the returned mock. +func buildDelivery(t *testing.T, ctrl *gomock.Controller, b entity.Build) *consumermock.MockDelivery { t.Helper() payload, err := entity.BuildID{ID: b.ID}.ToBytes() require.NoError(t, err) @@ -117,8 +118,8 @@ func TestController_Identity(t *testing.T) { } // TestController_Process_Terminal verifies a terminal poll persists the -// status, publishes the batch ID to speculate, and does NOT re-publish to -// buildsignal. +// status, publishes the batch ID to speculate, and does NOT hold the +// delivery for another poll. func TestController_Process_Terminal(t *testing.T) { tests := []struct { name string @@ -148,7 +149,7 @@ func TestController_Process_Terminal(t *testing.T) { assert.Equal(t, build.BatchID, bid.ID) return nil }).Times(1) - // No PublishAfter expected on terminal. + // No Hold expected on terminal — any Hold call fails the test. err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) require.NoError(t, err) @@ -157,8 +158,8 @@ func TestController_Process_Terminal(t *testing.T) { } // TestController_Process_NonTerminal verifies a non-terminal poll persists -// the status, publishes to speculate, AND re-publishes to buildsignal via -// PublishAfter with the per-status delay. +// the status, publishes to speculate, AND holds the delivery for the next +// poll with the per-status delay. func TestController_Process_NonTerminal(t *testing.T) { tests := []struct { name string @@ -181,17 +182,11 @@ func TestController_Process_NonTerminal(t *testing.T) { h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil) h.buildStore.EXPECT().UpdateStatus(gomock.Any(), build.ID, tt.status).Return(nil) h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil).Times(1) - h.signalPub.EXPECT(). - PublishAfter(gomock.Any(), "buildsignal", gomock.AssignableToTypeOf(entityqueue.Message{}), tt.wantDelayMs). - DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message, _ int64) error { - bid, err := entity.BuildIDFromBytes(msg.Payload) - require.NoError(t, err) - // Re-published payload carries only the build ID. - assert.Equal(t, build.ID, bid.ID) - return nil - }).Times(1) - err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) + d := buildDelivery(t, ctrl, build) + d.EXPECT().Hold(tt.wantDelayMs) + + err := h.controller.Process(context.Background(), d) require.NoError(t, err) }) } @@ -206,7 +201,7 @@ func TestController_Process_StatusError(t *testing.T) { h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil) h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil) h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusUnknown, nil, errors.New("provider down")) - // No UpdateStatus, no Publish, no PublishAfter expected. + // No UpdateStatus, no Publish, no Hold expected. err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) require.Error(t, err) @@ -225,7 +220,7 @@ func TestController_Process_UpdateStatusError(t *testing.T) { h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil) h.buildStore.EXPECT().UpdateStatus(gomock.Any(), build.ID, entity.BuildStatusRunning). Return(errors.New("db unreachable")) - // No Publish / PublishAfter expected after the store failure. + // No Publish / Hold expected after the store failure. err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) require.Error(t, err) @@ -233,28 +228,6 @@ func TestController_Process_UpdateStatusError(t *testing.T) { assert.False(t, errs.IsRetryable(err)) } -// TestController_Process_RepublishError verifies that a failure to re-publish -// the delayed poll message surfaces an error. The preceding -// status/persist/speculate steps all succeed. -func TestController_Process_RepublishError(t *testing.T) { - ctrl := gomock.NewController(t) - h := newTestHarness(t, ctrl) - - build := entity.Build{ID: "b-5", BatchID: "batch-5", Status: entity.BuildStatusAccepted} - - h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil) - h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusRunning, entity.BuildMetadata{}, nil) - h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil) - h.buildStore.EXPECT().UpdateStatus(gomock.Any(), build.ID, entity.BuildStatusRunning).Return(nil) - h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil).Times(1) - h.signalPub.EXPECT(). - PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs). - Return(errors.New("queue unavailable")).Times(1) - - err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build)) - require.Error(t, err) -} - // TestController_Process_GetError verifies that a failure to load the Build // from storage (only the ID is on the queue) surfaces an error. Non-retryable: // it rejects to DLQ on first failure, consistent with other storage reads. @@ -307,9 +280,9 @@ func TestController_Process_HaltedShortCircuit(t *testing.T) { h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil) h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusRunning, entity.BuildMetadata{}, nil) h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: state}, nil) - // Halted: no UpdateStatus, no speculate Publish, no buildsignal - // PublishAfter. The harness publishers have no expectations, so any - // publish fails the test. + // Halted: no UpdateStatus, no speculate Publish, no Hold. The + // harness publishers have no expectations, so any publish fails + // the test. require.NoError(t, h.controller.Process(context.Background(), buildDelivery(t, ctrl, build))) })