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
2 changes: 1 addition & 1 deletion doc/rfc/stovepipe/steps/build.md
Original file line number Diff line number Diff line change
Expand Up @@ -316,7 +316,7 @@ Either could be adopted independently: the idempotency token, if a backend that

Plus the `BuildID{ID string}` wire type in `stovepipe/entity` (same "id only travels" convention as `RequestID`, shaped like SubmitQueue's own `entity.BuildID` but not the same Go type — see the [contract sketch](#stovepipe-buildrunner-contract-design-sketch)), wrapping the one runner-assigned id everywhere it appears — `Trigger`'s return, the queue payload, `Status`/`Cancel`'s parameter. `buildsignal` reaches a build by the id carried in its message, and `record` reads the `Request` (whose state carries the build's outcome) rather than a `Build`, so no reverse index from `Request` to its builds is ever needed.

**`BuildStore`** (new, added to the `Storage` aggregator via `GetBuildStore()`), matching stovepipe's existing `RequestStore` conventions — **generic `Update` with caller-owned version arithmetic, not SubmitQueue's field-specific `UpdateStatus`**:
**`BuildStore`** (new, added to the `Storage` aggregator via `GetBuildStore()`), matching stovepipe's existing `RequestStore` conventions — **generic `Update` with caller-owned version arithmetic**:

- `Create(ctx, build entity.Build) error` — `ErrAlreadyExists` if the id is taken.
- `Get(ctx, id string) (entity.Build, error)` — `ErrNotFound` if absent.
Expand Down
4 changes: 2 additions & 2 deletions submitqueue/extension/storage/build_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,6 @@ type BuildStore interface {
// Returns ErrAlreadyExists if a build with the same ID already exists.
Create(ctx context.Context, build entity.Build) error

// UpdateStatus updates the status of a build.
UpdateStatus(ctx context.Context, id string, newStatus entity.BuildStatus) error
// Update replaces all non-key fields of a build.
Update(ctx context.Context, build entity.Build) error
}
12 changes: 6 additions & 6 deletions submitqueue/extension/storage/mock/build_store_mock.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

14 changes: 7 additions & 7 deletions submitqueue/extension/storage/mysql/build_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -80,26 +80,26 @@ func (s *buildStore) Create(ctx context.Context, build entity.Build) (retErr err
return nil
}

