feat(orchestrator): migrate the buildsignal poll loop to the hold primitive

## 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`.

Author: behinddwalls

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,30 +88,21 @@ 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:
 
 - **No DLQ escape valve.** A malformed `buildID`, or a build the provider has lost, fails `Status` every call. With unbounded retries the message spins forever; the operator gets no signal that something is permanently wrong. DLQ exists for exactly this case; opting out for the buildsignal subscription means opting out of every poison-message signal it offers.
 - **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
 

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"

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)

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, Update, 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"
 
@@ -187,14 +185,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, updatedBuild, 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", updatedBuild.ID,
 		"status", string(status),
 		"delay_ms", delayMs,
@@ -213,34 +210,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}

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
@@ -150,7 +151,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)
@@ -159,8 +160,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
@@ -185,17 +186,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().Update(gomock.Any(), updatedBuild).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)
 		})
 	}
@@ -210,7 +205,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 Update, no Publish, no PublishAfter expected.
+	// No Update, no Publish, no Hold expected.
 
 	err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build))
 	require.Error(t, err)
@@ -231,38 +226,14 @@ func TestController_Process_UpdateError(t *testing.T) {
 	h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil)
 	h.buildStore.EXPECT().Update(gomock.Any(), updatedBuild).
 		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)
 	// Non-retryable: rejects to DLQ on first failure; republish is the recovery path.
 	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}
-	updatedBuild := build
-	updatedBuild.Status = entity.BuildStatusRunning
-
-	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().Update(gomock.Any(), updatedBuild).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.
@@ -315,9 +286,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 Update, no speculate Publish, no buildsignal
-			// PublishAfter. The harness publishers have no expectations, so any
-			// publish fails the test.
+			// Halted: no Update, 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)))
 		})