From af820bddbe55c5596301a06c4ad7b0d89cf0de46 Mon Sep 17 00:00:00 2001 From: abettigole Date: Fri, 31 Jul 2026 23:44:31 +0000 Subject: [PATCH] refactor(storage): replace complete request summaries Persist every non-key RequestSummary field while retaining optimistic version guards and full-row integration coverage. Jira Issues CODEM-204 --- .../storage/mysql/request_summary_store.go | 17 ++-- .../mysql/request_summary_store_test.go | 68 ++++++++++++---- .../storage/request_summary_store.go | 2 +- .../submitqueue/extension/storage/suite.go | 77 +++++++++++++++---- 4 files changed, 127 insertions(+), 37 deletions(-) diff --git a/submitqueue/extension/storage/mysql/request_summary_store.go b/submitqueue/extension/storage/mysql/request_summary_store.go index 3894564e..3f38cadd 100644 --- a/submitqueue/extension/storage/mysql/request_summary_store.go +++ b/submitqueue/extension/storage/mysql/request_summary_store.go @@ -45,7 +45,7 @@ func (s *requestSummaryStore) Create(ctx context.Context, summary entity.Request changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary.ChangeURIs, summary.Metadata) if err != nil { - return fmt.Errorf("failed to marshal request summary request_id=%s: %w", summary.RequestID, err) + return fmt.Errorf("failed to marshal request summary metadata request_id=%s: %w", summary.RequestID, err) } _, err = s.db.ExecContext(ctx, ` @@ -103,19 +103,20 @@ func (s *requestSummaryStore) Update(ctx context.Context, summary entity.Request op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets) defer func() { op.Complete(retErr) }() - metadata := normalizeMetadata(summary.Metadata) - metadataJSON, err := json.Marshal(metadata) + changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary.ChangeURIs, summary.Metadata) if err != nil { - return fmt.Errorf("failed to marshal request summary metadata request_id=%s: %w", summary.RequestID, err) + return fmt.Errorf("failed to marshal request summary request_id=%s: %w", summary.RequestID, err) } result, err := s.db.ExecContext(ctx, ` UPDATE request_summary - SET status = ?, request_version = ?, status_timestamp_ms = ?, - version = ?, last_error = ?, metadata = ? + SET queue = ?, change_uris = ?, received_at_ms = ?, status = ?, + request_version = ?, status_timestamp_ms = ?, version = ?, + last_error = ?, metadata = ? WHERE request_id = ? AND version = ?`, - summary.Status, summary.RequestVersion, summary.StatusTimestampMs, - newVersion, summary.LastError, metadataJSON, + summary.Queue, changeURIsJSON, summary.ReceivedAtMs, summary.Status, + summary.RequestVersion, summary.StatusTimestampMs, newVersion, + summary.LastError, metadataJSON, summary.RequestID, oldVersion, ) if err != nil { diff --git a/submitqueue/extension/storage/mysql/request_summary_store_test.go b/submitqueue/extension/storage/mysql/request_summary_store_test.go index 9399794b..de1f03bf 100644 --- a/submitqueue/extension/storage/mysql/request_summary_store_test.go +++ b/submitqueue/extension/storage/mysql/request_summary_store_test.go @@ -199,51 +199,93 @@ func TestRequestSummaryStore_Get(t *testing.T) { func TestRequestSummaryStore_Update(t *testing.T) { summary := entity.RequestSummary{ RequestID: "monorepo/1", - Queue: "monorepo", - ReceivedAtMs: 1000, + Queue: "monorepo-updated", + ChangeURIs: []string{"github://github.example.com/uber/submitqueue/pull/456/cafebabe"}, + ReceivedAtMs: 1500, Status: entity.RequestStatusValidated, RequestVersion: 2, StatusTimestampMs: 2000, - LastError: "", + Version: 1, + LastError: "validation detail", + Metadata: map[string]string{"result": "validated"}, } const oldVersion, newVersion = int32(1), int32(2) tests := []struct { name string + summary entity.RequestSummary setup func(mock sqlmock.Sqlmock) wantErr bool wantErrIs error }{ { - name: "success", + name: "success", + summary: summary, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request_summary"). - WithArgs(summary.Status, summary.RequestVersion, summary.StatusTimestampMs, - newVersion, summary.LastError, sqlmock.AnyArg(), summary.RequestID, oldVersion). + WithArgs(summary.Queue, []byte(`["github://github.example.com/uber/submitqueue/pull/456/cafebabe"]`), + summary.ReceivedAtMs, summary.Status, summary.RequestVersion, summary.StatusTimestampMs, + newVersion, summary.LastError, []byte(`{"result":"validated"}`), summary.RequestID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, { - name: "version mismatch", + name: "version mismatch", + summary: summary, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request_summary"). - WithArgs(summary.Status, summary.RequestVersion, summary.StatusTimestampMs, - newVersion, summary.LastError, sqlmock.AnyArg(), summary.RequestID, oldVersion). + WithArgs(summary.Queue, []byte(`["github://github.example.com/uber/submitqueue/pull/456/cafebabe"]`), + summary.ReceivedAtMs, summary.Status, summary.RequestVersion, summary.StatusTimestampMs, + newVersion, summary.LastError, []byte(`{"result":"validated"}`), summary.RequestID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 0)) }, wantErr: true, wantErrIs: storage.ErrVersionMismatch, }, { - name: "exec error", + name: "exec error", + summary: summary, setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE request_summary"). - WithArgs(summary.Status, summary.RequestVersion, summary.StatusTimestampMs, - newVersion, summary.LastError, sqlmock.AnyArg(), summary.RequestID, oldVersion). + WithArgs(summary.Queue, []byte(`["github://github.example.com/uber/submitqueue/pull/456/cafebabe"]`), + summary.ReceivedAtMs, summary.Status, summary.RequestVersion, summary.StatusTimestampMs, + newVersion, summary.LastError, []byte(`{"result":"validated"}`), summary.RequestID, oldVersion). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, }, + { + name: "rows affected error", + summary: summary, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("UPDATE request_summary"). + WithArgs(summary.Queue, []byte(`["github://github.example.com/uber/submitqueue/pull/456/cafebabe"]`), + summary.ReceivedAtMs, summary.Status, summary.RequestVersion, summary.StatusTimestampMs, + newVersion, summary.LastError, []byte(`{"result":"validated"}`), summary.RequestID, oldVersion). + WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("rows unavailable"))) + }, + wantErr: true, + }, + { + name: "nil collections normalize to empty JSON", + summary: entity.RequestSummary{ + RequestID: summary.RequestID, + Queue: summary.Queue, + ReceivedAtMs: summary.ReceivedAtMs, + Status: summary.Status, + RequestVersion: summary.RequestVersion, + StatusTimestampMs: summary.StatusTimestampMs, + Version: summary.Version, + LastError: summary.LastError, + }, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("UPDATE request_summary"). + WithArgs(summary.Queue, []byte(`[]`), summary.ReceivedAtMs, summary.Status, + summary.RequestVersion, summary.StatusTimestampMs, newVersion, summary.LastError, + []byte(`{}`), summary.RequestID, oldVersion). + WillReturnResult(sqlmock.NewResult(0, 1)) + }, + }, } for _, tt := range tests { @@ -253,7 +295,7 @@ func TestRequestSummaryStore_Update(t *testing.T) { tt.setup(mock) - err := store.Update(context.Background(), summary, oldVersion, newVersion) + err := store.Update(context.Background(), tt.summary, oldVersion, newVersion) if tt.wantErr { require.Error(t, err) if tt.wantErrIs != nil { diff --git a/submitqueue/extension/storage/request_summary_store.go b/submitqueue/extension/storage/request_summary_store.go index 5e2a657a..faf2cb3f 100644 --- a/submitqueue/extension/storage/request_summary_store.go +++ b/submitqueue/extension/storage/request_summary_store.go @@ -31,7 +31,7 @@ type RequestSummaryStore interface { // Get returns the summary for requestID, or ErrNotFound when absent. Get(ctx context.Context, requestID string) (entity.RequestSummary, error) - // Update conditionally replaces the mutable status fields when the persisted projection version equals oldVersion. + // Update conditionally replaces every non-key field when the persisted projection version equals oldVersion. // The store writes newVersion exactly as supplied and returns ErrVersionMismatch when the guard does not match. Update(ctx context.Context, summary entity.RequestSummary, oldVersion, newVersion int32) error } diff --git a/test/integration/submitqueue/extension/storage/suite.go b/test/integration/submitqueue/extension/storage/suite.go index c5fc2280..7306f0fe 100644 --- a/test/integration/submitqueue/extension/storage/suite.go +++ b/test/integration/submitqueue/extension/storage/suite.go @@ -442,33 +442,80 @@ func (s *StorageContractSuite) TestStorage_RequestSummaryCreateGetAndCAS() { ctx := s.ctx summary := entity.RequestSummary{ RequestID: "summary/1", Queue: "summary-q", ChangeURIs: nil, ReceivedAtMs: 100, - Status: entity.RequestStatusAccepted, StatusTimestampMs: 100, Version: 1, Metadata: nil, + Status: entity.RequestStatusAccepted, RequestVersion: 1, StatusTimestampMs: 100, Version: 1, + LastError: "", Metadata: nil, } + store := s.storage.GetRequestSummaryStore() - require.NoError(t, s.storage.GetRequestSummaryStore().Create(ctx, summary)) - require.ErrorIs(t, s.storage.GetRequestSummaryStore().Create(ctx, summary), storage.ErrAlreadyExists) + require.NoError(t, store.Create(ctx, summary)) + require.ErrorIs(t, store.Create(ctx, summary), storage.ErrAlreadyExists) - got, err := s.storage.GetRequestSummaryStore().Get(ctx, summary.RequestID) + got, err := store.Get(ctx, summary.RequestID) require.NoError(t, err) - assert.NotNil(t, got.ChangeURIs) - assert.NotNil(t, got.Metadata) - _, err = s.storage.GetRequestSummaryStore().Get(ctx, "summary/missing") + assert.Equal(t, []string{}, got.ChangeURIs) + assert.Equal(t, map[string]string{}, got.Metadata) + _, err = store.Get(ctx, "summary/missing") require.ErrorIs(t, err, storage.ErrNotFound) + got.Queue = "summary-q-updated" + got.ChangeURIs = []string{"change/updated"} + got.ReceivedAtMs = 200 got.Status = entity.RequestStatusLanded got.RequestVersion = 2 - got.StatusTimestampMs = 200 + got.StatusTimestampMs = 300 got.LastError = "terminal detail" got.Metadata = map[string]string{"source": "test"} - require.NoError(t, s.storage.GetRequestSummaryStore().Update(ctx, got, 1, 2)) - require.ErrorIs(t, s.storage.GetRequestSummaryStore().Update(ctx, got, 1, 3), storage.ErrVersionMismatch) + require.NoError(t, store.Update(ctx, got, 1, 2)) - updated, err := s.storage.GetRequestSummaryStore().Get(ctx, summary.RequestID) + updated, err := store.Get(ctx, summary.RequestID) require.NoError(t, err) - assert.Equal(t, int32(2), updated.Version) - assert.Equal(t, entity.RequestStatusLanded, updated.Status) - assert.Equal(t, "terminal detail", updated.LastError) - assert.Equal(t, map[string]string{"source": "test"}, updated.Metadata) + assert.Equal(t, entity.RequestSummary{ + RequestID: summary.RequestID, + Queue: "summary-q-updated", + ChangeURIs: []string{"change/updated"}, + ReceivedAtMs: 200, + Status: entity.RequestStatusLanded, + RequestVersion: 2, + StatusTimestampMs: 300, + Version: 2, + LastError: "terminal detail", + Metadata: map[string]string{"source": "test"}, + }, updated) + + stale := updated + stale.Queue = "stale-q" + stale.ChangeURIs = []string{"change/stale"} + stale.ReceivedAtMs = 400 + stale.Status = entity.RequestStatusError + stale.RequestVersion = 3 + stale.StatusTimestampMs = 500 + stale.LastError = "stale detail" + stale.Metadata = map[string]string{"source": "stale"} + require.ErrorIs(t, store.Update(ctx, stale, 1, 3), storage.ErrVersionMismatch) + + afterStale, err := store.Get(ctx, summary.RequestID) + require.NoError(t, err) + assert.Equal(t, updated, afterStale) + + updated.ChangeURIs = nil + updated.Metadata = nil + require.NoError(t, store.Update(ctx, updated, 2, 3)) + + normalizedNil, err := store.Get(ctx, summary.RequestID) + require.NoError(t, err) + assert.Equal(t, []string{}, normalizedNil.ChangeURIs) + assert.Equal(t, map[string]string{}, normalizedNil.Metadata) + assert.Equal(t, int32(3), normalizedNil.Version) + + normalizedNil.ChangeURIs = []string{} + normalizedNil.Metadata = map[string]string{} + require.NoError(t, store.Update(ctx, normalizedNil, 3, 4)) + + normalizedEmpty, err := store.Get(ctx, summary.RequestID) + require.NoError(t, err) + assert.Equal(t, []string{}, normalizedEmpty.ChangeURIs) + assert.Equal(t, map[string]string{}, normalizedEmpty.Metadata) + assert.Equal(t, int32(4), normalizedEmpty.Version) } func (s *StorageContractSuite) TestStorage_RequestQueueSummaryListAndCursor() {