// UpdateStatus updates the status of a build. Returns ErrNotFound if the build is not found.
func (s *buildStore) UpdateStatus(ctx context.Context, id string, newStatus entity.BuildStatus) (retErr error) {
// Update replaces all non-key fields of a build. Returns ErrNotFound if the build is not found.
func (s *buildStore) Update(ctx context.Context, build entity.Build) (retErr error) {
op := metrics.Begin(s.scope, "update_status", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

result, err := s.db.ExecContext(ctx,
"UPDATE build SET status = ? WHERE id = ?",
newStatus, id,
"UPDATE build SET batch_id = ?, status = ? WHERE id = ?",
build.BatchID, build.Status, build.ID,
)
if err != nil {
return fmt.Errorf("failed to update build status for id=%q newStatus=%v: %w", id, newStatus, err)
return fmt.Errorf("failed to update build entity id=%q: %w", build.ID, err)
}

rowsAffected, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("failed to get rows affected from update for id=%q newStatus=%v: %w", id, newStatus, err)
return fmt.Errorf("failed to get rows affected from update for build entity id=%q: %w", build.ID, err)
}

if rowsAffected != 1 {
return storage.WrapNotFound(fmt.Errorf("build entity id=%s", id))
return storage.WrapNotFound(fmt.Errorf("build entity id=%s", build.ID))
}

return nil
Expand Down
19 changes: 11 additions & 8 deletions submitqueue/extension/storage/mysql/build_store_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -175,9 +175,12 @@ func TestBuildStore_Create(t *testing.T) {
}
}

func TestBuildStore_UpdateStatus(t *testing.T) {
const id = "bk-1001"
const newStatus = entity.BuildStatusSucceeded
func TestBuildStore_Update(t *testing.T) {
build := entity.Build{
ID: "bk-1001",
BatchID: "monorepo/batch/2",
Status: entity.BuildStatusSucceeded,
}

tests := []struct {
name string
Expand All @@ -189,15 +192,15 @@ func TestBuildStore_UpdateStatus(t *testing.T) {
name: "success",
setup: func(mock sqlmock.Sqlmock) {
mock.ExpectExec("UPDATE build").
WithArgs(newStatus, id).
WithArgs(build.BatchID, build.Status, build.ID).
WillReturnResult(sqlmock.NewResult(0, 1))
},
},
{
name: "not found",
setup: func(mock sqlmock.Sqlmock) {
mock.ExpectExec("UPDATE build").
WithArgs(newStatus, id).
WithArgs(build.BatchID, build.Status, build.ID).
WillReturnResult(sqlmock.NewResult(0, 0))
},
wantErr: true,
Expand All @@ -207,7 +210,7 @@ func TestBuildStore_UpdateStatus(t *testing.T) {
name: "exec error",
setup: func(mock sqlmock.Sqlmock) {
mock.ExpectExec("UPDATE build").
WithArgs(newStatus, id).
WithArgs(build.BatchID, build.Status, build.ID).
WillReturnError(fmt.Errorf("connection reset"))
},
wantErr: true,
Expand All @@ -216,7 +219,7 @@ func TestBuildStore_UpdateStatus(t *testing.T) {
name: "rows affected error",
setup: func(mock sqlmock.Sqlmock) {
mock.ExpectExec("UPDATE build").
WithArgs(newStatus, id).
WithArgs(build.BatchID, build.Status, build.ID).
WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error")))
},
wantErr: true,
Expand All @@ -230,7 +233,7 @@ func TestBuildStore_UpdateStatus(t *testing.T) {

tt.setup(mock)

err := store.UpdateStatus(context.Background(), id, newStatus)
err := store.Update(context.Background(), build)
if tt.wantErr {
require.Error(t, err)
if tt.wantErrIs != nil {
Expand Down
2 changes: 1 addition & 1 deletion submitqueue/orchestrator/controller/build/build.go
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
}

// Persist the initial Build snapshot so the buildsignal poll loop has a
// row to UpdateStatus against. ErrAlreadyExists is benign — a redelivery
// row to Update against. ErrAlreadyExists is benign — a redelivery
// of this message after a previous successful Create.
if err := c.store.GetBuildStore().Create(ctx, build); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
Expand Down
2 changes: 1 addition & 1 deletion submitqueue/orchestrator/controller/build/build_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -222,7 +222,7 @@ func TestController_Process_TriggersWithBaseAndHead(t *testing.T) {
// TestController_Process_BuildStoreAlreadyExistsIsSwallowed covers the
// redelivery case: Create returns ErrAlreadyExists, the controller proceeds
// to publish to buildsignal anyway. The polling loop will pick up the
// existing row via UpdateStatus.
// existing row via Update.
func TestController_Process_BuildStoreAlreadyExistsIsSwallowed(t *testing.T) {
ctrl := gomock.NewController(t)

Expand Down
17 changes: 9 additions & 8 deletions submitqueue/orchestrator/controller/buildsignal/buildsignal.go
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ func NewController(
// a delayed message back to this topic when the build is still in flight.
// Returns nil to ack (success), or error to nack/reject.
//
// Error classification: deserialize, Status, UpdateStatus, and the speculate
// Error classification: deserialize, Status, Update, and the speculate
// publish stay non-retryable — they reject straight to DLQ on the first
// failure, where the operational republish path is the recovery mechanism.
// Only the PublishAfter self-reschedule is retryable: it is the poll loop's
Expand Down Expand Up @@ -163,38 +163,39 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
return nil
}

build.Status = status
updatedBuild := build
updatedBuild.Status = status

if err := c.store.GetBuildStore().UpdateStatus(ctx, build.ID, status); err != nil {
if err := c.store.GetBuildStore().Update(ctx, updatedBuild); err != nil {
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
return fmt.Errorf("failed to update status for build %s: %w", build.ID, err)
}

// Re-evaluate the batch state machine with the latest build status.
if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, build.BatchID, msg.PartitionKey); err != nil {
if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, updatedBuild.BatchID, msg.PartitionKey); err != nil {
metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
return fmt.Errorf("failed to publish to speculate: %w", err)
}

if status.IsTerminal() {
metrics.NamedCounter(c.metricsScope, opName, "terminal", 1, metrics.NewTag("status", string(status)))
c.logger.Infow("build reached terminal status",
"build_id", build.ID,
"batch_id", build.BatchID,
"build_id", updatedBuild.ID,
"batch_id", updatedBuild.BatchID,
"status", string(status),
)
return nil
}

delayMs := pollDelay(status)
metrics.NamedCounter(c.metricsScope, opName, "rescheduled", 1, metrics.NewTag("status", string(status)))
if err := c.publishBuild(ctx, c.topicKey, build, delayMs); err != nil {
if err := c.publishBuild(ctx, c.topicKey, updatedBuild, delayMs); err != nil {
metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
return fmt.Errorf("failed to re-publish to buildsignal: %w", err)
}

c.logger.Debugw("rescheduled build status poll",
"build_id", build.ID,
"build_id", updatedBuild.ID,
"status", string(status),
"delay_ms", delayMs,
)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -134,11 +134,13 @@ func TestController_Process_Terminal(t *testing.T) {
h := newTestHarness(t, ctrl)

build := entity.Build{ID: "b-1", BatchID: "batch-1", Status: entity.BuildStatusAccepted}
updatedBuild := build
updatedBuild.Status = tt.status

h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil)
h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(tt.status, entity.BuildMetadata{}, nil)
h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil)
h.buildStore.EXPECT().UpdateStatus(gomock.Any(), build.ID, tt.status).Return(nil)
h.buildStore.EXPECT().Update(gomock.Any(), updatedBuild).Return(nil)
h.speculatePub.EXPECT().
Publish(gomock.Any(), "speculate", gomock.AssignableToTypeOf(entityqueue.Message{})).
DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error {
Expand Down Expand Up @@ -174,11 +176,13 @@ func TestController_Process_NonTerminal(t *testing.T) {
h := newTestHarness(t, ctrl)

build := entity.Build{ID: "b-2", BatchID: "batch-2", Status: entity.BuildStatusAccepted}
updatedBuild := build
updatedBuild.Status = tt.status

h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil)
h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(tt.status, entity.BuildMetadata{}, nil)
h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil)
h.buildStore.EXPECT().UpdateStatus(gomock.Any(), build.ID, tt.status).Return(nil)
h.buildStore.EXPECT().Update(gomock.Any(), updatedBuild).Return(nil)
h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil).Times(1)
h.signalPub.EXPECT().
PublishAfter(gomock.Any(), "buildsignal", gomock.AssignableToTypeOf(entityqueue.Message{}), tt.wantDelayMs).
Expand All @@ -205,24 +209,26 @@ func TestController_Process_StatusError(t *testing.T) {
h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil)
h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil)
h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusUnknown, nil, errors.New("provider down"))
// No UpdateStatus, no Publish, no PublishAfter expected.
// No Update, no Publish, no PublishAfter expected.

err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build))
require.Error(t, err)
// Non-retryable: rejects to DLQ on first failure; republish is the recovery path.
assert.False(t, errs.IsRetryable(err))
}

func TestController_Process_UpdateStatusError(t *testing.T) {
func TestController_Process_UpdateError(t *testing.T) {
ctrl := gomock.NewController(t)
h := newTestHarness(t, ctrl)

build := entity.Build{ID: "b-4", BatchID: "batch-4", Status: entity.BuildStatusAccepted}
updatedBuild := build
updatedBuild.Status = entity.BuildStatusRunning

h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil)
h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusRunning, nil, nil)
h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil)
h.buildStore.EXPECT().UpdateStatus(gomock.Any(), build.ID, entity.BuildStatusRunning).
h.buildStore.EXPECT().Update(gomock.Any(), updatedBuild).
Return(errors.New("db unreachable"))
// No Publish / PublishAfter expected after the store failure.

Expand All @@ -240,11 +246,13 @@ func TestController_Process_RepublishError(t *testing.T) {
h := newTestHarness(t, ctrl)

build := entity.Build{ID: "b-5", BatchID: "batch-5", Status: entity.BuildStatusAccepted}
updatedBuild := build
updatedBuild.Status = entity.BuildStatusRunning

h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil)
h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusRunning, entity.BuildMetadata{}, nil)
h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: entity.BatchStateSpeculating}, nil)
h.buildStore.EXPECT().UpdateStatus(gomock.Any(), build.ID, entity.BuildStatusRunning).Return(nil)
h.buildStore.EXPECT().Update(gomock.Any(), updatedBuild).Return(nil)
h.speculatePub.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).Return(nil).Times(1)
h.signalPub.EXPECT().
PublishAfter(gomock.Any(), "buildsignal", gomock.Any(), PollDelayRunningMs).
Expand All @@ -264,7 +272,7 @@ func TestController_Process_GetError(t *testing.T) {
build := entity.Build{ID: "b-6", BatchID: "batch-6", Status: entity.BuildStatusAccepted}

h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(entity.Build{}, errors.New("db unreachable"))
// No Status / UpdateStatus / Publish expected once the load fails.
// No Status / Update / Publish expected once the load fails.

err := h.controller.Process(context.Background(), buildDelivery(t, ctrl, build))
require.Error(t, err)
Expand Down Expand Up @@ -306,7 +314,7 @@ func TestController_Process_HaltedShortCircuit(t *testing.T) {
h.buildStore.EXPECT().Get(gomock.Any(), build.ID).Return(build, nil)
h.br.EXPECT().Status(gomock.Any(), entity.BuildID{ID: build.ID}).Return(entity.BuildStatusRunning, entity.BuildMetadata{}, nil)
h.batchStore.EXPECT().Get(gomock.Any(), build.BatchID).Return(entity.Batch{ID: build.BatchID, State: state}, nil)
// Halted: no UpdateStatus, no speculate Publish, no buildsignal
// Halted: no Update, no speculate Publish, no buildsignal
// PublishAfter. The harness publishers have no expectations, so any
// publish fails the test.

Expand Down
8 changes: 5 additions & 3 deletions submitqueue/orchestrator/controller/speculate/speculate.go
Original file line number Diff line number Diff line change
Expand Up @@ -265,7 +265,7 @@ func (c *Controller) failOnDependency(ctx context.Context, batch entity.Batch, d
// Order matters for correctness:
//
// 1. Cancel the in-flight Build entity (build.ID == batch.ID; one Get + one
// UpdateStatus covers all builds for this batch). A future external CI
// Update covers all builds for this batch). A future external CI
// integration hooks in here. Idempotent: tolerate ErrNotFound (no build
// was scheduled), skip if already terminal.
//
Expand Down Expand Up @@ -332,7 +332,7 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error
// This is the hook point for a future external CI integration: today the
// system has no external runner, so the local state flip is the complete
// cancellation. Once a runner exists, it must be invoked here before the
// local UpdateStatus.
// local Update.
func (c *Controller) cancelBuild(ctx context.Context, batch entity.Batch) error {
build, err := c.store.GetBuildStore().Get(ctx, batch.ID)
if err != nil {
Expand All @@ -349,7 +349,9 @@ func (c *Controller) cancelBuild(ctx context.Context, batch entity.Batch) error
return nil
}

if err := c.store.GetBuildStore().UpdateStatus(ctx, batch.ID, entity.BuildStatusCancelled); err != nil {
updatedBuild := build
updatedBuild.Status = entity.BuildStatusCancelled
if err := c.store.GetBuildStore().Update(ctx, updatedBuild); err != nil {
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
return fmt.Errorf("failed to cancel build for batch %s: %w", batch.ID, err)
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -358,10 +358,13 @@ func TestController_Process_CancellingTerminalFlow(t *testing.T) {
batchStore.EXPECT().UpdateState(gomock.Any(), batch.ID, int32(1), int32(2), entity.BatchStateCancelled).Return(nil)

buildStore := storagemock.NewMockBuildStore(ctrl)
buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{
build := entity.Build{
ID: batch.ID, BatchID: batch.ID, Status: entity.BuildStatusRunning,
}, nil)
buildStore.EXPECT().UpdateStatus(gomock.Any(), batch.ID, entity.BuildStatusCancelled).Return(nil)
}
buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(build, nil)
updatedBuild := build
updatedBuild.Status = entity.BuildStatusCancelled
buildStore.EXPECT().Update(gomock.Any(), updatedBuild).Return(nil)

depStore := storagemock.NewMockBatchDependentStore(ctrl)
depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{
Expand Down Expand Up @@ -412,7 +415,7 @@ func TestController_Process_CancellingTerminalFlow(t *testing.T) {

// If the build for the batch has already reached a terminal status (e.g. CI
// finished naturally between the cancel intent and the speculate pickup), the
// cancellation must not re-flip it — UpdateStatus must never fire. The rest
// cancellation must not re-flip it — Update must never fire. The rest
// of the flow (terminal batch CAS, dependent fan-out, conclude) still runs.
func TestController_Process_CancellingBuildAlreadyTerminal(t *testing.T) {
ctrl := gomock.NewController(t)
Expand All @@ -426,7 +429,7 @@ func TestController_Process_CancellingBuildAlreadyTerminal(t *testing.T) {
buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{
ID: batch.ID, BatchID: batch.ID, Status: entity.BuildStatusSucceeded,
}, nil)
// No UpdateStatus expected — the build is already terminal.
// No Update expected — the build is already terminal.

depStore := storagemock.NewMockBatchDependentStore(ctrl)
depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{
Expand Down Expand Up @@ -455,7 +458,7 @@ func TestController_Process_CancellingNoBuildYet(t *testing.T) {

buildStore := storagemock.NewMockBuildStore(ctrl)
buildStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.Build{}, storage.ErrNotFound)
// No UpdateStatus expected.
// No Update expected.

depStore := storagemock.NewMockBatchDependentStore(ctrl)
depStore.EXPECT().Get(gomock.Any(), batch.ID).Return(entity.BatchDependent{
Expand Down
Loading