From 31c18b4fbc0949cb1ca0cccd17faea1fe37c01c1 Mon Sep 17 00:00:00 2001 From: abettigole Date: Fri, 31 Jul 2026 23:44:31 +0000 Subject: [PATCH] refactor(storage): replace complete queue summaries Persist change URIs with the existing RequestQueueSummary replacement update and expand guarded round-trip coverage. Jira Issues CODEM-204 --- .../mysql/request_queue_summary_store.go | 11 ++- .../mysql/request_queue_summary_store_test.go | 77 ++++++++++++++++--- .../storage/request_queue_summary_store.go | 2 +- .../submitqueue/extension/storage/suite.go | 36 ++++++++- 4 files changed, 107 insertions(+), 19 deletions(-) diff --git a/submitqueue/extension/storage/mysql/request_queue_summary_store.go b/submitqueue/extension/storage/mysql/request_queue_summary_store.go index 3e59b9fd..2b4f537e 100644 --- a/submitqueue/extension/storage/mysql/request_queue_summary_store.go +++ b/submitqueue/extension/storage/mysql/request_queue_summary_store.go @@ -17,7 +17,6 @@ package mysql import ( "context" "database/sql" - "encoding/json" "errors" "fmt" @@ -45,7 +44,7 @@ func (s *requestQueueSummaryStore) Create(ctx context.Context, summary entity.Re changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary.ChangeURIs, summary.Metadata) if err != nil { - return fmt.Errorf("failed to marshal queue summary request_id=%s: %w", summary.RequestID, err) + return fmt.Errorf("failed to marshal queue summary metadata request_id=%s: %w", summary.RequestID, err) } _, err = s.db.ExecContext(ctx, ` INSERT INTO request_summary_by_queue ( @@ -93,15 +92,15 @@ func (s *requestQueueSummaryStore) Update(ctx context.Context, summary entity.Re op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets) defer func() { op.Complete(retErr) }() - metadataJSON, err := json.Marshal(normalizeMetadata(summary.Metadata)) + changeURIsJSON, metadataJSON, err := marshalSummaryJSON(summary.ChangeURIs, summary.Metadata) if err != nil { - return fmt.Errorf("failed to marshal queue summary metadata request_id=%s: %w", summary.RequestID, err) + return fmt.Errorf("failed to marshal queue summary request_id=%s: %w", summary.RequestID, err) } result, err := s.db.ExecContext(ctx, ` UPDATE request_summary_by_queue - SET status = ?, version = ?, last_error = ?, metadata = ? + SET change_uris = ?, status = ?, version = ?, last_error = ?, metadata = ? WHERE queue = ? AND received_at_ms = ? AND request_id = ? AND version = ?`, - summary.Status, newVersion, summary.LastError, metadataJSON, + changeURIsJSON, summary.Status, newVersion, summary.LastError, metadataJSON, summary.Queue, summary.ReceivedAtMs, summary.RequestID, oldVersion, ) if err != nil { diff --git a/submitqueue/extension/storage/mysql/request_queue_summary_store_test.go b/submitqueue/extension/storage/mysql/request_queue_summary_store_test.go index 1f04ba1e..8643030c 100644 --- a/submitqueue/extension/storage/mysql/request_queue_summary_store_test.go +++ b/submitqueue/extension/storage/mysql/request_queue_summary_store_test.go @@ -185,32 +185,73 @@ func TestRequestQueueSummaryStore_Update(t *testing.T) { summary := entity.RequestQueueSummary{ RequestID: "monorepo/1", Queue: "monorepo", + ChangeURIs: []string{"github://github.example.com/uber/submitqueue/pull/456/cafebabe"}, ReceivedAtMs: 1000, Status: entity.RequestStatusValidated, - LastError: "", + LastError: "validation detail", + Metadata: map[string]string{"result": "validated"}, } const oldVersion, newVersion = int32(1), int32(2) tests := []struct { name string + summary entity.RequestQueueSummary setup func(mock sqlmock.Sqlmock) wantErr bool wantErrIs error }{ { - name: "success", + name: "success replaces all non-key fields", + summary: summary, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("SET change_uris = \\?, status = \\?, version = \\?, last_error = \\?, metadata = \\?"). + WithArgs([]byte(`["github://github.example.com/uber/submitqueue/pull/456/cafebabe"]`), + summary.Status, newVersion, summary.LastError, []byte(`{"result":"validated"}`), + summary.Queue, summary.ReceivedAtMs, summary.RequestID, oldVersion). + WillReturnResult(sqlmock.NewResult(0, 1)) + }, + }, + { + name: "success normalizes nil collections", + summary: entity.RequestQueueSummary{ + RequestID: summary.RequestID, + Queue: summary.Queue, + ReceivedAtMs: summary.ReceivedAtMs, + Status: summary.Status, + LastError: summary.LastError, + }, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("SET change_uris = \\?, status = \\?, version = \\?, last_error = \\?, metadata = \\?"). + WithArgs([]byte(`[]`), summary.Status, newVersion, summary.LastError, []byte(`{}`), + summary.Queue, summary.ReceivedAtMs, summary.RequestID, oldVersion). + WillReturnResult(sqlmock.NewResult(0, 1)) + }, + }, + { + name: "success persists empty collections", + summary: entity.RequestQueueSummary{ + RequestID: summary.RequestID, + Queue: summary.Queue, + ChangeURIs: []string{}, + ReceivedAtMs: summary.ReceivedAtMs, + Status: summary.Status, + LastError: summary.LastError, + Metadata: map[string]string{}, + }, setup: func(mock sqlmock.Sqlmock) { - mock.ExpectExec("UPDATE request_summary_by_queue"). - WithArgs(summary.Status, newVersion, summary.LastError, sqlmock.AnyArg(), + mock.ExpectExec("SET change_uris = \\?, status = \\?, version = \\?, last_error = \\?, metadata = \\?"). + WithArgs([]byte(`[]`), summary.Status, newVersion, summary.LastError, []byte(`{}`), summary.Queue, summary.ReceivedAtMs, 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_by_queue"). - WithArgs(summary.Status, newVersion, summary.LastError, sqlmock.AnyArg(), + mock.ExpectExec("SET change_uris = \\?, status = \\?, version = \\?, last_error = \\?, metadata = \\?"). + WithArgs([]byte(`["github://github.example.com/uber/submitqueue/pull/456/cafebabe"]`), + summary.Status, newVersion, summary.LastError, []byte(`{"result":"validated"}`), summary.Queue, summary.ReceivedAtMs, summary.RequestID, oldVersion). WillReturnResult(sqlmock.NewResult(0, 0)) }, @@ -218,15 +259,29 @@ func TestRequestQueueSummaryStore_Update(t *testing.T) { wantErrIs: storage.ErrVersionMismatch, }, { - name: "exec error", + name: "exec error", + summary: summary, setup: func(mock sqlmock.Sqlmock) { - mock.ExpectExec("UPDATE request_summary_by_queue"). - WithArgs(summary.Status, newVersion, summary.LastError, sqlmock.AnyArg(), + mock.ExpectExec("SET change_uris = \\?, status = \\?, version = \\?, last_error = \\?, metadata = \\?"). + WithArgs([]byte(`["github://github.example.com/uber/submitqueue/pull/456/cafebabe"]`), + summary.Status, newVersion, summary.LastError, []byte(`{"result":"validated"}`), summary.Queue, summary.ReceivedAtMs, summary.RequestID, oldVersion). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, }, + { + name: "rows affected error", + summary: summary, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("SET change_uris = \\?, status = \\?, version = \\?, last_error = \\?, metadata = \\?"). + WithArgs([]byte(`["github://github.example.com/uber/submitqueue/pull/456/cafebabe"]`), + summary.Status, newVersion, summary.LastError, []byte(`{"result":"validated"}`), + summary.Queue, summary.ReceivedAtMs, summary.RequestID, oldVersion). + WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error"))) + }, + wantErr: true, + }, } for _, tt := range tests { @@ -236,7 +291,7 @@ func TestRequestQueueSummaryStore_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_queue_summary_store.go b/submitqueue/extension/storage/request_queue_summary_store.go index 68acf211..d2a7debc 100644 --- a/submitqueue/extension/storage/request_queue_summary_store.go +++ b/submitqueue/extension/storage/request_queue_summary_store.go @@ -54,7 +54,7 @@ type RequestQueueSummaryStore interface { // Get returns the row identified by its full primary key, or ErrNotFound when absent. Get(ctx context.Context, queue string, receivedAtMs int64, requestID string) (entity.RequestQueueSummary, error) - // Update conditionally replaces mutable fields when the persisted projection version equals oldVersion. + // Update conditionally replaces all non-key fields 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.RequestQueueSummary, oldVersion, newVersion int32) error diff --git a/test/integration/submitqueue/extension/storage/suite.go b/test/integration/submitqueue/extension/storage/suite.go index 7306f0fe..edaa3528 100644 --- a/test/integration/submitqueue/extension/storage/suite.go +++ b/test/integration/submitqueue/extension/storage/suite.go @@ -540,15 +540,49 @@ func (s *StorageContractSuite) TestStorage_RequestQueueSummaryListAndCursor() { require.ErrorIs(t, err, storage.ErrNotFound) got.Status = entity.RequestStatusLanded + got.ChangeURIs = []string{"uri/replacement/1", "uri/replacement/2"} got.LastError = "done" got.Metadata = map[string]string{"result": "landed"} require.NoError(t, store.Update(ctx, got, 1, 2)) - require.ErrorIs(t, store.Update(ctx, got, 1, 3), storage.ErrVersionMismatch) updated, err := store.Get(ctx, got.Queue, got.ReceivedAtMs, got.RequestID) require.NoError(t, err) assert.Equal(t, int32(2), updated.Version) + assert.Equal(t, []string{"uri/replacement/1", "uri/replacement/2"}, updated.ChangeURIs) assert.Equal(t, entity.RequestStatusLanded, updated.Status) assert.Equal(t, "done", updated.LastError) + assert.Equal(t, map[string]string{"result": "landed"}, updated.Metadata) + + stale := updated + stale.ChangeURIs = []string{} + stale.Status = entity.RequestStatusError + stale.LastError = "stale" + stale.Metadata = map[string]string{} + require.ErrorIs(t, store.Update(ctx, stale, 1, 3), storage.ErrVersionMismatch) + unchanged, err := store.Get(ctx, got.Queue, got.ReceivedAtMs, got.RequestID) + require.NoError(t, err) + assert.Equal(t, updated, unchanged) + + updated.ChangeURIs = nil + updated.Metadata = nil + require.NoError(t, store.Update(ctx, updated, 2, 3)) + normalized, err := store.Get(ctx, got.Queue, got.ReceivedAtMs, got.RequestID) + require.NoError(t, err) + assert.Equal(t, int32(3), normalized.Version) + assert.NotNil(t, normalized.ChangeURIs) + assert.Empty(t, normalized.ChangeURIs) + assert.NotNil(t, normalized.Metadata) + assert.Empty(t, normalized.Metadata) + + normalized.ChangeURIs = []string{} + normalized.Metadata = map[string]string{} + require.NoError(t, store.Update(ctx, normalized, 3, 4)) + emptyCollections, err := store.Get(ctx, got.Queue, got.ReceivedAtMs, got.RequestID) + require.NoError(t, err) + assert.Equal(t, int32(4), emptyCollections.Version) + assert.NotNil(t, emptyCollections.ChangeURIs) + assert.Empty(t, emptyCollections.ChangeURIs) + assert.NotNil(t, emptyCollections.Metadata) + assert.Empty(t, emptyCollections.Metadata) firstPage, err := store.List(ctx, storage.RequestQueueSummaryQuery{ Queue: "queue-summary", ReceivedAtOrAfterMs: 50, ReceivedBeforeMs: 250, Limit: 2,