feat(orchestrator): support creating batches

Add a pre-processing batch state and make downstream readers safely tolerate batches whose reverse indexes are still being initialized.

Jira Issues: CODEM-304

Author: ubettigole

submitqueue/entity/batch.go

@@ -22,7 +22,10 @@ type BatchState string
 const (
 	// BatchStateUnknown is the unreachable state. It is set by default when the structure is initialized. It should never be seen in the system.
 	BatchStateUnknown BatchState = ""
-	// BatchStateCreated is the state of a batch that has been created for processing.
+	// BatchStateCreating indicates that the batch has been persisted but its dependency reverse indexes may not yet be fully initialized.
+	// A Creating batch is not eligible to be referenced as a dependency.
+	BatchStateCreating BatchState = "creating"
+	// BatchStateCreated indicates that the batch and its dependency reverse indexes are fully initialized and ready for processing.
 	BatchStateCreated BatchState = "created"
 	// BatchStateSpeculating is the state of a batch that is undergoing speculative execution.
 	BatchStateSpeculating BatchState = "speculating"
@@ -72,13 +75,30 @@ func IsBatchStateHalted(s BatchState) bool {
 // batches that cancel redelivery must be able to resolve.
 func ActiveBatchStates() []BatchState {
 	return []BatchState{
+		BatchStateCreating,
 		BatchStateCreated,
 		BatchStateSpeculating,
 		BatchStateMerging,
 		BatchStateCancelling,
 	}
 }
 
+// AllBatchStates returns every persisted batch lifecycle state.
+// Use this only for bounded recovery paths that must find historical batches.
+// Normal processing should query the narrower active or dependency state sets.
+func AllBatchStates() []BatchState {
+	return []BatchState{
+		BatchStateCreating,
+		BatchStateCreated,
+		BatchStateSpeculating,
+		BatchStateMerging,
+		BatchStateSucceeded,
+		BatchStateFailed,
+		BatchStateCancelling,
+		BatchStateCancelled,
+	}
+}
+
 // DependencyBatchStates returns the batch states that make an in-flight batch eligible
 // to be a dependency of a newly created batch. When a batch is created, the conflict
 // analyzer picks the existing batches it conflicts with as its dependencies; the new

submitqueue/entity/batch_test.go

@@ -28,6 +28,7 @@ func TestBatchState_IsTerminal(t *testing.T) {
 		terminal bool
 	}{
 		{name: "unknown", state: BatchStateUnknown, terminal: false},
+		{name: "creating", state: BatchStateCreating, terminal: false},
 		{name: "created", state: BatchStateCreated, terminal: false},
 		{name: "speculating", state: BatchStateSpeculating, terminal: false},
 		{name: "merging", state: BatchStateMerging, terminal: false},
@@ -44,6 +45,21 @@ func TestBatchState_IsTerminal(t *testing.T) {
 	}
 }
 
+func TestBatchStateSets(t *testing.T) {
+	assert.Contains(t, ActiveBatchStates(), BatchStateCreating)
+	assert.NotContains(t, DependencyBatchStates(), BatchStateCreating)
+	assert.ElementsMatch(t, []BatchState{
+		BatchStateCreating,
+		BatchStateCreated,
+		BatchStateSpeculating,
+		BatchStateMerging,
+		BatchStateSucceeded,
+		BatchStateFailed,
+		BatchStateCancelling,
+		BatchStateCancelled,
+	}, AllBatchStates())
+}
+
 func TestBatch_SerializationRoundTrip(t *testing.T) {
 	tests := []struct {
 		name  string

submitqueue/extension/storage/batch_dependent_store.go

@@ -25,16 +25,12 @@ import (
 // BatchDependentStore is an interface that defines methods for managing batch dependent information in the database.
 //
 // A BatchDependent is a reverse index ("batches that depend on me") paired one-to-one with a Batch.
-// The batch-creation flow always calls Create here before creating the Batch itself, so every active
-// Batch is guaranteed to have a corresponding BatchDependent row. Lookups via Get are only performed
-// for batch IDs returned from the active-batch set, meaning a missing row indicates data corruption or
-// out-of-band manipulation rather than a normal "not found" outcome. ErrNotFound is therefore part of
-// the contract for completeness but is not expected to be returned in steady-state operation.
+// The batch-creation flow creates this row while the Batch is Creating and before making the Batch eligible for pipeline processing.
+// A Creating Batch can briefly exist without its row; every Batch that reaches Created is guaranteed to have one.
 type BatchDependentStore interface {
 	// Get retrieves the batch dependent by batch ID.
 	// If the batch contains no dependents, the returned BatchDependent will have an empty Dependents list.
-	// Returns ErrNotFound if the batch itself is not found, which should never happen in steady-state system and
-	// therefore does not need a special handling.
+	// Returns ErrNotFound if no reverse-index row exists for the batch.
 	Get(ctx context.Context, batchID string) (entity.BatchDependent, error)
 
 	// Create creates a new batch dependent.

submitqueue/orchestrator/controller/README.md

@@ -81,6 +81,7 @@ Version arithmetic follows the [storage optimistic-locking contract](../../exten
 
 | Batch state | Behavior |
 |---|---|
+| `Creating` | Acknowledge an early dependency wake; the batch controller publishes again after initialization. |
 | `Created` | Start speculation: publish to `build`, then record `Speculating`. |
 | `Speculating` | Once dependencies resolve, publish to `merge` and record `Merging`. |
 | `Merging` | Acknowledge without regressing the batch; the merge controller owns recovery. |

submitqueue/orchestrator/controller/speculate/speculate.go

@@ -36,6 +36,7 @@ import (
 // Per invocation, the controller advances the batch one step in the
 // state machine:
 //
+//   - Creating          → no-op; the batch controller publishes again after reverse-index initialization completes.
 //   - Created → publish to build, transition to Speculating.
 //   - Speculating       → if all deps are Succeeded, publish to merge and
 //     transition to Merging; otherwise no-op (or fail-fast if a dep is
@@ -133,6 +134,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	}
 
 	switch batch.State {
+	case entity.BatchStateCreating:
+		// A dependency can finish after this batch subscribes in its reverse index but before initialization transitions it to Created.
+		// Ack that early wake; the batch controller publishes again after initialization.
+		metrics.NamedCounter(c.metricsScope, opName, "noop_creating", 1)
+		return nil
 	case entity.BatchStateCreated:
 		return c.startSpeculation(ctx, batch)
 	case entity.BatchStateSpeculating:
@@ -357,13 +363,10 @@ func (c *Controller) cancelBuild(ctx context.Context, batch entity.Batch) error
 	return nil
 }
 
-// respeculateDependents publishes a speculate event for every batch that
-// depends on the given batch. The batch controller creates a BatchDependent
-// row (with Dependents possibly empty) for every batch it persists, so a
-// missing row at this point is a storage invariant violation, not a normal
-// "no dependents" case — surface ErrNotFound as a regular storage error so
-// the message nacks and either an operator or the batch controller's own
-// crash-recovery can resolve the inconsistency.
+// respeculateDependents publishes a speculate event for every batch that depends on the given batch.
+// The batch controller creates a BatchDependent row before transitioning Creating → Created.
+// Only fully initialized batches are published and can reach a terminal state, so a missing row is a storage invariant violation.
+// Surface ErrNotFound as a regular storage error so the message nacks.
 //
 // Called both from the cancelBatch terminal flow and from the terminal
 // self-heal branch on redelivery of an already-Cancelled batch.

submitqueue/orchestrator/controller/speculate/speculate_test.go

@@ -129,6 +129,20 @@ func TestController_Process_StartSpeculation(t *testing.T) {
 	}
 }
 
+func TestController_Process_CreatingNoOp(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	batch := testBatch(entity.BatchStateCreating)
+
+	batchStore := storagemock.NewMockBatchStore(ctrl)
+	batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil)
+
+	store := storagemock.NewMockStorage(ctrl)
+	store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
+
+	controller := newTestController(t, ctrl, store, nil)
+	require.NoError(t, runProcess(t, ctrl, controller, batch.ID))
+}
+
 // tryFinalize: Speculating with no deps should publish to merge and CAS to Merging.
 func TestController_Process_FinalizeNoDeps(t *testing.T) {
 	ctrl := gomock.NewController(t)