Skip to content

refactor(storage): replace complete request summaries #495

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ubettigole wants to merge 1 commit into from
+127 −37
Open

refactor(storage): replace complete request summaries #495

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
17 changes: 9 additions & 8 deletions submitqueue/extension/storage/mysql/request_summary_store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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, `
Expand Down Expand Up @@ -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 {
Expand Down
68 changes: 55 additions & 13 deletions submitqueue/extension/storage/mysql/request_summary_store_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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 {
Expand Down
2 changes: 1 addition & 1 deletion submitqueue/extension/storage/request_summary_store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
77 changes: 62 additions & 15 deletions test/integration/submitqueue/extension/storage/suite.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down
Loading