Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 21 additions & 1 deletion submitqueue/entity/batch.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand Down
16 changes: 16 additions & 0 deletions submitqueue/entity/batch_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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},
Expand All @@ -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
Expand Down
10 changes: 3 additions & 7 deletions submitqueue/extension/storage/batch_dependent_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
1 change: 1 addition & 0 deletions submitqueue/orchestrator/controller/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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. |
Expand Down
17 changes: 10 additions & 7 deletions submitqueue/orchestrator/controller/speculate/speculate.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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.
Expand Down
14 changes: 14 additions & 0 deletions submitqueue/orchestrator/controller/speculate/speculate_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading