refactor(storage): replace request state updates

Persist complete Request entities through guarded updates, migrate callers to candidate copies, and document full-entity replacement semantics.

Jira Issues

CODEM-204

Author: ubettigole

submitqueue/core/request/terminate.go

@@ -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 {
@@ -131,10 +132,12 @@ func TerminateRequest(
 		}, nil
 	default:
 		newVersion := request.Version + 1
-		if err := store.GetRequestStore().UpdateState(ctx, requestID, request.Version, newVersion, targetState); err != nil {
+		request.State = targetState
+		if err := store.GetRequestStore().Update(ctx, request, request.Version, newVersion); err != nil {
 			return TerminationResult{}, fmt.Errorf("failed to update request %s state to %s: %w", requestID, targetState, err)
 		}
-		logVersion = newVersion
+		request.Version = newVersion
+		logVersion = request.Version
 	}
 
 	logEntry := entity.NewRequestLog(requestID, status, logVersion, lastError, metadata)
@@ -144,7 +147,7 @@ func TerminateRequest(
 
 	return TerminationResult{
 		Outcome:     outcome,
-		BeforeState: request.State,
+		BeforeState: beforeState,
 		AfterState:  targetState,
 	}, nil
 }

submitqueue/core/request/terminate_test.go

@@ -31,6 +31,11 @@ import (
 	"go.uber.org/mock/gomock"
 )
 
+func requestWithState(request entity.Request, state entity.RequestState) entity.Request {
+	request.State = state
+	return request
+}
+
 // recordingRegistry returns a registry whose publisher appends every published
 // request log to *logs and returns publishErr. It lets tests assert both that a
 // terminal log was (or was not) published and what version/status it carried.
@@ -61,6 +66,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
@@ -101,7 +107,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(), requestWithState(validated, entity.RequestStateError), int32(3), int32(4)).Return(nil)
 			},
 			wantResult: TerminationResult{
 				Outcome:     TerminationOutcomeSuccess,
@@ -157,7 +163,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(), requestWithState(validated, entity.RequestStateError), int32(3), int32(4)).Return(storage.ErrVersionMismatch)
 			},
 			wantResult: TerminationResult{Outcome: TerminationOutcomeUnknown},
 			errMsg:     "version mismatch",
@@ -167,7 +173,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(), requestWithState(validated, entity.RequestStateError), int32(3), int32(4)).Return(nil)
 			},
 			publishErr: fmt.Errorf("connection refused"),
 			wantResult: TerminationResult{Outcome: TerminationOutcomeUnknown},
@@ -187,6 +193,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)

submitqueue/extension/storage/README.md

@@ -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
 

submitqueue/extension/storage/mock/request_store_mock.go

@@ -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,
 		)
 	}
 

submitqueue/extension/storage/mysql/request_store.go

@@ -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 {
@@ -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 {

submitqueue/extension/storage/mysql/request_store_test.go

@@ -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
 }

submitqueue/extension/storage/request_store.go

@@ -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
@@ -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 {
+	request.State = entity.RequestStateBatched
+	if err := c.store.GetRequestStore().Update(ctx, request, 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.
@@ -238,7 +239,6 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to claim request %s for batch %s: %w", request.ID, batch.ID, err)
 	}
 	request.Version = newRequestVersion
-	request.State = entity.RequestStateBatched
 
 	// 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 {