diff --git a/stovepipe/entity/BUILD.bazel b/stovepipe/entity/BUILD.bazel index 18eeeb9c7..3abaab2d3 100644 --- a/stovepipe/entity/BUILD.bazel +++ b/stovepipe/entity/BUILD.bazel @@ -8,6 +8,7 @@ go_library( "queue.go", "queue_config.go", "request.go", + "request_history.go", "request_id.go", "validation_fact.go", ], @@ -19,6 +20,7 @@ go_test( name = "go_default_test", srcs = [ "build_test.go", + "request_history_test.go", "request_id_test.go", "request_test.go", "validation_fact_test.go", diff --git a/stovepipe/entity/request_history.go b/stovepipe/entity/request_history.go new file mode 100644 index 000000000..c98190670 --- /dev/null +++ b/stovepipe/entity/request_history.go @@ -0,0 +1,175 @@ +// 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" + "math" +) + +// 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" +) + +// RequestHistoryEntry is one immutable request state or explanatory lifecycle occurrence. +type RequestHistoryEntry 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 history contains this entry. + 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 entry and is unset on an event entry. + State RequestState `json:"state"` + // Event identifies the occurrence recorded by an event entry and is unset on a state entry. + Event RequestEvent `json:"event"` + // RequestVersion is the durable request version recorded by a state entry and is zero on an event entry. + RequestVersion int32 `json:"request_version"` + // SupersededByRequestID identifies the newer request responsible for supersession and is otherwise empty. + SupersededByRequestID string `json:"superseded_by_request_id"` + // BuildID identifies the build associated with the occurrence and is empty when no build applies. + BuildID string `json:"build_id"` + // OutcomeReason is the durable domain reason for a terminal request state and is otherwise unset. + OutcomeReason RequestOutcomeReason `json:"outcome_reason"` + // FactDegree is the validation degree recorded by a validation-fact event; its event kind distinguishes zero from absence. + FactDegree float64 `json:"fact_degree"` +} + +// Validate verifies the invariants required for a newly persisted history entry. +func (e RequestHistoryEntry) Validate() error { + if e.ID == "" { + return fmt.Errorf("request history entry ID must not be empty") + } + if e.Queue == "" { + return fmt.Errorf("request history queue must not be empty") + } + if e.RequestID == "" { + return fmt.Errorf("request history request ID must not be empty") + } + if e.TimestampMs <= 0 { + return fmt.Errorf("request history timestamp must be positive") + } + if (e.State == RequestStateUnknown) == (e.Event == RequestEventUnknown) { + return fmt.Errorf("request history entry must contain exactly one of state and event") + } + if e.State != RequestStateUnknown { + return e.validateState() + } + return e.validateEvent() +} + +func (e RequestHistoryEntry) validateState() error { + if e.RequestVersion <= 0 { + return fmt.Errorf("state history entry must have a positive request version") + } + if e.FactDegree != 0 { + return fmt.Errorf("state history entry must not contain a validation degree") + } + + switch e.State { + case RequestStateAccepted, RequestStateProcessing: + if e.SupersededByRequestID != "" || e.BuildID != "" || e.OutcomeReason != RequestOutcomeReasonUnknown { + return fmt.Errorf("non-terminal state history entry must not contain terminal context") + } + case RequestStateSuperseded: + if e.SupersededByRequestID == "" { + return fmt.Errorf("superseded state history entry must identify the newer request") + } + if e.BuildID != "" || e.OutcomeReason != RequestOutcomeReasonSupersededByNewerHead { + return fmt.Errorf("superseded state history entry has invalid outcome context") + } + case RequestStateSucceeded: + if e.SupersededByRequestID != "" || e.BuildID == "" || e.OutcomeReason != RequestOutcomeReasonBuildSucceeded { + return fmt.Errorf("succeeded state history entry has invalid outcome context") + } + case RequestStateFailed: + if e.SupersededByRequestID != "" || !isFailureReason(e.OutcomeReason) { + return fmt.Errorf("failed state history entry has invalid outcome context") + } + case RequestStateCancelled: + if e.SupersededByRequestID != "" || e.BuildID == "" || e.OutcomeReason != RequestOutcomeReasonBuildCancelled { + return fmt.Errorf("cancelled state history entry has invalid outcome context") + } + default: + return fmt.Errorf("unknown request state %q", e.State) + } + return nil +} + +func (e RequestHistoryEntry) validateEvent() error { + if e.RequestVersion != 0 || e.SupersededByRequestID != "" || e.OutcomeReason != RequestOutcomeReasonUnknown { + return fmt.Errorf("event history entry must not contain request-state context") + } + + switch e.Event { + case RequestEventBuildTriggered, RequestEventBuildFinished: + if e.BuildID == "" || e.FactDegree != 0 { + return fmt.Errorf("build history event has invalid context") + } + case RequestEventValidationFactRecorded: + if e.BuildID != "" || math.IsNaN(e.FactDegree) || math.IsInf(e.FactDegree, 0) || e.FactDegree < DegreeGreen || e.FactDegree > DegreeBroken { + return fmt.Errorf("validation-fact history event has invalid context") + } + 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_history_test.go b/stovepipe/entity/request_history_test.go new file mode 100644 index 000000000..10130736b --- /dev/null +++ b/stovepipe/entity/request_history_test.go @@ -0,0 +1,276 @@ +// 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 ( + "math" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRequestHistoryEntryValidate(t *testing.T) { + base := RequestHistoryEntry{ + ID: "state/1", + Queue: "monorepo/main", + RequestID: "request/monorepo/main/1", + TimestampMs: 1735689600000, + State: RequestStateAccepted, + RequestVersion: 1, + } + + tests := []struct { + name string + mutate func(RequestHistoryEntry) RequestHistoryEntry + wantErr bool + }{ + {name: "accepted state"}, + { + name: "superseded state", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateSuperseded + entry.SupersededByRequestID = "request/monorepo/main/2" + entry.OutcomeReason = RequestOutcomeReasonSupersededByNewerHead + return entry + }, + }, + { + name: "failed without build", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateFailed + entry.OutcomeReason = RequestOutcomeReasonProcessingFailed + return entry + }, + }, + { + name: "succeeded state", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateSucceeded + entry.BuildID = "42" + entry.OutcomeReason = RequestOutcomeReasonBuildSucceeded + return entry + }, + }, + { + name: "failed build state", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateFailed + entry.BuildID = "42" + entry.OutcomeReason = RequestOutcomeReasonBuildFailed + return entry + }, + }, + { + name: "failed polling state", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateFailed + entry.OutcomeReason = RequestOutcomeReasonBuildPollingExhausted + return entry + }, + }, + { + name: "failed timeout state", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateFailed + entry.OutcomeReason = RequestOutcomeReasonValidationTimeout + return entry + }, + }, + { + name: "cancelled state", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateCancelled + entry.BuildID = "42" + entry.OutcomeReason = RequestOutcomeReasonBuildCancelled + return entry + }, + }, + { + name: "validation fact with green degree", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateUnknown + entry.Event = RequestEventValidationFactRecorded + entry.RequestVersion = 0 + return entry + }, + }, + { + name: "build event", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateUnknown + entry.Event = RequestEventBuildTriggered + entry.RequestVersion = 0 + entry.BuildID = "42" + return entry + }, + }, + { + name: "build finished event", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateUnknown + entry.Event = RequestEventBuildFinished + entry.RequestVersion = 0 + entry.BuildID = "42" + return entry + }, + }, + {name: "missing ID", mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { entry.ID = ""; return entry }, wantErr: true}, + {name: "missing queue", mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { entry.Queue = ""; return entry }, wantErr: true}, + {name: "missing request ID", mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { entry.RequestID = ""; return entry }, wantErr: true}, + {name: "non-positive timestamp", mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { entry.TimestampMs = 0; return entry }, wantErr: true}, + {name: "missing occurrence", mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { entry.State = RequestStateUnknown; return entry }, wantErr: true}, + {name: "two occurrences", mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.Event = RequestEventBuildTriggered + return entry + }, wantErr: true}, + {name: "state without version", mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { entry.RequestVersion = 0; return entry }, wantErr: true}, + {name: "unknown state", mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestState("future") + return entry + }, wantErr: true}, + {name: "non-terminal state with outcome", mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.OutcomeReason = RequestOutcomeReasonProcessingFailed + return entry + }, wantErr: true}, + {name: "state with fact degree", mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.FactDegree = 0.5 + return entry + }, wantErr: true}, + { + name: "superseded without newer request", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateSuperseded + entry.OutcomeReason = RequestOutcomeReasonSupersededByNewerHead + return entry + }, + wantErr: true, + }, + { + name: "superseded with build", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateSuperseded + entry.SupersededByRequestID = "request/monorepo/main/2" + entry.BuildID = "42" + entry.OutcomeReason = RequestOutcomeReasonSupersededByNewerHead + return entry + }, + wantErr: true, + }, + { + name: "succeeded without build", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateSucceeded + entry.OutcomeReason = RequestOutcomeReasonBuildSucceeded + return entry + }, + wantErr: true, + }, + { + name: "failed without reason", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateFailed + return entry + }, + wantErr: true, + }, + { + name: "cancelled without build", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateCancelled + entry.OutcomeReason = RequestOutcomeReasonBuildCancelled + return entry + }, + wantErr: true, + }, + { + name: "event with request version", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateUnknown + entry.Event = RequestEventBuildTriggered + entry.BuildID = "42" + return entry + }, + wantErr: true, + }, + { + name: "build event without build", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateUnknown + entry.Event = RequestEventBuildTriggered + entry.RequestVersion = 0 + return entry + }, + wantErr: true, + }, + { + name: "fact event with build", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateUnknown + entry.Event = RequestEventValidationFactRecorded + entry.RequestVersion = 0 + entry.BuildID = "42" + return entry + }, + wantErr: true, + }, + { + name: "unknown event", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateUnknown + entry.Event = RequestEvent("future") + entry.RequestVersion = 0 + return entry + }, + wantErr: true, + }, + { + name: "fact outside degree range", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateUnknown + entry.Event = RequestEventValidationFactRecorded + entry.RequestVersion = 0 + entry.FactDegree = 1.1 + return entry + }, + wantErr: true, + }, + { + name: "fact with NaN degree", + mutate: func(entry RequestHistoryEntry) RequestHistoryEntry { + entry.State = RequestStateUnknown + entry.Event = RequestEventValidationFactRecorded + entry.RequestVersion = 0 + entry.FactDegree = math.NaN() + 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..8241228a1 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_history_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..8d2885688 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_history_store_mock.go", "request_store_mock.go", "request_uri_store_mock.go", "storage_mock.go", diff --git a/stovepipe/extension/storage/mock/request_history_store_mock.go b/stovepipe/extension/storage/mock/request_history_store_mock.go new file mode 100644 index 000000000..ba5070092 --- /dev/null +++ b/stovepipe/extension/storage/mock/request_history_store_mock.go @@ -0,0 +1,87 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: request_history_store.go +// +// Generated by this command: +// +// mockgen -source=request_history_store.go -destination=mock/request_history_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" + storage "github.com/uber/submitqueue/stovepipe/extension/storage" + gomock "go.uber.org/mock/gomock" +) + +// MockRequestHistoryStore is a mock of RequestHistoryStore interface. +type MockRequestHistoryStore struct { + ctrl *gomock.Controller + recorder *MockRequestHistoryStoreMockRecorder + isgomock struct{} +} + +// MockRequestHistoryStoreMockRecorder is the mock recorder for MockRequestHistoryStore. +type MockRequestHistoryStoreMockRecorder struct { + mock *MockRequestHistoryStore +} + +// NewMockRequestHistoryStore creates a new mock instance. +func NewMockRequestHistoryStore(ctrl *gomock.Controller) *MockRequestHistoryStore { + mock := &MockRequestHistoryStore{ctrl: ctrl} + mock.recorder = &MockRequestHistoryStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockRequestHistoryStore) EXPECT() *MockRequestHistoryStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockRequestHistoryStore) Create(ctx context.Context, entry entity.RequestHistoryEntry) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, entry) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockRequestHistoryStoreMockRecorder) Create(ctx, entry any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockRequestHistoryStore)(nil).Create), ctx, entry) +} + +// Get mocks base method. +func (m *MockRequestHistoryStore) Get(ctx context.Context, requestID, entryID string) (entity.RequestHistoryEntry, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, requestID, entryID) + ret0, _ := ret[0].(entity.RequestHistoryEntry) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockRequestHistoryStoreMockRecorder) Get(ctx, requestID, entryID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockRequestHistoryStore)(nil).Get), ctx, requestID, entryID) +} + +// List mocks base method. +func (m *MockRequestHistoryStore) List(ctx context.Context, requestID string, query storage.RequestHistoryQuery) ([]entity.RequestHistoryEntry, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "List", ctx, requestID, query) + ret0, _ := ret[0].([]entity.RequestHistoryEntry) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// List indicates an expected call of List. +func (mr *MockRequestHistoryStoreMockRecorder) List(ctx, requestID, query any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "List", reflect.TypeOf((*MockRequestHistoryStore)(nil).List), ctx, requestID, query) +} diff --git a/stovepipe/extension/storage/mock/storage_mock.go b/stovepipe/extension/storage/mock/storage_mock.go index c73f78835..3bc812122 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)) } +// GetRequestHistoryStore mocks base method. +func (m *MockStorage) GetRequestHistoryStore() storage.RequestHistoryStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestHistoryStore") + ret0, _ := ret[0].(storage.RequestHistoryStore) + return ret0 +} + +// GetRequestHistoryStore indicates an expected call of GetRequestHistoryStore. +func (mr *MockStorageMockRecorder) GetRequestHistoryStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestHistoryStore", reflect.TypeOf((*MockStorage)(nil).GetRequestHistoryStore)) +} + // 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..de4b16743 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_history_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_history_store_test.go", "request_store_test.go", "request_uri_store_test.go", "storage_test.go", diff --git a/stovepipe/extension/storage/mysql/request_history_store.go b/stovepipe/extension/storage/mysql/request_history_store.go new file mode 100644 index 000000000..9e7061a00 --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_history_store.go @@ -0,0 +1,178 @@ +// 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" + "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 ( + listRequestHistoryFirstPageQuery = ` + SELECT queue, request_id, entry_id, timestamp_ms, state, event, + request_version, superseded_by_request_id, build_id, outcome_reason, fact_degree + FROM request_history + WHERE queue = ? AND request_id = ? + ORDER BY timestamp_ms ASC, entry_id ASC + LIMIT ?` + listRequestHistoryAfterCursorQuery = ` + SELECT queue, request_id, entry_id, timestamp_ms, state, event, + request_version, superseded_by_request_id, build_id, outcome_reason, fact_degree + FROM request_history + WHERE queue = ? AND request_id = ? + AND (timestamp_ms > ? OR (timestamp_ms = ? AND entry_id > ?)) + ORDER BY timestamp_ms ASC, entry_id ASC + LIMIT ?` +) + +type requestHistoryStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewRequestHistoryStore creates a MySQL-backed RequestHistoryStore. +func NewRequestHistoryStore(db *sql.DB, scope tally.Scope, queue string) storage.RequestHistoryStore { + return &requestHistoryStore{db: db, scope: scope, queue: queue} +} + +func (r *requestHistoryStore) Create(ctx context.Context, entry entity.RequestHistoryEntry) (retErr error) { + op := metrics.Begin(r.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := entry.Validate(); err != nil { + return fmt.Errorf("invalid request history entry id=%q: %w", entry.ID, err) + } + if entry.Queue != r.queue { + return fmt.Errorf("request history entry %q queue %q does not match the store's bound queue %q", entry.ID, entry.Queue, r.queue) + } + + _, err := r.db.ExecContext(ctx, ` + INSERT INTO request_history ( + queue, request_id, entry_id, timestamp_ms, state, event, request_version, + superseded_by_request_id, build_id, outcome_reason, fact_degree + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + entry.Queue, + entry.RequestID, + entry.ID, + entry.TimestampMs, + entry.State, + entry.Event, + entry.RequestVersion, + entry.SupersededByRequestID, + entry.BuildID, + entry.OutcomeReason, + entry.FactDegree, + ) + if err != nil { + if isDuplicateEntry(err) { + return fmt.Errorf("request history entry request_id=%q entry_id=%q: %w", entry.RequestID, entry.ID, storage.ErrAlreadyExists) + } + return fmt.Errorf("failed to insert request history entry request_id=%q entry_id=%q: %w", entry.RequestID, entry.ID, err) + } + return nil +} + +func (r *requestHistoryStore) Get(ctx context.Context, requestID, entryID string) (ret entity.RequestHistoryEntry, retErr error) { + op := metrics.Begin(r.scope, "get", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + entry, err := scanRequestHistoryEntry(r.db.QueryRowContext(ctx, ` + SELECT queue, request_id, entry_id, timestamp_ms, state, event, + request_version, superseded_by_request_id, build_id, outcome_reason, fact_degree + FROM request_history + WHERE queue = ? AND request_id = ? AND entry_id = ?`, + r.queue, requestID, entryID, + )) + if errors.Is(err, sql.ErrNoRows) { + return entity.RequestHistoryEntry{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.RequestHistoryEntry{}, fmt.Errorf("failed to get request history entry request_id=%q entry_id=%q: %w", requestID, entryID, err) + } + return entry, nil +} + +func (r *requestHistoryStore) List(ctx context.Context, requestID string, query storage.RequestHistoryQuery) (ret []entity.RequestHistoryEntry, retErr error) { + op := metrics.Begin(r.scope, "list", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if requestID == "" { + return nil, fmt.Errorf("request history request ID must not be empty") + } + if query.Limit <= 0 { + return nil, fmt.Errorf("request history query limit must be positive") + } + if query.HasCursor && (query.Cursor.TimestampMs <= 0 || query.Cursor.EntryID == "") { + return nil, fmt.Errorf("request history cursor must contain a positive timestamp and entry ID") + } + + statement := listRequestHistoryFirstPageQuery + args := []any{r.queue, requestID} + if query.HasCursor { + statement = listRequestHistoryAfterCursorQuery + args = append(args, query.Cursor.TimestampMs, query.Cursor.TimestampMs, query.Cursor.EntryID) + } + args = append(args, query.Limit) + + rows, err := r.db.QueryContext(ctx, statement, args...) + if err != nil { + return nil, fmt.Errorf("failed to list request history request_id=%q: %w", requestID, err) + } + defer rows.Close() + + entries := make([]entity.RequestHistoryEntry, 0) + for rows.Next() { + entry, err := scanRequestHistoryEntry(rows) + if err != nil { + return nil, fmt.Errorf("failed to scan request history request_id=%q: %w", requestID, err) + } + entries = append(entries, entry) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to iterate request history request_id=%q: %w", requestID, err) + } + return entries, nil +} + +type requestHistoryScanner interface { + Scan(dest ...any) error +} + +func scanRequestHistoryEntry(scanner requestHistoryScanner) (entity.RequestHistoryEntry, error) { + var entry entity.RequestHistoryEntry + err := scanner.Scan( + &entry.Queue, + &entry.RequestID, + &entry.ID, + &entry.TimestampMs, + &entry.State, + &entry.Event, + &entry.RequestVersion, + &entry.SupersededByRequestID, + &entry.BuildID, + &entry.OutcomeReason, + &entry.FactDegree, + ) + return entry, err +} diff --git a/stovepipe/extension/storage/mysql/request_history_store_test.go b/stovepipe/extension/storage/mysql/request_history_store_test.go new file mode 100644 index 000000000..f6744aa5f --- /dev/null +++ b/stovepipe/extension/storage/mysql/request_history_store_test.go @@ -0,0 +1,315 @@ +// 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" + "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 ( + testHistoryQueue = "monorepo/main" + testHistoryRequestID = "request/monorepo/main/1" +) + +var historyColumnNames = []string{ + "queue", "request_id", "entry_id", "timestamp_ms", "state", "event", + "request_version", "superseded_by_request_id", "build_id", "outcome_reason", "fact_degree", +} + +func setupRequestHistoryStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.RequestHistoryStore) { + t.Helper() + db, mock, err := sqlmock.New() + require.NoError(t, err) + return db, mock, NewRequestHistoryStore(db, testMetrics(), testHistoryQueue) +} + +func acceptedHistoryEntry() entity.RequestHistoryEntry { + return entity.RequestHistoryEntry{ + ID: "state/1", + Queue: testHistoryQueue, + RequestID: testHistoryRequestID, + TimestampMs: 1735689600000, + State: entity.RequestStateAccepted, + RequestVersion: 1, + } +} + +func historyRow(entry entity.RequestHistoryEntry) *sqlmock.Rows { + return sqlmock.NewRows(historyColumnNames).AddRow( + entry.Queue, + entry.RequestID, + entry.ID, + entry.TimestampMs, + entry.State, + entry.Event, + entry.RequestVersion, + entry.SupersededByRequestID, + entry.BuildID, + entry.OutcomeReason, + entry.FactDegree, + ) +} + +func TestRequestHistoryStoreCreate(t *testing.T) { + entry := acceptedHistoryEntry() + + tests := []struct { + name string + entry entity.RequestHistoryEntry + setup func(sqlmock.Sqlmock) + wantErrIs error + wantErr bool + }{ + { + name: "success", + entry: entry, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO request_history"). + WithArgs(entry.Queue, entry.RequestID, entry.ID, entry.TimestampMs, entry.State, entry.Event, entry.RequestVersion, entry.SupersededByRequestID, entry.BuildID, entry.OutcomeReason, entry.FactDegree). + WillReturnResult(sqlmock.NewResult(0, 1)) + }, + }, + { + name: "duplicate identity", + entry: entry, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectExec("INSERT INTO request_history"). + WithArgs(entry.Queue, entry.RequestID, entry.ID, entry.TimestampMs, entry.State, entry.Event, entry.RequestVersion, entry.SupersededByRequestID, entry.BuildID, entry.OutcomeReason, entry.FactDegree). + 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_history"). + WithArgs(entry.Queue, entry.RequestID, entry.ID, entry.TimestampMs, entry.State, entry.Event, entry.RequestVersion, entry.SupersededByRequestID, entry.BuildID, entry.OutcomeReason, entry.FactDegree). + WillReturnError(fmt.Errorf("connection reset")) + }, + wantErr: true, + }, + { + name: "invalid entry", + entry: func() entity.RequestHistoryEntry { + invalid := entry + invalid.State = entity.RequestStateUnknown + return invalid + }(), + wantErr: true, + }, + { + name: "wrong queue", + entry: func() entity.RequestHistoryEntry { + wrongQueue := entry + wrongQueue.Queue = "other" + return wrongQueue + }(), + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupRequestHistoryStoreTest(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 TestRequestHistoryStoreGet(t *testing.T) { + future := acceptedHistoryEntry() + future.State = entity.RequestState("future_state") + + tests := []struct { + name string + setup func(sqlmock.Sqlmock) + want entity.RequestHistoryEntry + wantErrIs error + wantErr bool + }{ + { + name: "found without validating future vocabulary", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT queue, request_id, entry_id, timestamp_ms, state, event"). + WithArgs(testHistoryQueue, future.RequestID, future.ID). + WillReturnRows(historyRow(future)) + }, + want: future, + }, + { + name: "not found", + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("SELECT queue, request_id, entry_id, timestamp_ms, state, event"). + WithArgs(testHistoryQueue, 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, entry_id, timestamp_ms, state, event"). + WithArgs(testHistoryQueue, 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 := setupRequestHistoryStoreTest(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 TestRequestHistoryStoreList(t *testing.T) { + first := acceptedHistoryEntry() + second := first + second.ID = "state/2" + second.State = entity.RequestStateProcessing + second.RequestVersion = 2 + + tests := []struct { + name string + query storage.RequestHistoryQuery + setup func(sqlmock.Sqlmock) + want []entity.RequestHistoryEntry + wantErr bool + }{ + { + name: "first page", + query: storage.RequestHistoryQuery{Limit: 2}, + setup: func(mock sqlmock.Sqlmock) { + rows := historyRow(first).AddRow(second.Queue, second.RequestID, second.ID, second.TimestampMs, second.State, second.Event, second.RequestVersion, second.SupersededByRequestID, second.BuildID, second.OutcomeReason, second.FactDegree) + mock.ExpectQuery("ORDER BY timestamp_ms ASC, entry_id ASC LIMIT"). + WithArgs(testHistoryQueue, testHistoryRequestID, 2). + WillReturnRows(rows) + }, + want: []entity.RequestHistoryEntry{first, second}, + }, + { + name: "exclusive cursor", + query: storage.RequestHistoryQuery{ + HasCursor: true, + Cursor: storage.RequestHistoryCursor{ + TimestampMs: first.TimestampMs, + EntryID: first.ID, + }, + Limit: 1, + }, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("timestamp_ms > .+ entry_id > .+ ORDER BY timestamp_ms ASC, entry_id ASC LIMIT"). + WithArgs(testHistoryQueue, testHistoryRequestID, first.TimestampMs, first.TimestampMs, first.ID, 1). + WillReturnRows(historyRow(second)) + }, + want: []entity.RequestHistoryEntry{second}, + }, + { + name: "empty page", + query: storage.RequestHistoryQuery{Limit: 2}, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("ORDER BY timestamp_ms ASC, entry_id ASC LIMIT"). + WithArgs(testHistoryQueue, testHistoryRequestID, 2). + WillReturnRows(sqlmock.NewRows(historyColumnNames)) + }, + want: []entity.RequestHistoryEntry{}, + }, + { + name: "invalid limit", + query: storage.RequestHistoryQuery{}, + wantErr: true, + }, + { + name: "invalid cursor", + query: storage.RequestHistoryQuery{ + HasCursor: true, + Limit: 1, + }, + wantErr: true, + }, + { + name: "database failure", + query: storage.RequestHistoryQuery{Limit: 2}, + setup: func(mock sqlmock.Sqlmock) { + mock.ExpectQuery("ORDER BY timestamp_ms ASC, entry_id ASC LIMIT"). + WithArgs(testHistoryQueue, testHistoryRequestID, 2). + WillReturnError(fmt.Errorf("connection reset")) + }, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, store := setupRequestHistoryStoreTest(t) + defer db.Close() + if tt.setup != nil { + tt.setup(mock) + } + + got, err := store.List(context.Background(), testHistoryRequestID, tt.query) + 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..544a412eb 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), + requestHistoryStore: NewRequestHistoryStore(s.db, s.scope.SubScope("request_history_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 + requestHistoryStore storage.RequestHistoryStore queueStore storage.QueueStore buildStore storage.BuildStore validationFactStore storage.ValidationFactStore @@ -80,6 +82,11 @@ func (f *mysqlStorage) GetRequestURIStore() storage.RequestURIStore { return f.requestURIStore } +// GetRequestHistoryStore returns the MySQL-backed RequestHistoryStore. +func (f *mysqlStorage) GetRequestHistoryStore() storage.RequestHistoryStore { + return f.requestHistoryStore +} + // 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..04cfb8441 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.GetRequestHistoryStore()) assert.NotNil(t, bound.GetQueueStore()) assert.NotNil(t, bound.GetBuildStore()) diff --git a/stovepipe/extension/storage/request_history_store.go b/stovepipe/extension/storage/request_history_store.go new file mode 100644 index 000000000..cdd65dc3f --- /dev/null +++ b/stovepipe/extension/storage/request_history_store.go @@ -0,0 +1,53 @@ +// 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_history_store.go -destination=mock/request_history_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/stovepipe/entity" +) + +// RequestHistoryCursor is the exclusive keyset boundary for a request-history query. +type RequestHistoryCursor struct { + // TimestampMs is the timestamp of the last entry returned by the previous page. + TimestampMs int64 + // EntryID is the ID of the last entry returned by the previous page. + EntryID string +} + +// RequestHistoryQuery specifies one bounded history page within a request partition. +type RequestHistoryQuery struct { + // Cursor is the exclusive continuation boundary when HasCursor is true. + Cursor RequestHistoryCursor + // HasCursor selects whether Cursor participates in the query. + HasCursor bool + // Limit is the positive maximum number of entries returned. + Limit int +} + +// RequestHistoryStore retains immutable occurrences for requests in its bound queue. +type RequestHistoryStore interface { + // Create persists entry and returns ErrAlreadyExists when its stable identity exists. + Create(ctx context.Context, entry entity.RequestHistoryEntry) error + + // Get returns one entry identified by requestID and entryID, or ErrNotFound when absent. + Get(ctx context.Context, requestID, entryID string) (entity.RequestHistoryEntry, error) + + // List returns at most query.Limit entries ordered by timestamp and entry ID ascending. + List(ctx context.Context, requestID string, query RequestHistoryQuery) ([]entity.RequestHistoryEntry, error) +} diff --git a/stovepipe/extension/storage/storage.go b/stovepipe/extension/storage/storage.go index 9aa74f272..8138736a8 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 + // GetRequestHistoryStore returns the RequestHistoryStore instance. + GetRequestHistoryStore() RequestHistoryStore + // 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 8915f850c..3bd486617 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("RequestHistoryStore", func(t *testing.T) { + resetStorage(t, db) + testSuite := new(MySQLRequestHistoryStoreSuite) + 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 } +// MySQLRequestHistoryStoreSuite exercises the MySQL-backed RequestHistoryStore against a real MySQL instance. +type MySQLRequestHistoryStoreSuite struct { + storagesuite.RequestHistoryStoreContractSuite +} + // 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..8b4af6e91 100644 --- a/test/integration/stovepipe/extension/storage/suite.go +++ b/test/integration/stovepipe/extension/storage/suite.go @@ -184,6 +184,134 @@ func (s *QueueStoreContractSuite) TestQueueStore_UpdateSequentialCAS() { assert.Equal(t, int32(3), got.Version) } +// RequestHistoryStoreContractSuite defines contract tests for storage.RequestHistoryStore. +// All RequestHistoryStore implementations must pass these tests. +type RequestHistoryStoreContractSuite struct { + suite.Suite + ctx context.Context + factory storage.Factory +} + +// SetContext sets the context for tests. +func (s *RequestHistoryStoreContractSuite) SetContext(ctx context.Context) { + s.ctx = ctx +} + +// SetFactory provides the Factory that resolves the store under test per queue. +func (s *RequestHistoryStoreContractSuite) SetFactory(factory storage.Factory) { + s.factory = factory +} + +func (s *RequestHistoryStoreContractSuite) storeFor(queue string) storage.RequestHistoryStore { + bound, err := s.factory.For(storage.Config{QueueName: queue}) + s.Require().NoError(err) + return bound.GetRequestHistoryStore() +} + +func (s *RequestHistoryStoreContractSuite) entry(queue, requestID, id string, timestampMs int64, version int32) entity.RequestHistoryEntry { + return entity.RequestHistoryEntry{ + ID: id, + Queue: queue, + RequestID: requestID, + TimestampMs: timestampMs, + State: entity.RequestStateAccepted, + RequestVersion: version, + } +} + +// TestRequestHistoryStore_CreateAndGet verifies all occurrence fields round-trip unchanged. +func (s *RequestHistoryStoreContractSuite) TestRequestHistoryStore_CreateAndGet() { + const ( + queue = "contract/history-create" + requestID = "request/contract/history-create/1" + ) + entry := entity.RequestHistoryEntry{ + ID: "fact/repository", + Queue: queue, + RequestID: requestID, + TimestampMs: 1735689600000, + Event: entity.RequestEventValidationFactRecorded, + FactDegree: 0.5, + } + 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) +} + +// TestRequestHistoryStore_CreateAlreadyExists verifies stable identities are create-only. +func (s *RequestHistoryStoreContractSuite) TestRequestHistoryStore_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) +} + +// TestRequestHistoryStore_GetNotFound verifies a missing stable identity returns ErrNotFound. +func (s *RequestHistoryStoreContractSuite) TestRequestHistoryStore_GetNotFound() { + _, err := s.storeFor("contract/history-missing").Get(s.ctx, "request/missing/1", "state/1") + assert.True(s.T(), storage.IsNotFound(err)) +} + +// TestRequestHistoryStore_List verifies chronological ordering, equal-time tie breaking, and exclusive cursors. +func (s *RequestHistoryStoreContractSuite) TestRequestHistoryStore_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.RequestHistoryEntry{last, second, first} { + require.NoError(s.T(), store.Create(s.ctx, entry)) + } + + page, err := store.List(s.ctx, requestID, storage.RequestHistoryQuery{Limit: 2}) + require.NoError(s.T(), err) + assert.Equal(s.T(), []entity.RequestHistoryEntry{first, second}, page) + + page, err = store.List(s.ctx, requestID, storage.RequestHistoryQuery{ + Cursor: storage.RequestHistoryCursor{ + TimestampMs: first.TimestampMs, + EntryID: first.ID, + }, + HasCursor: true, + Limit: 2, + }) + require.NoError(s.T(), err) + assert.Equal(s.T(), []entity.RequestHistoryEntry{second, last}, page) +} + +// TestRequestHistoryStore_QueueIsolation verifies identical request and entry IDs remain queue-scoped. +func (s *RequestHistoryStoreContractSuite) TestRequestHistoryStore_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 {