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"
@@ -67,9 +70,8 @@ func IsBatchStateHalted(s BatchState) bool {
 	return s.IsTerminal() || s == BatchStateCancelling
 }
 
-// ActiveBatchStates returns every non-terminal batch state that must be considered in-flight.
-// Use this when callers need to find batches that still own a request, including Cancelling
-// batches that cancel redelivery must be able to resolve.
+// ActiveBatchStates returns batch states eligible for active pipeline and cancellation lookups.
+// Creating is excluded because its reverse-index structure may still be incomplete.
 func ActiveBatchStates() []BatchState {
 	return []BatchState{
 		BatchStateCreated,

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,14 @@ func TestBatchState_IsTerminal(t *testing.T) {
 	}
 }
 
+func TestActiveBatchStates_ExcludesCreating(t *testing.T) {
+	assert.NotContains(t, ActiveBatchStates(), BatchStateCreating)
+}
+
+func TestDependencyBatchStates_ExcludesCreating(t *testing.T) {
+	assert.NotContains(t, DependencyBatchStates(), BatchStateCreating)
+}
+
 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.