fix(orchestrator): repair Created record when dependency-analysis re-announces

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:

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()

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,