diff --git a/submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis.go b/submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis.go index 9d84b4ad3..91fa13d5e 100644 --- a/submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis.go +++ b/submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis.go @@ -220,8 +220,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er ) case batch.State == entity.BatchStateCreated: - // A prior attempt transitioned but lost its announcement, so - // re-announcing is the whole job. + // A prior attempt transitioned but lost its announcement. It may also + // have crashed before filing the Created record, without which the next + // batch's analysis cannot see this one as a dependency. + if err := corebatch.EnsureRecord(ctx, store, batch); err != nil { + metrics.NamedCounter(c.metricsScope, opName, "batch_store_errors", 1) + return err + } metrics.NamedCounter(c.metricsScope, opName, "reannounced", 1) default: diff --git a/submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis_test.go b/submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis_test.go index d16c07ec2..b654f881a 100644 --- a/submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis_test.go +++ b/submitqueue/orchestrator/controller/dependencyanalysis/dependencyanalysis_test.go @@ -302,7 +302,8 @@ func TestController_Process_DedupesAnalyzerConflicts(t *testing.T) { } // A redelivery whose previous attempt transitioned but lost its announcement: -// re-announcing is the whole job, and re-analyzing would corrupt the graph. +// it repairs the Created record and re-announces, and re-analyzing would +// corrupt the graph. func TestController_Process_RedeliveryAfterTransitionOnlyRepublishes(t *testing.T) { ctrl := gomock.NewController(t) @@ -314,9 +315,17 @@ func TestController_Process_RedeliveryAfterTransitionOnlyRepublishes(t *testing. batchStore := storagemock.NewMockBatchStore(ctrl) batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil) + queueBatchStateStore := storagemock.NewMockQueueBatchStateStore(ctrl) + queueBatchStateStore.EXPECT().Put(gomock.Any(), entity.QueueBatchState{ + Queue: testQueue, + State: entity.BatchStateCreated, + BatchID: batch.ID, + }).Return(nil) + // Dependent store and request store with no EXPECTs — must not be touched. store := orchstoragemock.NewMockStorage(ctrl) store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetQueueBatchStateStore().Return(queueBatchStateStore).AnyTimes() store.EXPECT().GetBatchDependentStore().Return(storagemock.NewMockBatchDependentStore(ctrl)).AnyTimes() store.EXPECT().GetRequestStore().Return(storagemock.NewMockRequestStore(ctrl)).AnyTimes() store.EXPECT().GetRequestBatchStore().Return(noPriorEnrollment(ctrl)).AnyTimes() diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go index bcdadacb9..ed955125a 100644 --- a/submitqueue/orchestrator/pipeline.go +++ b/submitqueue/orchestrator/pipeline.go @@ -95,11 +95,11 @@ type Deps struct { // // Pipeline: // -// start → cancel → validate ⇢ (runway) ⇢ landconflictsignal → batch → speculate → build → buildsignal ─┐ -// ↑ ↘ ↻ poll │ -// │ land → conclude │ -// │ │ │ -// └─────┴───────────────────────┘ +// start → cancel → validate ⇢ (runway) ⇢ landconflictsignal → batch → dependency-analysis → speculate → build → buildsignal ─┐ +// ↑ ↘ ↻ poll │ +// │ land → conclude │ +// │ │ │ +// └─────┴───────────────────────┘ var Stages = []pipeline.Stage[Deps]{ { Key: topickey.TopicKeyStart,