Skip to content

Commit dc622e6

Browse files
authored
fix(orchestrator): repair Created record when dependency-analysis re-announces (#875)
## Summary ### Why? `corebatch.Transition` CASes the batch before filing its per-state membership record, and the transition protocol relies on the already-in-target-state retry calling `EnsureRecord` to repair a crash between the two. The dependency-analysis re-announce branch (batch already `Created`) skipped it, so a crash there left a `Created` batch invisible to `ListByStates` — the next batch's analysis would not serialize behind it until speculate happened to admit it. ### What? - Call `corebatch.EnsureRecord` in the `Created` re-announce branch before re-publishing to speculate. - Update the pipeline diagram in the `Stages` doc to include the `dependency-analysis` stage. ## Test Plan ✅ `bazel test //submitqueue/orchestrator/controller/dependencyanalysis/...` (re-announce test now requires the `Created` record put) ✅ `make lint` ## Stack 1. @ #875 1. #876 🤖 Generated with [Claude Code](https://claude.com/claude-code)
1 parent 0707f58 commit dc622e6

3 files changed

Lines changed: 22 additions & 8 deletions

File tree

‎submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis.go‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -220,8 +220,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
220220
)
221221

222222
case batch.State == entity.BatchStateCreated:
223-
// A prior attempt transitioned but lost its announcement, so
224-
// re-announcing is the whole job.
223+
// A prior attempt transitioned but lost its announcement. It may also
224+
// have crashed before filing the Created record, without which the next
225+
// batch's analysis cannot see this one as a dependency.
226+
if err := corebatch.EnsureRecord(ctx, store, batch); err != nil {
227+
metrics.NamedCounter(c.metricsScope, opName, "batch_store_errors", 1)
228+
return err
229+
}
225230
metrics.NamedCounter(c.metricsScope, opName, "reannounced", 1)
226231

227232
default:

‎submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis_test.go‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -302,7 +302,8 @@ func TestController_Process_DedupesAnalyzerConflicts(t *testing.T) {
302302
}
303303

304304
// A redelivery whose previous attempt transitioned but lost its announcement:
305-
// re-announcing is the whole job, and re-analyzing would corrupt the graph.
305+
// it repairs the Created record and re-announces, and re-analyzing would
306+
// corrupt the graph.
306307
func TestController_Process_RedeliveryAfterTransitionOnlyRepublishes(t *testing.T) {
307308
ctrl := gomock.NewController(t)
308309

@@ -314,9 +315,17 @@ func TestController_Process_RedeliveryAfterTransitionOnlyRepublishes(t *testing.
314315
batchStore := storagemock.NewMockBatchStore(ctrl)
315316
batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil)
316317

318+
queueBatchStateStore := storagemock.NewMockQueueBatchStateStore(ctrl)
319+
queueBatchStateStore.EXPECT().Put(gomock.Any(), entity.QueueBatchState{
320+
Queue: testQueue,
321+
State: entity.BatchStateCreated,
322+
BatchID: batch.ID,
323+
}).Return(nil)
324+
317325
// Dependent store and request store with no EXPECTs — must not be touched.
318326
store := orchstoragemock.NewMockStorage(ctrl)
319327
store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
328+
store.EXPECT().GetQueueBatchStateStore().Return(queueBatchStateStore).AnyTimes()
320329
store.EXPECT().GetBatchDependentStore().Return(storagemock.NewMockBatchDependentStore(ctrl)).AnyTimes()
321330
store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes()
322331
store.EXPECT().GetRequestBatchStore().Return(noPriorEnrollment(ctrl)).AnyTimes()

‎submitqueue/orchestrator/pipeline.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -95,11 +95,11 @@ type Deps struct {
9595
//
9696
// Pipeline:
9797
//
98-
// start → cancel → validate ⇢ (runway) ⇢ landconflictsignal → batch → speculate → build → buildsignal ─┐
99-
// ↑ ↘ ↻ poll │
100-
// │ land → conclude │
101-
// │ │ │
102-
// └─────┴───────────────────────┘
98+
// start → cancel → validate ⇢ (runway) ⇢ landconflictsignal → batch → dependency-analysis → speculate → build → buildsignal ─┐
99+
// ↑ ↘ ↻ poll │
100+
// │ land → conclude │
101+
// │ │ │
102+
// └─────┴───────────────────────┘
103103
var Stages = []pipeline.Stage[Deps]{
104104
{
105105
Key: topickey.TopicKeyStart,

0 commit comments

Comments
 (0)