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
10 changes: 7 additions & 3 deletions submitqueue/core/request/terminate.go
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,7 @@ func TerminateRequest(
// logVersion is the request version reflected in the published terminal log.
// It stays at the current version on the idempotent same-state path and
// advances to the new version only after a successful reconciling write.
beforeState := request.State
logVersion := request.Version
outcome := TerminationOutcomeSuccess
switch {
Expand All @@ -131,10 +132,13 @@ func TerminateRequest(
}, nil
default:
newVersion := request.Version + 1
if err := store.GetRequestStore().UpdateState(ctx, requestID, request.Version, newVersion, targetState); err != nil {
updatedRequest := request.WithState(targetState)
if err := store.GetRequestStore().Update(ctx, updatedRequest, request.Version, newVersion); err != nil {
return TerminationResult{}, fmt.Errorf("failed to update request %s state to %s: %w", requestID, targetState, err)
}
logVersion = newVersion
updatedRequest.Version = newVersion
request = updatedRequest
logVersion = request.Version
}

logEntry := entity.NewRequestLog(requestID, status, logVersion, lastError, metadata)
Expand All @@ -144,7 +148,7 @@ func TerminateRequest(

return TerminationResult{
Outcome: outcome,
BeforeState: request.State,
BeforeState: beforeState,
AfterState: targetState,
}, nil
}
Expand Down
8 changes: 5 additions & 3 deletions submitqueue/core/request/terminate_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ func TestTerminateRequest(t *testing.T) {
const requestID = "q/1"

validated := entity.Request{ID: requestID, Queue: "q", State: entity.RequestStateValidated, Version: 3}
originalValidated := validated

testCases := map[string]struct {
targetState entity.RequestState
Expand Down Expand Up @@ -101,7 +102,7 @@ func TestTerminateRequest(t *testing.T) {
metadata: map[string]string{"source": "validate"},
mockFunc: func(rs *storagemock.MockRequestStore) {
rs.EXPECT().Get(gomock.Any(), requestID).Return(validated, nil)
rs.EXPECT().UpdateState(gomock.Any(), requestID, int32(3), int32(4), entity.RequestStateError).Return(nil)
rs.EXPECT().Update(gomock.Any(), validated.WithState(entity.RequestStateError), int32(3), int32(4)).Return(nil)
},
wantResult: TerminationResult{
Outcome: TerminationOutcomeSuccess,
Expand Down Expand Up @@ -157,7 +158,7 @@ func TestTerminateRequest(t *testing.T) {
targetState: entity.RequestStateError,
mockFunc: func(rs *storagemock.MockRequestStore) {
rs.EXPECT().Get(gomock.Any(), requestID).Return(validated, nil)
rs.EXPECT().UpdateState(gomock.Any(), requestID, int32(3), int32(4), entity.RequestStateError).Return(storage.ErrVersionMismatch)
rs.EXPECT().Update(gomock.Any(), validated.WithState(entity.RequestStateError), int32(3), int32(4)).Return(storage.ErrVersionMismatch)
},
wantResult: TerminationResult{Outcome: TerminationOutcomeUnknown},
errMsg: "version mismatch",
Expand All @@ -167,7 +168,7 @@ func TestTerminateRequest(t *testing.T) {
targetState: entity.RequestStateError,
mockFunc: func(rs *storagemock.MockRequestStore) {
rs.EXPECT().Get(gomock.Any(), requestID).Return(validated, nil)
rs.EXPECT().UpdateState(gomock.Any(), requestID, int32(3), int32(4), entity.RequestStateError).Return(nil)
rs.EXPECT().Update(gomock.Any(), validated.WithState(entity.RequestStateError), int32(3), int32(4)).Return(nil)
},
publishErr: fmt.Errorf("connection refused"),
wantResult: TerminationResult{Outcome: TerminationOutcomeUnknown},
Expand All @@ -187,6 +188,7 @@ func TestTerminateRequest(t *testing.T) {

res, err := TerminateRequest(context.Background(), store, registry, requestID, tc.targetState, tc.lastError, tc.metadata)

assert.Equal(t, originalValidated, validated)
assert.Equal(t, tc.wantResult, res)
if tc.errMsg != "" {
assert.ErrorContains(t, err, tc.errMsg)
Expand Down
7 changes: 7 additions & 0 deletions submitqueue/entity/request.go
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,13 @@ type Request struct {
Version int32 `json:"version"`
}

// WithState returns a shallow copy of the request with State replaced.
// Slice fields continue to share their backing arrays.
func (r Request) WithState(state RequestState) Request {
r.State = state
return r
}

// ToBytes serializes the Request to JSON bytes for queue message payload.
func (r Request) ToBytes() ([]byte, error) {
return json.Marshal(r)
Expand Down
19 changes: 19 additions & 0 deletions submitqueue/entity/request_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,25 @@ func TestRequestFromBytes_EmptyData(t *testing.T) {
assert.Equal(t, int32(0), req.Version)
}

func TestRequest_WithState(t *testing.T) {
request := Request{
ID: "queueA/1",
Queue: "queueA",
State: RequestStateStarted,
Version: 1,
}

updated := request.WithState(RequestStateValidated)

assert.Equal(t, RequestStateStarted, request.State)
assert.Equal(t, Request{
ID: request.ID,
Queue: request.Queue,
State: RequestStateValidated,
Version: request.Version,
}, updated)
}

func TestIsRequestStateTerminal(t *testing.T) {
tests := []struct {
state RequestState
Expand Down
22 changes: 14 additions & 8 deletions submitqueue/extension/storage/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,25 +6,31 @@ Pluggable persistence interfaces for SubmitQueue entities (requests, batches, de

Entities that support concurrent mutation carry an `int32 Version` field. Updates are conditional on the version: the write only succeeds if the persisted version matches the caller's expected version. On mismatch, the implementation returns `storage.ErrVersionMismatch`, which is declared as a retryable infrastructure error so callers can return it without reclassifying it.

**Version arithmetic is owned by the controller, not the store.** Update methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write):
**Updates replace every non-primary-key field.** Callers must pass a complete authoritative entity loaded from storage or constructed with every persisted field; sparse patch entities can clear unrelated columns. The primary key identifies the row and is not rewritten.

**Version arithmetic is owned by the controller, not the store.** Versioned update methods take a complete entity plus both `oldVersion` (the where-clause guard) and `newVersion` (the value to write):

```go
UpdateState(ctx, id, oldVersion, newVersion int32, newState entity.RequestState) error
Update(ctx, request entity.Request, oldVersion, newVersion int32) error
```

The store performs a pure conditional write — it does not compute `oldVersion + 1` internally. This keeps the in-memory entity and the persisted row in sync without the storage layer mutating values the caller didn't supply.
The store writes `newVersion` rather than the entity's current `Version` and performs a pure conditional write — it does not compute `oldVersion + 1` internally. This keeps the in-memory entity and the persisted row in sync without the storage layer mutating values the caller didn't supply.

### Caller pattern

```go
newVersion := entity.Version + 1
if err := store.UpdateState(ctx, entity.ID, entity.Version, newVersion, newState); err != nil {
return err // entity.Version unchanged on failure — safe to retry
oldVersion := request.Version
newVersion := oldVersion + 1
updated := request
updated.State = newState
if err := store.Update(ctx, updated, oldVersion, newVersion); err != nil {
return err // request remains unchanged on failure — safe to retry
}
entity.Version = newVersion // only after the write succeeded
updated.Version = newVersion
request = updated // only after the write succeeded
```

The post-success assignment matters whenever the entity is read again later in the same flow. Pre-incrementing in memory before the call is a bug pattern: if the call fails and the caller swallows the error, the in-memory version is now ahead of the database and subsequent updates will fail with `ErrVersionMismatch` for non-obvious reasons.
The candidate-copy pattern keeps the caller-owned entity unchanged if the write fails. Clone slice and map fields before changing their contents so the candidate cannot mutate the original through shared backing storage. The post-success assignment matters whenever the entity is read again later in the same flow. Pre-incrementing in memory before the call is a bug pattern: if the call fails and the caller swallows the error, the in-memory version is now ahead of the database and subsequent updates will fail with `ErrVersionMismatch` for non-obvious reasons.

## Read-after-write consistency

Expand Down
12 changes: 6 additions & 6 deletions submitqueue/extension/storage/mock/request_store_mock.go

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

28 changes: 16 additions & 12 deletions submitqueue/extension/storage/mysql/request_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -93,36 +93,40 @@ func (r *requestStore) Create(ctx context.Context, request entity.Request) (retE
return nil
}

// UpdateState updates the state of a land request to newState and the version to newVersion
// if the current persisted version matches oldVersion. If versions do not match, returns ErrVersionMismatch.
// Version arithmetic is owned by the caller; this is a pure conditional write.
func (r *requestStore) UpdateState(ctx context.Context, id string, oldVersion, newVersion int32, newState entity.RequestState) (retErr error) {
// Update replaces every non-key field of a land request and writes newVersion if the current persisted version matches oldVersion.
// If versions do not match, returns ErrVersionMismatch. Version arithmetic is owned by the caller; this is a pure conditional write.
func (r *requestStore) Update(ctx context.Context, request entity.Request, oldVersion, newVersion int32) (retErr error) {
op := metrics.Begin(r.scope, "update_state", metrics.StorageLatencyBuckets)
defer func() { op.Complete(retErr) }()

changeURIsJSON, err := json.Marshal(request.Change.URIs)
if err != nil {
return fmt.Errorf("failed to marshal change URIs for request id=%s: %w", request.ID, err)
}

result, err := r.db.ExecContext(ctx,
"UPDATE request SET state = ?, version = ? WHERE id = ? AND version = ?",
newState, newVersion, id, oldVersion,
"UPDATE request SET queue = ?, change_uri = ?, land_strategy = ?, state = ?, version = ? WHERE id = ? AND version = ?",
request.Queue, changeURIsJSON, request.LandStrategy, request.State, newVersion, request.ID, oldVersion,
)
if err != nil {
return fmt.Errorf(
"failed to update request state for id=%q oldVersion=%d newVersion=%d newState=%v: %w",
id, oldVersion, newVersion, newState, err,
"failed to update request for id=%q oldVersion=%d newVersion=%d: %w",
request.ID, oldVersion, newVersion, err,
)
}

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

if rowsAffected != 1 {
return fmt.Errorf(
"version mismatch for request update: id=%q expected_version=%d newState=%v: %w",
id, oldVersion, newState, storage.ErrVersionMismatch,
"version mismatch for request update: id=%q expected_version=%d: %w",
request.ID, oldVersion, storage.ErrVersionMismatch,
)
}

Expand Down
62 changes: 50 additions & 12 deletions submitqueue/extension/storage/mysql/request_store_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -198,53 +198,91 @@ func TestRequestStore_Create(t *testing.T) {
}
}

func TestRequestStore_UpdateState(t *testing.T) {
const id = "monorepo/1"
func TestRequestStore_Update(t *testing.T) {
const oldVersion, newVersion = int32(1), int32(2)
const newState = entity.RequestStateValidated
request := entity.Request{
ID: "monorepo/1",
Queue: "monorepo-updated",
Change: change.Change{URIs: []string{"github://github.example.com/uber/submitqueue/pull/456/cafebabe"}},
LandStrategy: mergestrategy.MergeStrategySquashRebase,
State: entity.RequestStateValidated,
Version: oldVersion,
}
changeURIsJSON, err := json.Marshal(request.Change.URIs)
require.NoError(t, err)

tests := []struct {
name string
request entity.Request
setup func(mock sqlmock.Sqlmock)
wantErr bool
wantErrIs error
}{
{
name: "success",
name: "success",
request: request,
setup: func(mock sqlmock.Sqlmock) {
mock.ExpectExec("UPDATE request").
WithArgs(newState, newVersion, id, oldVersion).
WithArgs(request.Queue, changeURIsJSON, request.LandStrategy, request.State, newVersion, request.ID, oldVersion).
WillReturnResult(sqlmock.NewResult(0, 1))
},
},
{
name: "version mismatch",
name: "version mismatch",
request: request,
setup: func(mock sqlmock.Sqlmock) {
mock.ExpectExec("UPDATE request").
WithArgs(newState, newVersion, id, oldVersion).
WithArgs(request.Queue, changeURIsJSON, request.LandStrategy, request.State, newVersion, request.ID, oldVersion).
WillReturnResult(sqlmock.NewResult(0, 0))
},
wantErr: true,
wantErrIs: storage.ErrVersionMismatch,
},
{
name: "exec error",
name: "exec error",
request: request,
setup: func(mock sqlmock.Sqlmock) {
mock.ExpectExec("UPDATE request").
WithArgs(newState, newVersion, id, oldVersion).
WithArgs(request.Queue, changeURIsJSON, request.LandStrategy, request.State, newVersion, request.ID, oldVersion).
WillReturnError(fmt.Errorf("connection reset"))
},
wantErr: true,
},
{
name: "rows affected error",
name: "rows affected error",
request: request,
setup: func(mock sqlmock.Sqlmock) {
mock.ExpectExec("UPDATE request").
WithArgs(newState, newVersion, id, oldVersion).
WithArgs(request.Queue, changeURIsJSON, request.LandStrategy, request.State, newVersion, request.ID, oldVersion).
WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error")))
},
wantErr: true,
},
{
name: "nil change URIs",
request: entity.Request{ID: request.ID, Queue: request.Queue, LandStrategy: request.LandStrategy, State: request.State, Version: request.Version},
setup: func(mock sqlmock.Sqlmock) {
mock.ExpectExec("UPDATE request").
WithArgs(request.Queue, []byte("null"), request.LandStrategy, request.State, newVersion, request.ID, oldVersion).
WillReturnResult(sqlmock.NewResult(0, 1))
},
},
{
name: "empty change URIs",
request: entity.Request{
ID: request.ID,
Queue: request.Queue,
Change: change.Change{URIs: []string{}},
LandStrategy: request.LandStrategy,
State: request.State,
Version: request.Version,
},
setup: func(mock sqlmock.Sqlmock) {
mock.ExpectExec("UPDATE request").
WithArgs(request.Queue, []byte("[]"), request.LandStrategy, request.State, newVersion, request.ID, oldVersion).
WillReturnResult(sqlmock.NewResult(0, 1))
},
},
}

for _, tt := range tests {
Expand All @@ -254,7 +292,7 @@ func TestRequestStore_UpdateState(t *testing.T) {

tt.setup(mock)

err := store.UpdateState(context.Background(), id, oldVersion, newVersion, newState)
err := store.Update(context.Background(), tt.request, oldVersion, newVersion)
if tt.wantErr {
require.Error(t, err)
if tt.wantErrIs != nil {
Expand Down
7 changes: 3 additions & 4 deletions submitqueue/extension/storage/request_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,7 @@ type RequestStore interface {
// Returns ErrAlreadyExists if a request with the same ID already exists.
Create(ctx context.Context, request entity.Request) error

// UpdateState updates the state of a land request to newState and the version to newVersion
// if the current persisted version matches oldVersion. If versions do not match, returns ErrVersionMismatch.
// Version arithmetic is owned by the caller; the store performs a pure conditional write.
UpdateState(ctx context.Context, id string, oldVersion, newVersion int32, newState entity.RequestState) error
// Update replaces every non-key field of a land request and writes newVersion if the current persisted version matches oldVersion.
// If versions do not match, returns ErrVersionMismatch. Version arithmetic is owned by the caller; the store performs a pure conditional write.
Update(ctx context.Context, request entity.Request, oldVersion, newVersion int32) error
}
9 changes: 5 additions & 4 deletions submitqueue/orchestrator/controller/batch/batch.go
Original file line number Diff line number Diff line change
Expand Up @@ -190,7 +190,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
// state, so it would CAS the request from Cancelled back to Landed, silently
// undoing the user's cancel.
//
// The CAS below collapses that window. Whichever of batch.UpdateState(...,
// The CAS below collapses that window. Whichever of request.Update(...,
// RequestStateBatched) and cancel.markCancelling(... RequestStateCancelling)
// reaches storage first wins; the loser sees storage.ErrVersionMismatch:
// - If cancel won: this CAS fails. We ack the message (cancel will drive R
Expand Down Expand Up @@ -221,7 +221,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
// (request cancelled) is still correct — the orphan batch just gets
// reconciled by conclude as if it had no requests to act on.
newRequestVersion := request.Version + 1
if err := c.store.GetRequestStore().UpdateState(ctx, request.ID, request.Version, newRequestVersion, entity.RequestStateBatched); err != nil {
updatedRequest := request.WithState(entity.RequestStateBatched)
if err := c.store.GetRequestStore().Update(ctx, updatedRequest, request.Version, newRequestVersion); err != nil {
// ErrVersionMismatch == cancel (or another writer) advanced R first. Ack
// the message: there is nothing for us to do, and retrying would not help
// since the new state of R is now visible to the cancel pipeline.
Expand All @@ -237,8 +238,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
metrics.NamedCounter(c.metricsScope, opName, "request_claim_errors", 1)
return fmt.Errorf("failed to claim request %s for batch %s: %w", request.ID, batch.ID, err)
}
request.Version = newRequestVersion
request.State = entity.RequestStateBatched
updatedRequest.Version = newRequestVersion
request = updatedRequest

// Persist the batch before creating references to it. A Creating batch is not eligible for dependency analysis or normal processing.
if err := c.store.GetBatchStore().Create(ctx, batch); err != nil {
Expand Down
Loading
Loading