From 1d53d9f0e0e7db09273818a167ee0f9ed78a46f9 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Mon, 31 Aug 2026 22:16:26 +0000 Subject: [PATCH] feat(stovepipe): persist request log records --- stovepipe/entity/BUILD.bazel | 2 + stovepipe/entity/request_log.go | 153 ++++++++ stovepipe/entity/request_log_test.go | 195 +++++++++++ stovepipe/extension/storage/BUILD.bazel | 1 + stovepipe/extension/storage/mock/BUILD.bazel | 1 + .../storage/mock/request_log_store_mock.go | 86 +++++ .../extension/storage/mock/storage_mock.go | 14 + stovepipe/extension/storage/mysql/BUILD.bazel | 2 + .../storage/mysql/request_log_store.go | 168 +++++++++ .../storage/mysql/request_log_store_test.go | 331 ++++++++++++++++++ stovepipe/extension/storage/mysql/storage.go | 7 + .../extension/storage/mysql/storage_test.go | 1 + .../extension/storage/request_log_store.go | 35 ++ stovepipe/extension/storage/storage.go | 3 + .../extension/storage/mysql/storage_test.go | 13 + .../stovepipe/extension/storage/suite.go | 121 +++++++ 16 files changed, 1133 insertions(+) create mode 100644 stovepipe/entity/request_log.go create mode 100644 stovepipe/entity/request_log_test.go create mode 100644 stovepipe/extension/storage/mock/request_log_store_mock.go create mode 100644 stovepipe/extension/storage/mysql/request_log_store.go create mode 100644 stovepipe/extension/storage/mysql/request_log_store_test.go create mode 100644 stovepipe/extension/storage/request_log_store.go diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index 18eeeb9c7..19518f9e1 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -9,6 +9,7 @@ go_library( "queue_config.go", "request.go", "request_id.go", + "request_log.go", "validation_fact.go", ], importpath = "github.com/uber/submitqueue/stovepipe/entity", @@ -20,6 +21,7 @@ go_test( srcs = [ "build_test.go", "request_id_test.go", + "request_log_test.go", "request_test.go", "validation_fact_test.go", ], diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go new file mode 100644 index 000000000..2da89ed27 --- /dev/null +++ b/stovepipe/entity/request_log.go @@ -0,0 +1,153 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package entity + +import "fmt" + +// RequestEvent identifies a retained occurrence that does not change request state. +type RequestEvent string + +const ( + // RequestEventUnknown is the unset event value. + RequestEventUnknown RequestEvent = "" + // RequestEventBuildTriggered records that a build was durably accepted. + RequestEventBuildTriggered RequestEvent = "build_triggered" + // RequestEventBuildFinished records that a build first reached a terminal status. + RequestEventBuildFinished RequestEvent = "build_finished" + // RequestEventValidationFactRecorded records that an immutable validation verdict was established. + RequestEventValidationFactRecorded RequestEvent = "validation_fact_recorded" +) + +// RequestOutcomeReason identifies the durable domain reason for a terminal request state. +type RequestOutcomeReason string + +const ( + // RequestOutcomeReasonUnknown is the unset outcome reason. + RequestOutcomeReasonUnknown RequestOutcomeReason = "" + // RequestOutcomeReasonBuildSucceeded indicates that the request's build succeeded. + RequestOutcomeReasonBuildSucceeded RequestOutcomeReason = "build_succeeded" + // RequestOutcomeReasonBuildFailed indicates that the request's build failed. + RequestOutcomeReasonBuildFailed RequestOutcomeReason = "build_failed" + // RequestOutcomeReasonBuildCancelled indicates that the request's build was cancelled. + RequestOutcomeReasonBuildCancelled RequestOutcomeReason = "build_cancelled" + // RequestOutcomeReasonProcessingFailed indicates that validation could not be prepared. + RequestOutcomeReasonProcessingFailed RequestOutcomeReason = "processing_failed" + // RequestOutcomeReasonBuildPollingExhausted indicates that build status could not be resolved. + RequestOutcomeReasonBuildPollingExhausted RequestOutcomeReason = "build_polling_exhausted" + // RequestOutcomeReasonValidationTimeout indicates that validation exceeded its allowed duration. + RequestOutcomeReasonValidationTimeout RequestOutcomeReason = "validation_timeout" + // RequestOutcomeReasonSupersededByNewerHead indicates that a newer request replaced this one. + RequestOutcomeReasonSupersededByNewerHead RequestOutcomeReason = "superseded_by_newer_head" +) + +// RequestLog is one immutable request state or explanatory lifecycle occurrence. +type RequestLog struct { + // ID is the stable opaque identity of the occurrence within the request. + ID string `json:"id"` + // Queue is the logical queue containing the request and scopes RequestID. + Queue string `json:"queue"` + // RequestID identifies the request whose log contains this record. + RequestID string `json:"request_id"` + // TimestampMs is the occurrence time in Unix milliseconds. + TimestampMs int64 `json:"timestamp_ms"` + // State is the durable request state recorded by a state record and is unset on an event record. + State RequestState `json:"state"` + // Event identifies the occurrence recorded by an event record and is unset on a state record. + Event RequestEvent `json:"event"` + // RequestVersion is the durable request version recorded by a state record and is zero on an event record. + RequestVersion int32 `json:"request_version"` + // OutcomeReason is the durable domain reason for a terminal request state and is otherwise unset. + OutcomeReason RequestOutcomeReason `json:"outcome_reason"` + // Metadata contains optional occurrence context; nil and empty maps are equivalent. + Metadata map[string]string `json:"metadata"` +} + +// Validate verifies the invariants required for a newly persisted request log. +func (e RequestLog) Validate() error { + if e.ID == "" { + return fmt.Errorf("request log ID must not be empty") + } + if e.Queue == "" { + return fmt.Errorf("request log queue must not be empty") + } + if e.RequestID == "" { + return fmt.Errorf("request log request ID must not be empty") + } + if e.TimestampMs <= 0 { + return fmt.Errorf("request log timestamp must be positive") + } + if (e.State == RequestStateUnknown) == (e.Event == RequestEventUnknown) { + return fmt.Errorf("request log must contain exactly one of state and event") + } + if e.State != RequestStateUnknown { + return e.validateState() + } + return e.validateEvent() +} + +func (e RequestLog) validateState() error { + if e.RequestVersion <= 0 { + return fmt.Errorf("state log must have a positive request version") + } + switch e.State { + case RequestStateAccepted, RequestStateProcessing: + if e.OutcomeReason != RequestOutcomeReasonUnknown { + return fmt.Errorf("non-terminal state log must not contain terminal context") + } + case RequestStateSuperseded: + if e.OutcomeReason != RequestOutcomeReasonSupersededByNewerHead { + return fmt.Errorf("superseded state log has invalid outcome context") + } + case RequestStateSucceeded: + if e.OutcomeReason != RequestOutcomeReasonBuildSucceeded { + return fmt.Errorf("succeeded state log has invalid outcome context") + } + case RequestStateFailed: + if !isFailureReason(e.OutcomeReason) { + return fmt.Errorf("failed state log has invalid outcome context") + } + case RequestStateCancelled: + if e.OutcomeReason != RequestOutcomeReasonBuildCancelled { + return fmt.Errorf("cancelled state log has invalid outcome context") + } + default: + return fmt.Errorf("unknown request state %q", e.State) + } + return nil +} + +func (e RequestLog) validateEvent() error { + if e.RequestVersion != 0 || e.OutcomeReason != RequestOutcomeReasonUnknown { + return fmt.Errorf("event log must not contain request-state context") + } + switch e.Event { + case RequestEventBuildTriggered, RequestEventBuildFinished, RequestEventValidationFactRecorded: + default: + return fmt.Errorf("unknown request event %q", e.Event) + } + return nil +} + +func isFailureReason(reason RequestOutcomeReason) bool { + switch reason { + case RequestOutcomeReasonBuildFailed, + RequestOutcomeReasonProcessingFailed, + RequestOutcomeReasonBuildPollingExhausted, + RequestOutcomeReasonValidationTimeout: + return true + default: + return false + } +} diff --git a/stovepipe/entity/request_log_test.go b/stovepipe/entity/request_log_test.go new file mode 100644 index 000000000..b44d0ed3d --- /dev/null +++ b/stovepipe/entity/request_log_test.go @@ -0,0 +1,195 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package entity + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRequestLogValidate(t *testing.T) { + base := RequestLog{ + ID: "state/1", + Queue: "monorepo/main", + RequestID: "request/monorepo/main/1", + TimestampMs: 1735689600000, + State: RequestStateAccepted, + RequestVersion: 1, + } + + tests := []struct { + name string + mutate func(RequestLog) RequestLog + wantErr bool + }{ + {name: "accepted state"}, + { + name: "superseded state", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateSuperseded + entry.Metadata = map[string]string{"superseded_by_request_id": "request/monorepo/main/2"} + entry.OutcomeReason = RequestOutcomeReasonSupersededByNewerHead + return entry + }, + }, + { + name: "failed without build", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateFailed + entry.OutcomeReason = RequestOutcomeReasonProcessingFailed + return entry + }, + }, + { + name: "succeeded state", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateSucceeded + entry.Metadata = map[string]string{"build_id": "42"} + entry.OutcomeReason = RequestOutcomeReasonBuildSucceeded + return entry + }, + }, + { + name: "failed build state", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateFailed + entry.Metadata = map[string]string{"build_id": "42"} + entry.OutcomeReason = RequestOutcomeReasonBuildFailed + return entry + }, + }, + { + name: "failed polling state", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateFailed + entry.OutcomeReason = RequestOutcomeReasonBuildPollingExhausted + return entry + }, + }, + { + name: "failed timeout state", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateFailed + entry.OutcomeReason = RequestOutcomeReasonValidationTimeout + return entry + }, + }, + { + name: "cancelled state", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateCancelled + entry.Metadata = map[string]string{"build_id": "42"} + entry.OutcomeReason = RequestOutcomeReasonBuildCancelled + return entry + }, + }, + { + name: "validation fact with green degree", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEventValidationFactRecorded + entry.RequestVersion = 0 + entry.Metadata = map[string]string{"fact_degree": "0"} + return entry + }, + }, + { + name: "build event", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEventBuildTriggered + entry.RequestVersion = 0 + entry.Metadata = map[string]string{"build_id": "42"} + return entry + }, + }, + { + name: "build finished event", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEventBuildFinished + entry.RequestVersion = 0 + entry.Metadata = map[string]string{"build_id": "42"} + return entry + }, + }, + {name: "missing ID", mutate: func(entry RequestLog) RequestLog { entry.ID = ""; return entry }, wantErr: true}, + {name: "missing queue", mutate: func(entry RequestLog) RequestLog { entry.Queue = ""; return entry }, wantErr: true}, + {name: "missing request ID", mutate: func(entry RequestLog) RequestLog { entry.RequestID = ""; return entry }, wantErr: true}, + {name: "non-positive timestamp", mutate: func(entry RequestLog) RequestLog { entry.TimestampMs = 0; return entry }, wantErr: true}, + {name: "missing occurrence", mutate: func(entry RequestLog) RequestLog { entry.State = RequestStateUnknown; return entry }, wantErr: true}, + {name: "two occurrences", mutate: func(entry RequestLog) RequestLog { + entry.Event = RequestEventBuildTriggered + return entry + }, wantErr: true}, + {name: "state without version", mutate: func(entry RequestLog) RequestLog { entry.RequestVersion = 0; return entry }, wantErr: true}, + {name: "unknown state", mutate: func(entry RequestLog) RequestLog { + entry.State = RequestState("future") + return entry + }, wantErr: true}, + {name: "non-terminal state with outcome", mutate: func(entry RequestLog) RequestLog { + entry.OutcomeReason = RequestOutcomeReasonProcessingFailed + return entry + }, wantErr: true}, + {name: "opaque metadata", mutate: func(entry RequestLog) RequestLog { + entry.Metadata = map[string]string{"arbitrary": "value"} + return entry + }}, + { + name: "failed without reason", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateFailed + return entry + }, + wantErr: true, + }, + { + name: "event with request version", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEventBuildTriggered + entry.Metadata = map[string]string{"build_id": "42"} + return entry + }, + wantErr: true, + }, + { + name: "unknown event", + mutate: func(entry RequestLog) RequestLog { + entry.State = RequestStateUnknown + entry.Event = RequestEvent("future") + entry.RequestVersion = 0 + return entry + }, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + entry := base + if tt.mutate != nil { + entry = tt.mutate(entry) + } + err := entry.Validate() + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + }) + } +} diff --git a/stovepipe/extension/storage/BUILD.bazel b/stovepipe/extension/storage/BUILD.bazel index 5eb02a5b6..02056713c 100644 --- a/stovepipe/extension/storage/BUILD.bazel +++ b/stovepipe/extension/storage/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "build_store.go", "queue_store.go", + "request_log_store.go", "request_store.go", "request_uri_store.go", "storage.go", diff --git a/stovepipe/extension/storage/mock/BUILD.bazel b/stovepipe/extension/storage/mock/BUILD.bazel index bed9de9c2..f1796eef6 100644 --- a/stovepipe/extension/storage/mock/BUILD.bazel +++ b/stovepipe/extension/storage/mock/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "build_store_mock.go", "queue_store_mock.go", + "request_log_store_mock.go", "request_store_mock.go", "request_uri_store_mock.go", "storage_mock.go", diff --git a/stovepipe/extension/storage/mock/request_log_store_mock.go b/stovepipe/extension/storage/mock/request_log_store_mock.go new file mode 100644 index 000000000..4b77dd66a --- /dev/null +++ b/stovepipe/extension/storage/mock/request_log_store_mock.go @@ -0,0 +1,86 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: request_log_store.go +// +// Generated by this command: +// +// mockgen -source=request_log_store.go -destination=mock/request_log_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/stovepipe/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockRequestLogStore is a mock of RequestLogStore interface. +type MockRequestLogStore struct { + ctrl *gomock.Controller + recorder *MockRequestLogStoreMockRecorder + isgomock struct{} +} + +// MockRequestLogStoreMockRecorder is the mock recorder for MockRequestLogStore. +type MockRequestLogStoreMockRecorder struct { + mock *MockRequestLogStore +} + +// NewMockRequestLogStore creates a new mock instance. +func NewMockRequestLogStore(ctrl *gomock.Controller) *MockRequestLogStore { + mock := &MockRequestLogStore{ctrl: ctrl} + mock.recorder = &MockRequestLogStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRequestLogStore) EXPECT() *MockRequestLogStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockRequestLogStore) Create(ctx context.Context, log entity.RequestLog) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, log) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockRequestLogStoreMockRecorder) Create(ctx, log any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestLogStore)(nil).Create), ctx, log) +} + +// Get mocks base method. +func (m *MockRequestLogStore) Get(ctx context.Context, requestID, logID string) (entity.RequestLog, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, requestID, logID) + ret0, _ := ret[0].(entity.RequestLog) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockRequestLogStoreMockRecorder) Get(ctx, requestID, logID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockRequestLogStore)(nil).Get), ctx, requestID, logID) +} + +// List mocks base method. +func (m *MockRequestLogStore) List(ctx context.Context, requestID string) ([]entity.RequestLog, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, requestID) + ret0, _ := ret[0].([]entity.RequestLog) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockRequestLogStoreMockRecorder) List(ctx, requestID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockRequestLogStore)(nil).List), ctx, requestID) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index c73f78835..eb901e6f4 100644 --- a/stovepipe/extension/storage/mock/storage_mock.go +++ b/stovepipe/extension/storage/mock/storage_mock.go @@ -107,6 +107,20 @@ func (mr *MockStorageMockRecorder) GetQueueStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetQueueStore", reflect.TypeOf((*MockStorage)(nil).GetQueueStore)) } +// GetRequestLogStore mocks base method. +func (m *MockStorage) GetRequestLogStore() storage.RequestLogStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestLogStore") + ret0, _ := ret[0].(storage.RequestLogStore) + return ret0 +} + +// GetRequestLogStore indicates an expected call of GetRequestLogStore. +func (mr *MockStorageMockRecorder) GetRequestLogStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestLogStore", reflect.TypeOf((*MockStorage)(nil).GetRequestLogStore)) +} + // GetRequestStore mocks base method. func (m *MockStorage) GetRequestStore() storage.RequestStore { m.ctrl.T.Helper() diff --git a/stovepipe/extension/storage/mysql/BUILD.bazel b/stovepipe/extension/storage/mysql/BUILD.bazel index 3b771cba6..bc875e5d1 100644 --- a/stovepipe/extension/storage/mysql/BUILD.bazel +++ b/stovepipe/extension/storage/mysql/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "build_store.go", "queue_store.go", + "request_log_store.go", "request_store.go", "request_uri_store.go", "storage.go", @@ -26,6 +27,7 @@ go_test( srcs = [ "build_store_test.go", "queue_store_test.go", + "request_log_store_test.go", "request_store_test.go", "request_uri_store_test.go", "storage_test.go", diff --git a/stovepipe/extension/storage/mysql/request_log_store.go b/stovepipe/extension/storage/mysql/request_log_store.go new file mode 100644 index 000000000..54ed16774 --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_log_store.go @@ -0,0 +1,168 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + + "github.com/uber-go/tally" + + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +const listRequestLogQuery = ` + SELECT queue, request_id, log_id, timestamp_ms, state, event, + request_version, outcome_reason, metadata + FROM request_log + WHERE queue = ? AND request_id = ? + ORDER BY timestamp_ms ASC, log_id ASC` + +type requestLogStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewRequestLogStore creates a MySQL-backed RequestLogStore. +func NewRequestLogStore(db *sql.DB, scope tally.Scope, queue string) storage.RequestLogStore { + return &requestLogStore{db: db, scope: scope, queue: queue} +} + +func (r *requestLogStore) Create(ctx context.Context, log entity.RequestLog) (retErr error) { + op := metrics.Begin(r.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := log.Validate(); err != nil { + return fmt.Errorf("invalid request log id=%q: %w", log.ID, err) + } + if log.Queue != r.queue { + return fmt.Errorf("request log %q queue %q does not match the store's bound queue %q", log.ID, log.Queue, r.queue) + } + metadata := log.Metadata + if metadata == nil { + metadata = map[string]string{} + } + metadataJSON, err := json.Marshal(metadata) + if err != nil { + return fmt.Errorf("failed to marshal request log metadata request_id=%q log_id=%q: %w", log.RequestID, log.ID, err) + } + + _, err = r.db.ExecContext(ctx, ` + INSERT INTO request_log ( + queue, request_id, log_id, timestamp_ms, state, event, request_version, + outcome_reason, metadata + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + log.Queue, + log.RequestID, + log.ID, + log.TimestampMs, + log.State, + log.Event, + log.RequestVersion, + log.OutcomeReason, + metadataJSON, + ) + if err != nil { + if isDuplicateEntry(err) { + return fmt.Errorf("request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to insert request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err) + } + return nil +} + +func (r *requestLogStore) Get(ctx context.Context, requestID, logID string) (ret entity.RequestLog, retErr error) { + op := metrics.Begin(r.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + log, err := scanRequestLog(r.db.QueryRowContext(ctx, ` + SELECT queue, request_id, log_id, timestamp_ms, state, event, + request_version, outcome_reason, metadata + FROM request_log + WHERE queue = ? AND request_id = ? AND log_id = ?`, + r.queue, requestID, logID, + )) + if errors.Is(err, sql.ErrNoRows) { + return entity.RequestLog{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.RequestLog{}, fmt.Errorf("failed to get request log request_id=%q log_id=%q: %w", requestID, logID, err) + } + return log, nil +} + +func (r *requestLogStore) List(ctx context.Context, requestID string) (ret []entity.RequestLog, retErr error) { + op := metrics.Begin(r.scope, "list", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if requestID == "" { + return nil, fmt.Errorf("request log request ID must not be empty") + } + + rows, err := r.db.QueryContext(ctx, listRequestLogQuery, r.queue, requestID) + if err != nil { + return nil, fmt.Errorf("failed to list request log request_id=%q: %w", requestID, err) + } + defer rows.Close() + + logs := make([]entity.RequestLog, 0) + for rows.Next() { + log, err := scanRequestLog(rows) + if err != nil { + return nil, fmt.Errorf("failed to scan request log request_id=%q: %w", requestID, err) + } + logs = append(logs, log) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate request log request_id=%q: %w", requestID, err) + } + return logs, nil +} + +type requestLogScanner interface { + Scan(dest ...any) error +} + +func scanRequestLog(scanner requestLogScanner) (entity.RequestLog, error) { + var log entity.RequestLog + var metadataJSON []byte + err := scanner.Scan( + &log.Queue, + &log.RequestID, + &log.ID, + &log.TimestampMs, + &log.State, + &log.Event, + &log.RequestVersion, + &log.OutcomeReason, + &metadataJSON, + ) + if err != nil { + return entity.RequestLog{}, err + } + if err := json.Unmarshal(metadataJSON, &log.Metadata); err != nil { + return entity.RequestLog{}, fmt.Errorf("failed to unmarshal request log metadata: %w", err) + } + if log.Metadata == nil { + log.Metadata = map[string]string{} + } + return log, nil +} diff --git a/stovepipe/extension/storage/mysql/request_log_store_test.go b/stovepipe/extension/storage/mysql/request_log_store_test.go new file mode 100644 index 000000000..8b42aad9f --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_log_store_test.go @@ -0,0 +1,331 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/uber/submitqueue/stovepipe/entity" + "github.com/uber/submitqueue/stovepipe/extension/storage" +) + +const ( + testLogQueue = "monorepo/main" + testLogRequestID = "request/monorepo/main/1" +) + +var requestLogColumnNames = []string{ + "queue", "request_id", "log_id", "timestamp_ms", "state", "event", + "request_version", "outcome_reason", "metadata", +} + +func setupRequestLogStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.RequestLogStore) { + t.Helper() + db, mock, err := sqlmock.New() + require.NoError(t, err) + return db, mock, NewRequestLogStore(db, testMetrics(), testLogQueue) +} + +func acceptedRequestLog() entity.RequestLog { + return entity.RequestLog{ + ID: "state/1", + Queue: testLogQueue, + RequestID: testLogRequestID, + TimestampMs: 1735689600000, + State: entity.RequestStateAccepted, + RequestVersion: 1, + Metadata: map[string]string{"source": "test"}, + } +} + +func requestLogMetadataJSON(t *testing.T, entry entity.RequestLog) []byte { + t.Helper() + metadata := entry.Metadata + if metadata == nil { + metadata = map[string]string{} + } + metadataJSON, err := json.Marshal(metadata) + require.NoError(t, err) + return metadataJSON +} + +func requestLogRow(t *testing.T, entry entity.RequestLog) *sqlmock.Rows { + t.Helper() + return requestLogRowWithMetadata(entry, requestLogMetadataJSON(t, entry)) +} + +func requestLogRowWithMetadata(entry entity.RequestLog, metadata any) *sqlmock.Rows { + return sqlmock.NewRows(requestLogColumnNames).AddRow( + entry.Queue, + entry.RequestID, + entry.ID, + entry.TimestampMs, + entry.State, + entry.Event, + entry.RequestVersion, + entry.OutcomeReason, + metadata, + ) +} + +func TestRequestLogStoreCreate(t *testing.T) { + entry := acceptedRequestLog() + metadataJSON, err := json.Marshal(entry.Metadata) + require.NoError(t, err) + emptyMetadata := entry + emptyMetadata.ID = "state/empty-metadata" + emptyMetadata.Metadata = nil + + tests := []struct { + name string + entry entity.RequestLog + setup func(sqlmock.Sqlmock) + wantErrIs error + wantErr bool + }{ + { + name: "success", + entry: entry, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO request_log"). + WithArgs(entry.Queue, entry.RequestID, entry.ID, entry.TimestampMs, entry.State, entry.Event, entry.RequestVersion, entry.OutcomeReason, metadataJSON). + WillReturnResult(sqlmock.NewResult(0, 1)) + }, + }, + { + name: "nil metadata normalized", + entry: emptyMetadata, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO request_log"). + WithArgs(emptyMetadata.Queue, emptyMetadata.RequestID, emptyMetadata.ID, emptyMetadata.TimestampMs, emptyMetadata.State, emptyMetadata.Event, emptyMetadata.RequestVersion, emptyMetadata.OutcomeReason, []byte("{}")). + WillReturnResult(sqlmock.NewResult(0, 1)) + }, + }, + { + name: "duplicate identity", + entry: entry, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO request_log"). + WithArgs(entry.Queue, entry.RequestID, entry.ID, entry.TimestampMs, entry.State, entry.Event, entry.RequestVersion, entry.OutcomeReason, metadataJSON). + WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) + }, + wantErr: true, + wantErrIs: storage.ErrAlreadyExists, + }, + { + name: "database failure", + entry: entry, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO request_log"). + WithArgs(entry.Queue, entry.RequestID, entry.ID, entry.TimestampMs, entry.State, entry.Event, entry.RequestVersion, entry.OutcomeReason, metadataJSON). + WillReturnError(fmt.Errorf("connection reset")) + }, + wantErr: true, + }, + { + name: "invalid entry", + entry: func() entity.RequestLog { + invalid := entry + invalid.State = entity.RequestStateUnknown + return invalid + }(), + wantErr: true, + }, + { + name: "wrong queue", + entry: func() entity.RequestLog { + wrongQueue := entry + wrongQueue.Queue = "other" + return wrongQueue + }(), + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupRequestLogStoreTest(t) + defer db.Close() + if tt.setup != nil { + tt.setup(mock) + } + + err := store.Create(context.Background(), tt.entry) + if tt.wantErr { + require.Error(t, err) + if tt.wantErrIs != nil { + assert.ErrorIs(t, err, tt.wantErrIs) + } + } else { + require.NoError(t, err) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestRequestLogStoreGet(t *testing.T) { + future := acceptedRequestLog() + future.State = entity.RequestState("future_state") + emptyMetadata := future + emptyMetadata.Metadata = map[string]string{} + + tests := []struct { + name string + setup func(sqlmock.Sqlmock) + want entity.RequestLog + wantErrIs error + wantErr bool + }{ + { + name: "found without validating future vocabulary", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT queue, request_id, log_id, timestamp_ms, state, event"). + WithArgs(testLogQueue, future.RequestID, future.ID). + WillReturnRows(requestLogRow(t, future)) + }, + want: future, + }, + { + name: "JSON null normalized to empty metadata", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT queue, request_id, log_id, timestamp_ms, state, event"). + WithArgs(testLogQueue, future.RequestID, future.ID). + WillReturnRows(requestLogRowWithMetadata(future, []byte("null"))) + }, + want: emptyMetadata, + }, + { + name: "invalid metadata value type", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT queue, request_id, log_id, timestamp_ms, state, event"). + WithArgs(testLogQueue, future.RequestID, future.ID). + WillReturnRows(requestLogRowWithMetadata(future, []byte(`{"attempt":2}`))) + }, + wantErr: true, + }, + { + name: "not found", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT queue, request_id, log_id, timestamp_ms, state, event"). + WithArgs(testLogQueue, future.RequestID, future.ID). + WillReturnError(sql.ErrNoRows) + }, + wantErr: true, + wantErrIs: storage.ErrNotFound, + }, + { + name: "database failure", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT queue, request_id, log_id, timestamp_ms, state, event"). + WithArgs(testLogQueue, future.RequestID, future.ID). + WillReturnError(fmt.Errorf("connection reset")) + }, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupRequestLogStoreTest(t) + defer db.Close() + tt.setup(mock) + + got, err := store.Get(context.Background(), future.RequestID, future.ID) + if tt.wantErr { + require.Error(t, err) + if tt.wantErrIs != nil { + assert.ErrorIs(t, err, tt.wantErrIs) + } + } else { + require.NoError(t, err) + assert.Equal(t, tt.want, got) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestRequestLogStoreList(t *testing.T) { + first := acceptedRequestLog() + second := first + second.ID = "state/2" + second.State = entity.RequestStateProcessing + second.RequestVersion = 2 + + tests := []struct { + name string + setup func(sqlmock.Sqlmock) + want []entity.RequestLog + wantErr bool + }{ + { + name: "all records", + setup: func(mock sqlmock.Sqlmock) { + rows := requestLogRow(t, first).AddRow(second.Queue, second.RequestID, second.ID, second.TimestampMs, second.State, second.Event, second.RequestVersion, second.OutcomeReason, requestLogMetadataJSON(t, second)) + mock.ExpectQuery("ORDER BY timestamp_ms ASC, log_id ASC"). + WithArgs(testLogQueue, testLogRequestID). + WillReturnRows(rows) + }, + want: []entity.RequestLog{first, second}, + }, + { + name: "empty log", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("ORDER BY timestamp_ms ASC, log_id ASC"). + WithArgs(testLogQueue, testLogRequestID). + WillReturnRows(sqlmock.NewRows(requestLogColumnNames)) + }, + want: []entity.RequestLog{}, + }, + { + name: "database failure", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("ORDER BY timestamp_ms ASC, log_id ASC"). + WithArgs(testLogQueue, testLogRequestID). + WillReturnError(fmt.Errorf("connection reset")) + }, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupRequestLogStoreTest(t) + defer db.Close() + if tt.setup != nil { + tt.setup(mock) + } + + got, err := store.List(context.Background(), testLogRequestID) + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + assert.Equal(t, tt.want, got) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} diff --git a/stovepipe/extension/storage/mysql/storage.go b/stovepipe/extension/storage/mysql/storage.go index b00a273b5..c25b8a32c 100644 --- a/stovepipe/extension/storage/mysql/storage.go +++ b/stovepipe/extension/storage/mysql/storage.go @@ -47,6 +47,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { return &mysqlStorage{ requestStore: NewRequestStore(s.db, s.scope.SubScope("request_store"), queueName), requestURIStore: NewRequestURIStore(s.db, s.scope.SubScope("request_uri_store"), queueName), + requestLogStore: NewRequestLogStore(s.db, s.scope.SubScope("request_log_store"), queueName), queueStore: NewQueueStore(s.db, s.scope.SubScope("queue_store"), queueName), buildStore: NewBuildStore(s.db, s.scope.SubScope("build_store"), queueName), validationFactStore: NewValidationFactStore(s.db, s.scope.SubScope("validation_fact_store"), queueName), @@ -62,6 +63,7 @@ func (s *Storage) Close() error { type mysqlStorage struct { requestStore storage.RequestStore requestURIStore storage.RequestURIStore + requestLogStore storage.RequestLogStore queueStore storage.QueueStore buildStore storage.BuildStore validationFactStore storage.ValidationFactStore @@ -80,6 +82,11 @@ func (f *mysqlStorage) GetRequestURIStore() storage.RequestURIStore { return f.requestURIStore } +// GetRequestLogStore returns the MySQL-backed RequestLogStore. +func (f *mysqlStorage) GetRequestLogStore() storage.RequestLogStore { + return f.requestLogStore +} + // GetQueueStore returns the MySQL-backed QueueStore. func (f *mysqlStorage) GetQueueStore() storage.QueueStore { return f.queueStore diff --git a/stovepipe/extension/storage/mysql/storage_test.go b/stovepipe/extension/storage/mysql/storage_test.go index 04405ff95..bbd8b6fdc 100644 --- a/stovepipe/extension/storage/mysql/storage_test.go +++ b/stovepipe/extension/storage/mysql/storage_test.go @@ -40,6 +40,7 @@ func TestNewStorage(t *testing.T) { require.NoError(t, err) assert.NotNil(t, bound.GetRequestStore()) assert.NotNil(t, bound.GetRequestURIStore()) + assert.NotNil(t, bound.GetRequestLogStore()) assert.NotNil(t, bound.GetQueueStore()) assert.NotNil(t, bound.GetBuildStore()) diff --git a/stovepipe/extension/storage/request_log_store.go b/stovepipe/extension/storage/request_log_store.go new file mode 100644 index 000000000..cd19b8698 --- /dev/null +++ b/stovepipe/extension/storage/request_log_store.go @@ -0,0 +1,35 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package storage + +//go:generate mockgen -source=request_log_store.go -destination=mock/request_log_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// RequestLogStore retains immutable occurrences for requests in its bound queue. +type RequestLogStore interface { + // Create persists log and returns ErrAlreadyExists when its stable identity exists. + Create(ctx context.Context, log entity.RequestLog) error + + // Get returns one record identified by requestID and logID, or ErrNotFound when absent. + Get(ctx context.Context, requestID, logID string) (entity.RequestLog, error) + + // List returns all records for one request ordered by timestamp and log ID ascending. + List(ctx context.Context, requestID string) ([]entity.RequestLog, error) +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index 9aa74f272..f9470f88a 100644 --- a/stovepipe/extension/storage/storage.go +++ b/stovepipe/extension/storage/storage.go @@ -75,6 +75,9 @@ type Storage interface { // GetRequestURIStore returns the RequestURIStore instance. GetRequestURIStore() RequestURIStore + // GetRequestLogStore returns the RequestLogStore instance. + GetRequestLogStore() RequestLogStore + // GetQueueStore returns the QueueStore instance. GetQueueStore() QueueStore diff --git a/test/integration/stovepipe/extension/storage/mysql/storage_test.go b/test/integration/stovepipe/extension/storage/mysql/storage_test.go index 184483b9b..242030d48 100644 --- a/test/integration/stovepipe/extension/storage/mysql/storage_test.go +++ b/test/integration/stovepipe/extension/storage/mysql/storage_test.go @@ -94,6 +94,14 @@ func TestMySQLStorage(t *testing.T) { testSuite.SetLogger(testutil.NewTestLogger(t)) suite.Run(t, testSuite) }) + + t.Run("RequestLogStore", func(t *testing.T) { + resetStorage(t, db) + testSuite := new(MySQLRequestLogStoreSuite) + testSuite.SetContext(ctx) + testSuite.SetFactory(factory) + suite.Run(t, testSuite) + }) } func resetStorage(t *testing.T, db *sql.DB) { @@ -273,6 +281,11 @@ type MySQLBuildStoreSuite struct { storagesuite.BuildStoreContractSuite } +// MySQLRequestLogStoreSuite exercises the MySQL-backed RequestLogStore against a real MySQL instance. +type MySQLRequestLogStoreSuite struct { + storagesuite.RequestLogStoreContractSuite +} + // mysqlFactory adapts the MySQL storage backend's queue binding to the // storage.Factory seam for the contract suite, mirroring the host wiring. type mysqlFactory struct { diff --git a/test/integration/stovepipe/extension/storage/suite.go b/test/integration/stovepipe/extension/storage/suite.go index b0a3bf613..0b660ac29 100644 --- a/test/integration/stovepipe/extension/storage/suite.go +++ b/test/integration/stovepipe/extension/storage/suite.go @@ -184,6 +184,127 @@ func (s *QueueStoreContractSuite) TestQueueStore_UpdateSequentialCAS() { assert.Equal(t, int32(3), got.Version) } +// RequestLogStoreContractSuite defines contract tests for storage.RequestLogStore. +// All RequestLogStore implementations must pass these tests. +type RequestLogStoreContractSuite struct { + suite.Suite + ctx context.Context + factory storage.Factory +} + +// SetContext sets the context for tests. +func (s *RequestLogStoreContractSuite) SetContext(ctx context.Context) { + s.ctx = ctx +} + +// SetFactory provides the Factory that resolves the store under test per queue. +func (s *RequestLogStoreContractSuite) SetFactory(factory storage.Factory) { + s.factory = factory +} + +func (s *RequestLogStoreContractSuite) storeFor(queue string) storage.RequestLogStore { + bound, err := s.factory.For(storage.Config{QueueName: queue}) + s.Require().NoError(err) + return bound.GetRequestLogStore() +} + +func (s *RequestLogStoreContractSuite) entry(queue, requestID, id string, timestampMs int64, version int32) entity.RequestLog { + return entity.RequestLog{ + ID: id, + Queue: queue, + RequestID: requestID, + TimestampMs: timestampMs, + State: entity.RequestStateAccepted, + RequestVersion: version, + Metadata: map[string]string{"source": "contract"}, + } +} + +// TestRequestLogStore_CreateAndGet verifies all occurrence fields round-trip unchanged. +func (s *RequestLogStoreContractSuite) TestRequestLogStore_CreateAndGet() { + const ( + queue = "contract/history-create" + requestID = "request/contract/history-create/1" + ) + entry := entity.RequestLog{ + ID: "fact/repository", + Queue: queue, + RequestID: requestID, + TimestampMs: 1735689600000, + Event: entity.RequestEventValidationFactRecorded, + Metadata: map[string]string{ + "fact_degree": "0.5", + "scope": "repository", + }, + } + store := s.storeFor(queue) + require.NoError(s.T(), store.Create(s.ctx, entry)) + + got, err := store.Get(s.ctx, requestID, entry.ID) + require.NoError(s.T(), err) + assert.Equal(s.T(), entry, got) +} + +// TestRequestLogStore_CreateAlreadyExists verifies stable identities are create-only. +func (s *RequestLogStoreContractSuite) TestRequestLogStore_CreateAlreadyExists() { + const ( + queue = "contract/history-duplicate" + requestID = "request/contract/history-duplicate/1" + ) + store := s.storeFor(queue) + entry := s.entry(queue, requestID, "state/1", 1735689600000, 1) + require.NoError(s.T(), store.Create(s.ctx, entry)) + + err := store.Create(s.ctx, entry) + assert.ErrorIs(s.T(), err, storage.ErrAlreadyExists) + + got, err := store.Get(s.ctx, requestID, entry.ID) + require.NoError(s.T(), err) + assert.Equal(s.T(), entry, got) +} + +// TestRequestLogStore_GetNotFound verifies a missing stable identity returns ErrNotFound. +func (s *RequestLogStoreContractSuite) TestRequestLogStore_GetNotFound() { + _, err := s.storeFor("contract/history-missing").Get(s.ctx, "request/missing/1", "state/1") + assert.True(s.T(), storage.IsNotFound(err)) +} + +// TestRequestLogStore_List verifies chronological ordering and equal-time tie breaking. +func (s *RequestLogStoreContractSuite) TestRequestLogStore_List() { + const ( + queue = "contract/history-list" + requestID = "request/contract/history-list/1" + ) + store := s.storeFor(queue) + last := s.entry(queue, requestID, "state/z", 2000, 3) + first := s.entry(queue, requestID, "state/a", 1000, 1) + second := s.entry(queue, requestID, "state/b", 1000, 2) + for _, entry := range []entity.RequestLog{last, second, first} { + require.NoError(s.T(), store.Create(s.ctx, entry)) + } + + logs, err := store.List(s.ctx, requestID) + require.NoError(s.T(), err) + assert.Equal(s.T(), []entity.RequestLog{first, second, last}, logs) +} + +// TestRequestLogStore_QueueIsolation verifies identical request and entry IDs remain queue-scoped. +func (s *RequestLogStoreContractSuite) TestRequestLogStore_QueueIsolation() { + const requestID = "request/shared/1" + entryA := s.entry("contract/history-a", requestID, "state/1", 1000, 1) + entryB := s.entry("contract/history-b", requestID, "state/1", 2000, 1) + require.NoError(s.T(), s.storeFor(entryA.Queue).Create(s.ctx, entryA)) + require.NoError(s.T(), s.storeFor(entryB.Queue).Create(s.ctx, entryB)) + + gotA, err := s.storeFor(entryA.Queue).Get(s.ctx, requestID, entryA.ID) + require.NoError(s.T(), err) + assert.Equal(s.T(), entryA, gotA) + + gotB, err := s.storeFor(entryB.Queue).Get(s.ctx, requestID, entryB.ID) + require.NoError(s.T(), err) + assert.Equal(s.T(), entryB, gotB) +} + // BuildStoreContractSuite defines contract tests for storage.BuildStore. // All BuildStore implementations must pass these tests. type BuildStoreContractSuite struct {