diff --git a/submitqueue/orchestrator/controller/prioritize/prioritize.go b/submitqueue/orchestrator/controller/prioritize/prioritize.go index 29c5c815..6fe4ec25 100644 --- a/submitqueue/orchestrator/controller/prioritize/prioritize.go +++ b/submitqueue/orchestrator/controller/prioritize/prioritize.go @@ -392,7 +392,15 @@ func (c *Controller) republishBuilds(ctx context.Context, queue string, trees ma if !needsBuild { continue } - if err := c.publishBuild(ctx, batchID, queue); err != nil { + // The message ID encodes (batch, tree version): a round that changed + // the tree (new promotions) mints a fresh ID and is guaranteed + // delivery, while pure heal republishes reuse the last one and + // coalesce under the queue's publish idempotency on + // (topic, partition_key, id) — a bare batch ID would coalesce a + // later promotion round away against the first round's + // not-yet-collected row. + msgID := fmt.Sprintf("%s/v%d", batchID, tree.Version) + if err := c.publishBuild(ctx, msgID, batchID, queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish batch %s to build: %w", batchID, err) } @@ -404,14 +412,14 @@ func (c *Controller) republishBuilds(ctx context.Context, queue string, trees ma // publishBuild publishes a batch ID to the build topic. Only the identifier // travels on the queue; the build controller reloads the full Batch (and, // eventually, its speculation tree) from storage. -func (c *Controller) publishBuild(ctx context.Context, batchID, partitionKey string) error { +func (c *Controller) publishBuild(ctx context.Context, msgID, batchID, partitionKey string) error { bid := entity.BatchID{ID: batchID} payload, err := bid.ToBytes() if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil) + msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil) q, ok := c.registry.Queue(topickey.TopicKeyBuild) if !ok { diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index 2b5e3311..b6d4f68e 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -200,17 +200,19 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return c.cancelBatch(ctx, batch) } - // Terminal state: wake dependents and re-publish conclude. This branch is - // the dependent wake-up for every terminal transition — mergesignal routes - // merge outcomes back through speculate under the batch's own ID, the - // finalize pass below re-publishes a batch it fails, and re-publishing the - // dependents here lets each of them re-run its own dependency gate / - // finalize pass against the new dependency state. On redelivery the same - // branch doubles as the crash self-heal: both the dependent fan-out and - // the conclude publish are idempotent re-sends. + // Terminal state: wake dependents and re-publish conclude. Dependents are + // re-fanned for EVERY terminal state — each is a fact some dependent's + // tree needs to reconcile against: Succeeded rules the dep in + // (dead-pathing every sibling path that bet against it; mergesignal + // publishes the terminal batch here on success precisely so this branch + // runs), Failed and Cancelled rule it out. Failed additionally has no + // other stage covering it: a batch failed by finalize never passes + // through mergesignal's own fan-out. On redelivery the same branch + // doubles as the crash self-heal: both the dependent fan-out and the + // conclude publish are idempotent re-sends. if batch.State.IsTerminal() { metrics.NamedCounter(c.metricsScope, opName, "self_heal_terminal", 1) - if err := c.respeculateDependents(ctx, batch); err != nil { + if err := c.respeculateDependents(ctx, batch, "dep-terminal"); err != nil { return err } return c.fanout(ctx, batch.ID, batch.Queue) @@ -335,7 +337,14 @@ func (c *Controller) speculateBatch(ctx context.Context, batch entity.Batch) err // healed by redelivery — an error or crash here nacks the message, and // the next pass republishes, since this runs on every pass regardless of // whether anything changed. - if err := c.publishQueue(ctx, topickey.TopicKeyPrioritize, batch.Queue); err != nil { + // + // The message ID encodes (batch, tree version): every persisted tree + // change mints a fresh ID and so is guaranteed a round, while no-change + // passes reuse their last ID and coalesce under the queue's publish + // idempotency — every prioritizer input change is some tree's version + // bump, so coalescing the rest loses nothing. + roundID := fmt.Sprintf("%s/%s/v%d", batch.Queue, batch.ID, tree.Version) + if err := c.publishQueue(ctx, topickey.TopicKeyPrioritize, roundID, batch.Queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish queue %s to prioritize: %w", batch.Queue, err) } @@ -812,8 +821,10 @@ func (c *Controller) finalize(ctx context.Context, batch entity.Batch, tree enti // their merge readiness. Fan out AFTER the CAS so a woken dependent // reads this batch as Merging; a crash in between is healed by // superviseMerging, which re-runs the fan-out on every later - // delivery. - return c.respeculateDependents(ctx, batch) + // delivery. dep-merging (not dep-terminal): this wake announces a + // different event than the terminal fan-out and must not consume + // its one-per-(dependent, dep) message ID. + return c.respeculateDependents(ctx, batch, "dep-merging") } anyViable := false @@ -867,7 +878,7 @@ func (c *Controller) failBatch(ctx context.Context, batch entity.Batch) error { batch.Version = newVersion batch.State = entity.BatchStateFailed - if err := c.respeculateDependents(ctx, batch); err != nil { + if err := c.respeculateDependents(ctx, batch, "dep-terminal"); err != nil { return err } @@ -928,7 +939,7 @@ func (c *Controller) superviseMerging(ctx context.Context, batch entity.Batch) e metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to re-arm merge trigger for batch %s: %w", batch.ID, err) } - return c.respeculateDependents(ctx, batch) + return c.respeculateDependents(ctx, batch, "dep-merging") } } @@ -1005,7 +1016,7 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error // collateral requests need a fresh request ID and a re-publish to TopicKeyStart so // they can be re-batched without the cancelled change. - pending, err := c.cancelTree(ctx, batch) + pending, treeVersion, err := c.cancelTree(ctx, batch) if err != nil { return err } @@ -1014,13 +1025,17 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error // Builds are still winding down. The Cancelling intents are already // persisted; hand them to the build stage through the queue-wide // prioritize stage and wait for the next buildsignal wake to - // re-evaluate. + // re-evaluate. The round ID follows speculateBatch's scheme — one + // per persisted tree state — so a sweep that changed the tree is + // guaranteed a prioritize round while no-change re-publishes + // coalesce against it. metrics.NamedCounter(c.metricsScope, opName, "cancel_pending_builds", 1) c.logger.Debugw("waiting for builds to quiesce before terminal cancel", "batch_id", batch.ID, "pending_paths", pending, ) - if err := c.publishQueue(ctx, topickey.TopicKeyPrioritize, batch.Queue); err != nil { + roundID := fmt.Sprintf("%s/%s/v%d", batch.Queue, batch.ID, treeVersion) + if err := c.publishQueue(ctx, topickey.TopicKeyPrioritize, roundID, batch.Queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish queue %s to prioritize: %w", batch.Queue, err) } @@ -1039,7 +1054,7 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error "queue", batch.Queue, ) - if err := c.respeculateDependents(ctx, batch); err != nil { + if err := c.respeculateDependents(ctx, batch, "dep-terminal"); err != nil { return err } @@ -1053,7 +1068,9 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error // cancelTree sweeps the batch's speculation tree toward cancellation and // reports how many paths remain Cancelling — that is, whose builds have not -// yet been observed terminal. Terminal paths are settled outcomes and are +// yet been observed terminal — together with the tree's persisted version +// after the sweep (zero when the batch has no tree), which the caller folds +// into its prioritize round ID. Terminal paths are settled outcomes and are // skipped without I/O on ordinary passes (a pass about to settle re-checks // Cancelled paths once — see the terminal-transition guard below). Every // other path is resolved through the path->build mapping keyed by its @@ -1068,15 +1085,15 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error // sweep is persisted under the tree's version lock only when a status // actually moved, and costs one pass with at most two point reads per // unresolved path, mirroring reconcile. -func (c *Controller) cancelTree(ctx context.Context, batch entity.Batch) (int, error) { +func (c *Controller) cancelTree(ctx context.Context, batch entity.Batch) (int, int32, error) { tree, err := c.store.GetSpeculationTreeStore().Get(ctx, batch.ID) if err != nil { if errors.Is(err, storage.ErrNotFound) { metrics.NamedCounter(c.metricsScope, opName, "cancel_tree_not_found", 1) - return 0, nil + return 0, 0, nil } metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return 0, fmt.Errorf("failed to get speculation tree for batch %s: %w", batch.ID, err) + return 0, 0, fmt.Errorf("failed to get speculation tree for batch %s: %w", batch.ID, err) } paths := tree.Paths @@ -1094,7 +1111,7 @@ func (c *Controller) cancelTree(ctx context.Context, batch entity.Batch) (int, e status, err := c.cancelPathStatus(ctx, batch, p) if err != nil { - return 0, err + return 0, 0, err } if status == entity.SpeculationPathStatusCancelling { pending++ @@ -1124,7 +1141,7 @@ func (c *Controller) cancelTree(ctx context.Context, batch entity.Batch) (int, e } status, err := c.cancelPathStatus(ctx, batch, p) if err != nil { - return 0, err + return 0, 0, err } if status != entity.SpeculationPathStatusCancelling { continue @@ -1141,16 +1158,16 @@ func (c *Controller) cancelTree(ctx context.Context, batch entity.Batch) (int, e } if !changed { - return pending, nil + return pending, tree.Version, nil } newVersion := tree.Version + 1 if err := c.store.GetSpeculationTreeStore().Update(ctx, batch.ID, tree.Version, newVersion, paths); err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return 0, fmt.Errorf("failed to update speculation tree for batch %s: %w", batch.ID, err) + return 0, 0, fmt.Errorf("failed to update speculation tree for batch %s: %w", batch.ID, err) } metrics.NamedCounter(c.metricsScope, opName, "cancel_tree_updated", 1) - return pending, nil + return pending, newVersion, nil } // cancelPathStatus resolves the cancellation status for one non-terminal @@ -1229,7 +1246,7 @@ func (c *Controller) republishMergeTrigger(ctx context.Context, batch entity.Bat // Failed flow, and from the terminal branch in Process — the latter on both // the first pass after a batch goes terminal (the normal dependent wake-up) // and on redelivery (self-heal). -func (c *Controller) respeculateDependents(ctx context.Context, batch entity.Batch) error { +func (c *Controller) respeculateDependents(ctx context.Context, batch entity.Batch, reason string) error { bd, err := c.store.GetBatchDependentStore().Get(ctx, batch.ID) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) @@ -1244,7 +1261,18 @@ func (c *Controller) respeculateDependents(ctx context.Context, batch entity.Bat // the existing state-machine dispatch in Process all argue for the // publish. Revisit if the extra message hop ever shows up as latency // or cost. - if err := c.publish(ctx, topickey.TopicKeySpeculate, depID, batch.Queue); err != nil { + // + // The message ID marks this specific wake-up — "depID, because batch + // hit reason" — rather than reusing the dependent's bare batch ID, + // which other stages (score) have already published under: the + // queue's publish idempotency would coalesce the wake-up away against + // any such not-yet-collected row. One ID per (dependent, reason, dep) + // also makes the self-heal redeliveries of this fan-out coalesce + // among themselves, which is exactly the dedup wanted. reason keeps + // distinct events distinct: the optimistic dep-merging wake must not + // consume the ID of the later, load-bearing dep-terminal wake. + msgID := fmt.Sprintf("%s/%s/%s", depID, reason, batch.ID) + if err := c.publishAs(ctx, topickey.TopicKeySpeculate, msgID, depID, batch.Queue); err != nil { metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to publish dependent batch %s to speculate: %w", depID, err) } @@ -1281,15 +1309,29 @@ func (c *Controller) fanout(ctx context.Context, batchID, partitionKey string) e return nil } -// publish publishes a batch ID to the specified topic key. +// publish publishes a batch ID to the specified topic key, with the batch ID +// itself as the message ID. Appropriate for once-per-batch handoffs (merge, +// conclude): queue publishes are idempotent on (topic, partition_key, id), so +// reusing the batch ID makes accidental double-publishes coalesce. func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error { + return c.publishAs(ctx, key, batchID, batchID, partitionKey) +} + +// publishAs publishes a batch ID to the specified topic key under an explicit +// message ID. Re-trigger publishes (waking a dependent that has already been +// woken by other stages under other IDs) must NOT reuse an ID any other +// publisher uses for the same batch: the queue's publish idempotency on +// (topic, partition_key, id) silently coalesces a same-key publish against any +// row not yet garbage-collected — including already-consumed ones — so a +// reused ID can swallow the wake-up entirely. +func (c *Controller) publishAs(ctx context.Context, key consumer.TopicKey, msgID, batchID string, partitionKey string) error { bid := entity.BatchID{ID: batchID} payload, err := bid.ToBytes() if err != nil { return fmt.Errorf("failed to serialize batch ID: %w", err) } - msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil) + msg := entityqueue.NewMessage(msgID, payload, partitionKey, nil) q, ok := c.registry.Queue(key) if !ok { @@ -1308,17 +1350,21 @@ func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID return nil } -// publishQueue publishes a queue name to the specified topic key. Used for -// queue-scoped stages (prioritize) whose message payload carries no batch -// identity at all — just the queue to re-evaluate. -func (c *Controller) publishQueue(ctx context.Context, key consumer.TopicKey, queue string) error { +// publishQueue publishes a queue name to the specified topic key under the +// given message ID. Used for queue-scoped stages (prioritize) whose message +// payload carries no batch identity at all — just the queue to re-evaluate. +// The caller derives the message ID from what changed (see speculateBatch): +// a fixed ID (e.g. the bare queue name) would collide with the queue's +// publish idempotency on (topic, partition_key, id) and silently swallow +// every round after the first until garbage collection. +func (c *Controller) publishQueue(ctx context.Context, key consumer.TopicKey, msgID, queue string) error { qid := entity.QueueID{Name: queue} payload, err := qid.ToBytes() if err != nil { return fmt.Errorf("failed to serialize queue ID: %w", err) } - msg := entityqueue.NewMessage(queue, payload, queue, nil) + msg := entityqueue.NewMessage(msgID, payload, queue, nil) q, ok := c.registry.Queue(key) if !ok { diff --git a/submitqueue/orchestrator/controller/speculate/speculate_test.go b/submitqueue/orchestrator/controller/speculate/speculate_test.go index 1bdf6a68..6d9e26b8 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate_test.go +++ b/submitqueue/orchestrator/controller/speculate/speculate_test.go @@ -278,13 +278,16 @@ func TestController_Process_MergingWakesDependentsOnly(t *testing.T) { require.Len(t, *h.records, 2) assert.Equal(t, "submitqueue-merge", (*h.records)[0].topic) assert.True(t, strings.HasPrefix((*h.records)[0].msgID, batch.ID+"/mergeheal/")) - assert.Equal(t, pubRec{topic: "speculate", msgID: "test-queue/batch/2"}, (*h.records)[1]) + assert.Equal(t, pubRec{topic: "speculate", msgID: "test-queue/batch/2/dep-merging/" + batch.ID}, (*h.records)[1]) } -// Terminal states wake dependents (re-publish speculate for each) and then -// re-publish conclude — this is both the normal dependent wake-up (every -// terminal transition is routed back through speculate under the batch's own -// ID) and the crash self-heal for a lost publish. State must not change (no +// Terminal states wake dependents (re-publish speculate for each, under a +// dep-terminal-scoped message ID) and then re-publish conclude — each +// terminal state is a fact some dependent's tree needs to reconcile against: +// Succeeded's re-fan is how a dependent learns a dep landed (mergesignal +// publishes the terminal batch to speculate on success precisely so this +// branch runs), and Failed's re-fan covers a batch failed by finalize, which +// never passes through mergesignal's own fan-out. State must not change (no // UpdateState), and neither BuildStore nor SpeculationTreeStore is touched. func TestController_Process_TerminalSelfHeals(t *testing.T) { for _, state := range []entity.BatchState{ @@ -307,8 +310,8 @@ func TestController_Process_TerminalSelfHeals(t *testing.T) { require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) assert.Equal(t, []pubRec{ - {topic: "speculate", msgID: "test-queue/batch/2"}, - {topic: "speculate", msgID: "test-queue/batch/3"}, + {topic: "speculate", msgID: fmt.Sprintf("test-queue/batch/2/dep-terminal/%s", batch.ID)}, + {topic: "speculate", msgID: fmt.Sprintf("test-queue/batch/3/dep-terminal/%s", batch.ID)}, {topic: "conclude", msgID: batch.ID}, }, *h.records) }) @@ -386,7 +389,7 @@ func TestController_Process_CreatedEnumeratesScoresSelects(t *testing.T) { h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateSpeculating).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: fmt.Sprintf("%s/%s/v%d", batch.Queue, batch.ID, 2)}}, *h.records) } // TestController_Process_CreateTreeMintsPathIDs pins down the exact minted @@ -416,7 +419,7 @@ func TestController_Process_CreateTreeMintsPathIDs(t *testing.T) { h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateSpeculating).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: fmt.Sprintf("%s/%s/v1", batch.Queue, batch.ID)}}, *h.records) } // Create racing with a concurrent creator: ErrAlreadyExists must fall back to @@ -451,7 +454,7 @@ func TestController_Process_CreateTreeAlreadyExistsRereads(t *testing.T) { h.batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateSpeculating).Return(nil) require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: fmt.Sprintf("%s/%s/v%d", batch.Queue, batch.ID, 1)}}, *h.records) } // Dependency gate: too many active dependencies for a batch with no tree yet @@ -496,7 +499,7 @@ func TestController_Process_NoChangeSkipsTreeUpdate(t *testing.T) { // No treeStore.Update, no batchStore.UpdateState expected (already Speculating). require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: fmt.Sprintf("%s/%s/v%d", batch.Queue, batch.ID, 4)}}, *h.records) } // A version mismatch on the speculation tree Update must surface as an error @@ -579,11 +582,11 @@ func TestController_Process_MergeGate(t *testing.T) { require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) - wantRecords := []pubRec{{topic: "prioritize", msgID: "test-queue"}} + wantRecords := []pubRec{{topic: "prioritize", msgID: fmt.Sprintf("%s/%s/v%d", batch.Queue, batch.ID, 2)}} if tt.wantMerge { wantRecords = append(wantRecords, pubRec{topic: "submitqueue-merge", msgID: batch.ID}, - pubRec{topic: "speculate", msgID: "test-queue/batch/9"}, + pubRec{topic: "speculate", msgID: "test-queue/batch/9/dep-merging/" + batch.ID}, ) } assert.ElementsMatch(t, wantRecords, *h.records) @@ -619,7 +622,7 @@ func TestController_Process_CancelledBaseDepToleratedMerges(t *testing.T) { require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) assert.ElementsMatch(t, []pubRec{ - {topic: "prioritize", msgID: "test-queue"}, + {topic: "prioritize", msgID: fmt.Sprintf("%s/%s/v%d", batch.Queue, batch.ID, 1)}, {topic: "submitqueue-merge", msgID: batch.ID}, }, *h.records) } @@ -673,19 +676,20 @@ func TestController_Process_MergingSupervision(t *testing.T) { if tt.wantFail { assert.Equal(t, []pubRec{ - {topic: "speculate", msgID: "test-queue/batch/2"}, + {topic: "speculate", msgID: "test-queue/batch/2/dep-terminal/" + batch.ID}, {topic: "conclude", msgID: batch.ID}, }, *h.records) return } // A healthy wake re-arms the merge trigger (minted ID — never // the bare batch ID, which publish dedup could swallow) before - // waking dependents. + // waking dependents; the dep-merging wake keeps its own ID space + // so it cannot consume the later dep-terminal wake's. require.Len(t, *h.records, 2) rearm := (*h.records)[0] assert.Equal(t, "submitqueue-merge", rearm.topic) assert.True(t, strings.HasPrefix(rearm.msgID, batch.ID+"/mergeheal/"), "re-arm must mint a fresh ID, got %q", rearm.msgID) - assert.Equal(t, pubRec{topic: "speculate", msgID: "test-queue/batch/2"}, (*h.records)[1]) + assert.Equal(t, pubRec{topic: "speculate", msgID: "test-queue/batch/2/dep-merging/" + batch.ID}, (*h.records)[1]) }) } } @@ -738,7 +742,7 @@ func TestController_Process_FailedBaseDepDeadsPath(t *testing.T) { require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) assert.ElementsMatch(t, []pubRec{ - {topic: "prioritize", msgID: batch.Queue}, + {topic: "prioritize", msgID: fmt.Sprintf("%s/%s/v%d", batch.Queue, batch.ID, 3)}, {topic: "conclude", msgID: batch.ID}, }, *h.records) }) @@ -781,8 +785,8 @@ func TestController_Process_OwnBuildFailedFailsBatch(t *testing.T) { require.NoError(t, runProcess(t, ctrl, h.controller, batch.ID)) assert.ElementsMatch(t, []pubRec{ - {topic: "prioritize", msgID: batch.Queue}, - {topic: "speculate", msgID: "test-queue/batch/9"}, + {topic: "prioritize", msgID: fmt.Sprintf("%s/%s/v%d", batch.Queue, batch.ID, 6)}, + {topic: "speculate", msgID: fmt.Sprintf("test-queue/batch/9/dep-terminal/%s", batch.ID)}, {topic: "conclude", msgID: batch.ID}, }, *h.records) } @@ -817,7 +821,7 @@ func TestController_Process_FinalizeDependentPublishFailure(t *testing.T) { }, nil) require.Error(t, runProcess(t, ctrl, h.controller, batch.ID)) - assert.Equal(t, []pubRec{{topic: "prioritize", msgID: batch.Queue}}, *h.records) + assert.Equal(t, []pubRec{{topic: "prioritize", msgID: fmt.Sprintf("%s/%s/v%d", batch.Queue, batch.ID, 6)}}, *h.records) } // Hand-built two-path tree: a dependency of the head landing (Succeeded) @@ -880,13 +884,17 @@ func TestController_Process_CancellingFlow(t *testing.T) { name string // wantPending is true when the pass must leave the batch in // Cancelling and only publish prioritize; false when the pass - // completes the terminal sequence. - wantPending bool - setup func(t *testing.T, h *testHarness, batch entity.Batch) + // completes the terminal sequence. roundVersion is the tree version + // the pending pass's prioritize round ID must carry (the post-sweep + // persisted version). + wantPending bool + roundVersion int32 + setup func(t *testing.T, h *testHarness, batch entity.Batch) }{ { - name: "live_builds_marked_cancelling_and_waits", - wantPending: true, + name: "live_builds_marked_cancelling_and_waits", + wantPending: true, + roundVersion: 4, setup: func(t *testing.T, h *testHarness, batch entity.Batch) { pathA := entity.SpeculationPath{Head: batch.ID} pathB := entity.SpeculationPath{Base: []string{"test-queue/batch/dep"}, Head: batch.ID} @@ -922,8 +930,9 @@ func TestController_Process_CancellingFlow(t *testing.T) { }, }, { - name: "already_cancelling_live_build_waits_without_write", - wantPending: true, + name: "already_cancelling_live_build_waits_without_write", + wantPending: true, + roundVersion: 4, setup: func(t *testing.T, h *testHarness, batch entity.Batch) { path := entity.SpeculationPath{Head: batch.ID} pathID := "test-queue/batch/1/path/0" @@ -1012,8 +1021,9 @@ func TestController_Process_CancellingFlow(t *testing.T) { // trigger). A pass about to settle must pull it back to // Cancelling so the build stage enacts the stop, instead of // CASing the batch terminal over a running build. - name: "cancelled_path_with_live_build_reopened", - wantPending: true, + name: "cancelled_path_with_live_build_reopened", + wantPending: true, + roundVersion: 8, setup: func(t *testing.T, h *testHarness, batch entity.Batch) { path := entity.SpeculationPath{Head: batch.ID} pathID := "test-queue/batch/1/path/0" @@ -1114,13 +1124,13 @@ func TestController_Process_CancellingFlow(t *testing.T) { if tt.wantPending { assert.Equal(t, []pubRec{ - {topic: "prioritize", msgID: "test-queue"}, + {topic: "prioritize", msgID: fmt.Sprintf("test-queue/%s/v%d", batch.ID, tt.roundVersion)}, }, *h.records) return } assert.Equal(t, []pubRec{ - {topic: "speculate", msgID: "test-queue/batch/2"}, - {topic: "speculate", msgID: "test-queue/batch/3"}, + {topic: "speculate", msgID: fmt.Sprintf("test-queue/batch/2/dep-terminal/%s", batch.ID)}, + {topic: "speculate", msgID: fmt.Sprintf("test-queue/batch/3/dep-terminal/%s", batch.ID)}, {topic: "conclude", msgID: batch.ID}, }, *h.records) })