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

Author: ubettigole

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 {

submitqueue/extension/storage/mysql/request_queue_summary_store_test.go

@@ -185,48 +185,103 @@ 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))
 			},
 			wantErr:   true,
 			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 {

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
 

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,