diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel index 6e1561ab1..6b04c55a1 100644 --- a/submitqueue/entity/BUILD.bazel +++ b/submitqueue/entity/BUILD.bazel @@ -11,6 +11,7 @@ go_library( "change_provider.go", "change_record.go", "conflict.go", + "dlq_reconciliation.go", "land.go", "list.go", "path_build.go", diff --git a/submitqueue/entity/dlq_reconciliation.go b/submitqueue/entity/dlq_reconciliation.go new file mode 100644 index 000000000..10de01277 --- /dev/null +++ b/submitqueue/entity/dlq_reconciliation.go @@ -0,0 +1,43 @@ +// 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 + +// DLQReconciliationSource identifies a logical message intent within a queue. +// Its values are immutable and compared verbatim, including trailing spaces. +type DLQReconciliationSource struct { + // Topic is the logical source topic, not a backend-specific topic name. + Topic string + // PartitionKey is the original opaque source partition key. + PartitionKey string + // MessageID is the opaque logical-cause ID within the topic and partition. + MessageID string +} + +// DLQReconciliation is a versioned snapshot of a final message's queue repair. +// Queue and Source together form its immutable primary key. +type DLQReconciliation struct { + // ID is an immutable positive canonical decimal, unique within Queue. + ID string + // Queue is the immutable queue name scoping the source intent and batch IDs. + Queue string + // Source is the immutable logical message identity within Queue. + Source DLQReconciliationSource + // BatchIDs retains every captured target ID; successive snapshots never drop IDs. + BatchIDs []string + // Completed is monotone: a completed obligation never becomes pending again. + Completed bool + // Version starts at one and increases on each successful conditional replacement. + Version int32 +} diff --git a/submitqueue/extension/storage/BUILD.bazel b/submitqueue/extension/storage/BUILD.bazel index ec9a4d025..061e70c48 100644 --- a/submitqueue/extension/storage/BUILD.bazel +++ b/submitqueue/extension/storage/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "batch_store.go", "build_store.go", "change_store.go", + "dlq_reconciliation_store.go", "path_build_store.go", "queue_batch_state_store.go", "request_batch_store.go", diff --git a/submitqueue/extension/storage/README.md b/submitqueue/extension/storage/README.md index 026429d84..fda6b6262 100644 --- a/submitqueue/extension/storage/README.md +++ b/submitqueue/extension/storage/README.md @@ -72,3 +72,11 @@ Take the first branch that applies: 3. **Map** — a pipeline controller needs the lookup at runtime → a dedicated mapping store keyed by the attribute. The bar for 3 is a hot-path need: one mapping per access path, never per attribute, and never for ops/debug queries — run those against SQL replicas directly. But don't contort 1–2 to dodge a legitimate 3; a primary key that hashes half the entity's fields, or an aggregate bloated into listing everything, is the same duplication hidden in a worse place. + +## Final-message reconciliation snapshots + +`DLQReconciliationStore` belongs only to the orchestrator aggregate. Its primary key is the bound queue plus the logical source topic, original partition key, and message ID, all compared verbatim. A message ID names its logical cause within that scope: replay repairs the same intent, while a distinct cause needs a distinct ID. A backend delivery ID, attempt, or timestamp does not identify a new obligation. + +Records have immutable decimal IDs and full versioned snapshots of captured batch IDs and completion. Callers retain every captured ID, preserve completion monotonically, and clone slices before constructing replacements. An unsuccessful Create caused by duplication requires loading the first persisted record, including its ID. Update guards both that immutable ID and the explicit old version and writes the supplied new version without arithmetic. The ordinary read-after-write contract applies. + +Completed markers outlive transport message garbage collection so acknowledgment loss cannot recreate completed work. There is no deletion or automatic retention policy; deployments must budget and monitor exceptional final-message volume independently. No cross-service storage access is introduced. This persistence foundation does not enable queue repair: runtime reconciliation still requires authoritative merge-outcome protection and queue-wide planner fencing. diff --git a/submitqueue/extension/storage/dlq_reconciliation_store.go b/submitqueue/extension/storage/dlq_reconciliation_store.go new file mode 100644 index 000000000..8c8b91056 --- /dev/null +++ b/submitqueue/extension/storage/dlq_reconciliation_store.go @@ -0,0 +1,41 @@ +// 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=dlq_reconciliation_store.go -destination=mock/dlq_reconciliation_store_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +// DLQReconciliationStore retains snapshots independently of transport message GC. +// Only the full source key within the bound queue addresses a record, never its ID. +// Callers clone BatchIDs before changes and preserve captured targets and completion. +type DLQReconciliationStore interface { + // Get returns the source intent's snapshot, or ErrNotFound if none exists. + Get(ctx context.Context, source entity.DLQReconciliationSource) (entity.DLQReconciliation, error) + + // Create returns ErrAlreadyExists if this source intent already has a record. + // A losing creator must Get the winner rather than reuse its proposed record ID. + Create(ctx context.Context, record entity.DLQReconciliation) error + + // Update replaces all mutable fields if ID and oldVersion match the stored record, + // or returns ErrVersionMismatch without writing. Queue, Source and ID are immutable. + // record.Version is ignored; the caller computes newVersion and assigns it only + // after success. Callers retain Completed records; this contract has no Delete. + Update(ctx context.Context, record entity.DLQReconciliation, oldVersion, newVersion int32) error +} diff --git a/submitqueue/extension/storage/mock/BUILD.bazel b/submitqueue/extension/storage/mock/BUILD.bazel index 49f2907c6..a3ec2d370 100644 --- a/submitqueue/extension/storage/mock/BUILD.bazel +++ b/submitqueue/extension/storage/mock/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "batch_store_mock.go", "build_store_mock.go", "change_store_mock.go", + "dlq_reconciliation_store_mock.go", "path_build_store_mock.go", "queue_batch_state_store_mock.go", "request_batch_store_mock.go", diff --git a/submitqueue/extension/storage/mock/dlq_reconciliation_store_mock.go b/submitqueue/extension/storage/mock/dlq_reconciliation_store_mock.go new file mode 100644 index 000000000..abccad1c4 --- /dev/null +++ b/submitqueue/extension/storage/mock/dlq_reconciliation_store_mock.go @@ -0,0 +1,85 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: dlq_reconciliation_store.go +// +// Generated by this command: +// +// mockgen -source=dlq_reconciliation_store.go -destination=mock/dlq_reconciliation_store_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/submitqueue/entity" + gomock "go.uber.org/mock/gomock" +) + +// MockDLQReconciliationStore is a mock of DLQReconciliationStore interface. +type MockDLQReconciliationStore struct { + ctrl *gomock.Controller + recorder *MockDLQReconciliationStoreMockRecorder + isgomock struct{} +} + +// MockDLQReconciliationStoreMockRecorder is the mock recorder for MockDLQReconciliationStore. +type MockDLQReconciliationStoreMockRecorder struct { + mock *MockDLQReconciliationStore +} + +// NewMockDLQReconciliationStore creates a new mock instance. +func NewMockDLQReconciliationStore(ctrl *gomock.Controller) *MockDLQReconciliationStore { + mock := &MockDLQReconciliationStore{ctrl: ctrl} + mock.recorder = &MockDLQReconciliationStoreMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockDLQReconciliationStore) EXPECT() *MockDLQReconciliationStoreMockRecorder { + return m.recorder +} + +// Create mocks base method. +func (m *MockDLQReconciliationStore) Create(ctx context.Context, record entity.DLQReconciliation) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Create", ctx, record) + ret0, _ := ret[0].(error) + return ret0 +} + +// Create indicates an expected call of Create. +func (mr *MockDLQReconciliationStoreMockRecorder) Create(ctx, record any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Create", reflect.TypeOf((*MockDLQReconciliationStore)(nil).Create), ctx, record) +} + +// Get mocks base method. +func (m *MockDLQReconciliationStore) Get(ctx context.Context, source entity.DLQReconciliationSource) (entity.DLQReconciliation, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Get", ctx, source) + ret0, _ := ret[0].(entity.DLQReconciliation) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Get indicates an expected call of Get. +func (mr *MockDLQReconciliationStoreMockRecorder) Get(ctx, source any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockDLQReconciliationStore)(nil).Get), ctx, source) +} + +// Update mocks base method. +func (m *MockDLQReconciliationStore) Update(ctx context.Context, record entity.DLQReconciliation, oldVersion, newVersion int32) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Update", ctx, record, oldVersion, newVersion) + ret0, _ := ret[0].(error) + return ret0 +} + +// Update indicates an expected call of Update. +func (mr *MockDLQReconciliationStoreMockRecorder) Update(ctx, record, oldVersion, newVersion any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockDLQReconciliationStore)(nil).Update), ctx, record, oldVersion, newVersion) +} diff --git a/submitqueue/orchestrator/extension/storage/mock/storage_mock.go b/submitqueue/orchestrator/extension/storage/mock/storage_mock.go index 9169780e7..7c2a9b32c 100644 --- a/submitqueue/orchestrator/extension/storage/mock/storage_mock.go +++ b/submitqueue/orchestrator/extension/storage/mock/storage_mock.go @@ -136,6 +136,20 @@ func (mr *MockStorageMockRecorder) GetChangeStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetChangeStore", reflect.TypeOf((*MockStorage)(nil).GetChangeStore)) } +// GetDLQReconciliationStore mocks base method. +func (m *MockStorage) GetDLQReconciliationStore() storage.DLQReconciliationStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetDLQReconciliationStore") + ret0, _ := ret[0].(storage.DLQReconciliationStore) + return ret0 +} + +// GetDLQReconciliationStore indicates an expected call of GetDLQReconciliationStore. +func (mr *MockStorageMockRecorder) GetDLQReconciliationStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetDLQReconciliationStore", reflect.TypeOf((*MockStorage)(nil).GetDLQReconciliationStore)) +} + // GetPathBuildStore mocks base method. func (m *MockStorage) GetPathBuildStore() storage.PathBuildStore { m.ctrl.T.Helper() diff --git a/submitqueue/orchestrator/extension/storage/mysql/BUILD.bazel b/submitqueue/orchestrator/extension/storage/mysql/BUILD.bazel index 1ebf683c4..bf2ed2ed9 100644 --- a/submitqueue/orchestrator/extension/storage/mysql/BUILD.bazel +++ b/submitqueue/orchestrator/extension/storage/mysql/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "batch_store.go", "build_store.go", "change_store.go", + "dlq_reconciliation_store.go", "path_build_store.go", "queue_batch_state_store.go", "request_batch_store.go", @@ -33,6 +34,7 @@ go_test( "batch_store_test.go", "build_store_test.go", "change_store_test.go", + "dlq_reconciliation_store_test.go", "queue_batch_state_store_test.go", "request_batch_store_test.go", "request_store_test.go", diff --git a/submitqueue/orchestrator/extension/storage/mysql/dlq_reconciliation_store.go b/submitqueue/orchestrator/extension/storage/mysql/dlq_reconciliation_store.go new file mode 100644 index 000000000..76bf3596e --- /dev/null +++ b/submitqueue/orchestrator/extension/storage/mysql/dlq_reconciliation_store.go @@ -0,0 +1,116 @@ +// 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/go-sql-driver/mysql" + "github.com/uber-go/tally" + + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +type dlqReconciliationStore struct { + db *sql.DB + scope tally.Scope + queue string +} + +// NewDLQReconciliationStore binds reconciliation records to one queue. +func NewDLQReconciliationStore(db *sql.DB, scope tally.Scope, queue string) storage.DLQReconciliationStore { + return &dlqReconciliationStore{db: db, scope: scope, queue: queue} +} + +func (s *dlqReconciliationStore) Get(ctx context.Context, source entity.DLQReconciliationSource) (ret entity.DLQReconciliation, retErr error) { + op := metrics.Begin(s.scope, "get", metrics.StorageLatencyBuckets) + defer func() { storage.CompleteReadMetricsOp(op, retErr) }() + + var batchIDsJSON []byte + record := entity.DLQReconciliation{Queue: s.queue, Source: source} + err := s.db.QueryRowContext(ctx, + "SELECT id, batch_ids, completed, version FROM dlq_reconciliation WHERE queue = ? AND source_topic = ? AND source_partition_key = ? AND source_message_id = ?", + s.queue, source.Topic, source.PartitionKey, source.MessageID, + ).Scan(&record.ID, &batchIDsJSON, &record.Completed, &record.Version) + if errors.Is(err, sql.ErrNoRows) { + return entity.DLQReconciliation{}, storage.WrapNotFound(err) + } + if err != nil { + return entity.DLQReconciliation{}, fmt.Errorf("get reconciliation queue=%q source=%+v: %w", s.queue, source, err) + } + if err := json.Unmarshal(batchIDsJSON, &record.BatchIDs); err != nil { + return entity.DLQReconciliation{}, fmt.Errorf("decode reconciliation targets queue=%q source=%+v: %w", s.queue, source, err) + } + return record, nil +} + +func (s *dlqReconciliationStore) Create(ctx context.Context, record entity.DLQReconciliation) (retErr error) { + op := metrics.Begin(s.scope, "create", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if record.Queue != s.queue { + return fmt.Errorf("reconciliation queue %q does not match bound queue %q", record.Queue, s.queue) + } + batchIDsJSON, err := json.Marshal(record.BatchIDs) + if err != nil { + return fmt.Errorf("encode reconciliation targets: %w", err) + } + _, err = s.db.ExecContext(ctx, + "INSERT INTO dlq_reconciliation (queue, source_topic, source_partition_key, source_message_id, id, batch_ids, completed, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + record.Queue, record.Source.Topic, record.Source.PartitionKey, record.Source.MessageID, record.ID, batchIDsJSON, record.Completed, record.Version, + ) + if err != nil { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == mysqlErrDuplicateEntry { + return fmt.Errorf("reconciliation queue=%q source=%+v: %w", record.Queue, record.Source, storage.ErrAlreadyExists) + } + return fmt.Errorf("create reconciliation queue=%q source=%+v: %w", record.Queue, record.Source, err) + } + return nil +} + +func (s *dlqReconciliationStore) Update(ctx context.Context, record entity.DLQReconciliation, oldVersion, newVersion int32) (retErr error) { + op := metrics.Begin(s.scope, "update", metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if record.Queue != s.queue { + return fmt.Errorf("reconciliation queue %q does not match bound queue %q", record.Queue, s.queue) + } + batchIDsJSON, err := json.Marshal(record.BatchIDs) + if err != nil { + return fmt.Errorf("encode reconciliation targets: %w", err) + } + result, err := s.db.ExecContext(ctx, + "UPDATE dlq_reconciliation SET batch_ids = ?, completed = ?, version = ? WHERE queue = ? AND source_topic = ? AND source_partition_key = ? AND source_message_id = ? AND id = ? AND version = ?", + batchIDsJSON, record.Completed, newVersion, record.Queue, record.Source.Topic, record.Source.PartitionKey, record.Source.MessageID, record.ID, oldVersion, + ) + if err != nil { + return fmt.Errorf("update reconciliation queue=%q source=%+v: %w", record.Queue, record.Source, err) + } + rowsAffected, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("read reconciliation update result: %w", err) + } + if rowsAffected != 1 { + return fmt.Errorf("reconciliation queue=%q source=%+v version=%d: %w", record.Queue, record.Source, oldVersion, storage.ErrVersionMismatch) + } + return nil +} diff --git a/submitqueue/orchestrator/extension/storage/mysql/dlq_reconciliation_store_test.go b/submitqueue/orchestrator/extension/storage/mysql/dlq_reconciliation_store_test.go new file mode 100644 index 000000000..d79f2207c --- /dev/null +++ b/submitqueue/orchestrator/extension/storage/mysql/dlq_reconciliation_store_test.go @@ -0,0 +1,144 @@ +// 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" + "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/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +func reconciliationSnapshot() entity.DLQReconciliation { + return entity.DLQReconciliation{ + ID: "7", Queue: "queue", + Source: entity.DLQReconciliationSource{Topic: "speculate.final", PartitionKey: "partition", MessageID: "message"}, + BatchIDs: []string{"1", "2"}, Completed: true, Version: 99, + } +} + +func TestDLQReconciliationStore_Get(t *testing.T) { + want := reconciliationSnapshot() + tests := []struct { + name string + rows *sqlmock.Rows + queryErr error + wantErr error + }{ + {name: "round trip", rows: sqlmock.NewRows([]string{"id", "batch_ids", "completed", "version"}).AddRow(want.ID, `["1","2"]`, want.Completed, want.Version)}, + {name: "missing", queryErr: sql.ErrNoRows, wantErr: storage.ErrNotFound}, + {name: "database failure", queryErr: errors.New("database unavailable")}, + {name: "corrupt targets", rows: sqlmock.NewRows([]string{"id", "batch_ids", "completed", "version"}).AddRow(want.ID, `{}`, true, 1)}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + query := mock.ExpectQuery("SELECT id, batch_ids, completed, version FROM dlq_reconciliation"). + WithArgs(want.Queue, want.Source.Topic, want.Source.PartitionKey, want.Source.MessageID) + if tt.queryErr != nil { + query.WillReturnError(tt.queryErr) + } else { + query.WillReturnRows(tt.rows) + } + store := NewDLQReconciliationStore(db, testMetrics(), want.Queue) + got, err := store.Get(context.Background(), want.Source) + if tt.name == "round trip" { + require.NoError(t, err) + assert.Equal(t, want, got) + } else { + require.Error(t, err) + if tt.wantErr != nil { + assert.ErrorIs(t, err, tt.wantErr) + } + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestDLQReconciliationStore_Write(t *testing.T) { + databaseFailure := errors.New("database unavailable") + tests := []struct { + name string + update bool + otherQueue bool + execErr error + result sql.Result + wantErr error + }{ + {name: "create"}, + {name: "duplicate create", execErr: &mysql.MySQLError{Number: mysqlErrDuplicateEntry}, wantErr: storage.ErrAlreadyExists}, + {name: "create failure", execErr: databaseFailure, wantErr: databaseFailure}, + {name: "create queue mismatch", otherQueue: true}, + {name: "replace with explicit version", update: true}, + {name: "stale replacement", update: true, result: sqlmock.NewResult(0, 0), wantErr: storage.ErrVersionMismatch}, + {name: "update failure", update: true, execErr: databaseFailure, wantErr: databaseFailure}, + {name: "update result failure", update: true, result: sqlmock.NewErrorResult(databaseFailure), wantErr: databaseFailure}, + {name: "update queue mismatch", update: true, otherQueue: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + record := reconciliationSnapshot() + store := NewDLQReconciliationStore(db, testMetrics(), record.Queue) + if tt.otherQueue { + record.Queue = "other" + } + if !tt.otherQueue { + var query *sqlmock.ExpectedExec + if tt.update { + query = mock.ExpectExec("UPDATE dlq_reconciliation SET batch_ids"). + WithArgs([]byte(`["1","2"]`), record.Completed, int32(4), record.Queue, record.Source.Topic, record.Source.PartitionKey, record.Source.MessageID, record.ID, int32(3)) + } else { + query = mock.ExpectExec("INSERT INTO dlq_reconciliation"). + WithArgs(record.Queue, record.Source.Topic, record.Source.PartitionKey, record.Source.MessageID, record.ID, []byte(`["1","2"]`), record.Completed, record.Version) + } + if tt.execErr != nil { + query.WillReturnError(tt.execErr) + } else if tt.result != nil { + query.WillReturnResult(tt.result) + } else { + query.WillReturnResult(sqlmock.NewResult(0, 1)) + } + } + if tt.update { + err = store.Update(context.Background(), record, 3, 4) + } else { + err = store.Create(context.Background(), record) + } + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else if tt.otherQueue { + require.Error(t, err) + } else { + require.NoError(t, err) + } + assert.Equal(t, int32(99), record.Version) + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} diff --git a/submitqueue/orchestrator/extension/storage/mysql/schema/README.md b/submitqueue/orchestrator/extension/storage/mysql/schema/README.md index 14c09d619..0f5b16e2e 100644 --- a/submitqueue/orchestrator/extension/storage/mysql/schema/README.md +++ b/submitqueue/orchestrator/extension/storage/mysql/schema/README.md @@ -27,3 +27,11 @@ Terminal-state records (and their batches) accumulate as the queue processes wor ### Composite primary key: `(queue, uri, request_id)` The `change` table records per-URI claims by in-flight requests. `request_id` is part of the primary key so that concurrent claims on the same URI by different requests coexist as distinct rows — a same-request retry collides on the PK and is a no-op (`INSERT IGNORE`), while a different-request claim is a new row that `GetByURI` surfaces for overlap detection. `queue` leads the key so queue-scoped lookups are primary-key-prefix scans and the table is shardable by queue. + +## dlq_reconciliation table + +`dlq_reconciliation` stores final-message reconciliation snapshots under `(queue, source_topic, source_partition_key, source_message_id)`, with no secondary indexes or lookup by the generated record ID. The record ID and source identity are immutable; mutable fields are replaced under the record ID and explicit version guard. The store rejects another queue's snapshot and reads from the primary under the storage read-after-write contract. + +Queue and logical topic use `VARBINARY(255)` to preserve exact bytes, including trailing spaces. Opaque partition and message IDs use `VARCHAR(255)` with MySQL 8's `utf8mb4_0900_bin` NO PAD collation; unlike legacy binary collations, it does not equate an identifier with its trailing-space variant. The primary key's maximum payload is 2,550 bytes (255 + 255 + 1,020 + 1,020), below InnoDB's 3,072-byte limit. Identifiers are stored directly, never hashed or re-encoded. + +This additive table supplies persistence only; no final handler or planner uses it yet. Keep pending and completed rows independently of message-row garbage collection, including during rollout and rollback. Removing application access does not authorize dropping records: acknowledgment-loss replay needs the completed markers. diff --git a/submitqueue/orchestrator/extension/storage/mysql/schema/dlq_reconciliation.sql b/submitqueue/orchestrator/extension/storage/mysql/schema/dlq_reconciliation.sql new file mode 100644 index 000000000..a29499cbd --- /dev/null +++ b/submitqueue/orchestrator/extension/storage/mysql/schema/dlq_reconciliation.sql @@ -0,0 +1,14 @@ +-- VARBINARY keeps queue/topic byte-exact, including trailing spaces. +-- utf8mb4_0900_bin is NO PAD: opaque partition/message IDs remain distinct. +-- Maximum primary-key payload: 255 + 255 + 4*255 + 4*255 = 2550 bytes. +CREATE TABLE IF NOT EXISTS dlq_reconciliation ( + queue VARBINARY(255) NOT NULL, + source_topic VARBINARY(255) NOT NULL, + source_partition_key VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin NOT NULL, + source_message_id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin NOT NULL, + id VARCHAR(20) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, + batch_ids JSON NOT NULL, + completed BOOLEAN NOT NULL, + version INT NOT NULL, + PRIMARY KEY (queue, source_topic, source_partition_key, source_message_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_bin; diff --git a/submitqueue/orchestrator/extension/storage/mysql/storage.go b/submitqueue/orchestrator/extension/storage/mysql/storage.go index 1309ee86d..6d2eee28e 100644 --- a/submitqueue/orchestrator/extension/storage/mysql/storage.go +++ b/submitqueue/orchestrator/extension/storage/mysql/storage.go @@ -59,6 +59,7 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { queueBatchStateStore: NewQueueBatchStateStore(s.db, s.scope.SubScope("queue_batch_state_store"), queueName), buildStore: NewBuildStore(s.db, s.scope.SubScope("build_store"), queueName), speculationPathSetStore: NewSpeculationPathSetStore(s.db, s.scope.SubScope("speculation_path_set_store"), queueName), + dlqReconciliationStore: NewDLQReconciliationStore(s.db, s.scope.SubScope("dlq_reconciliation_store"), queueName), pathBuildStore: NewPathBuildStore(s.db, s.scope.SubScope("path_build_store"), queueName), }, nil } @@ -78,6 +79,7 @@ type boundStorage struct { queueBatchStateStore basestorage.QueueBatchStateStore buildStore basestorage.BuildStore speculationPathSetStore basestorage.SpeculationPathSetStore + dlqReconciliationStore basestorage.DLQReconciliationStore pathBuildStore basestorage.PathBuildStore } @@ -128,3 +130,8 @@ func (f *boundStorage) GetSpeculationPathSetStore() basestorage.SpeculationPathS func (f *boundStorage) GetPathBuildStore() basestorage.PathBuildStore { return f.pathBuildStore } + +// GetDLQReconciliationStore returns the bound MySQL-backed reconciliation store. +func (f *boundStorage) GetDLQReconciliationStore() basestorage.DLQReconciliationStore { + return f.dlqReconciliationStore +} diff --git a/submitqueue/orchestrator/extension/storage/storage.go b/submitqueue/orchestrator/extension/storage/storage.go index 5b6ebb4c0..a862aad9c 100644 --- a/submitqueue/orchestrator/extension/storage/storage.go +++ b/submitqueue/orchestrator/extension/storage/storage.go @@ -68,6 +68,9 @@ type Storage interface { // GetSpeculationPathSetStore returns the SpeculationPathSetStore instance. GetSpeculationPathSetStore() basestorage.SpeculationPathSetStore + // GetDLQReconciliationStore returns the queue-bound reconciliation snapshots. + GetDLQReconciliationStore() basestorage.DLQReconciliationStore + // GetPathBuildStore returns the PathBuildStore instance. GetPathBuildStore() basestorage.PathBuildStore } diff --git a/test/integration/submitqueue/extension/storage/BUILD.bazel b/test/integration/submitqueue/extension/storage/BUILD.bazel index a7a2f1efe..e639b3711 100644 --- a/test/integration/submitqueue/extension/storage/BUILD.bazel +++ b/test/integration/submitqueue/extension/storage/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library") go_library( name = "go_default_library", srcs = [ + "dlq_reconciliation.go", "request_receipt.go", "suite.go", ], diff --git a/test/integration/submitqueue/extension/storage/dlq_reconciliation.go b/test/integration/submitqueue/extension/storage/dlq_reconciliation.go new file mode 100644 index 000000000..438619099 --- /dev/null +++ b/test/integration/submitqueue/extension/storage/dlq_reconciliation.go @@ -0,0 +1,110 @@ +// 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 + +import ( + "fmt" + "slices" + "strings" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +func (s *StorageContractSuite) TestStorage_DLQReconciliationSnapshots() { + t := s.T() + ctx := s.ctx + store := s.forQueue("dlq-snapshots").GetDLQReconciliationStore() + source := entity.DLQReconciliationSource{Topic: "speculate.final", PartitionKey: "queue", MessageID: "intent"} + _, err := store.Get(ctx, source) + require.ErrorIs(t, err, storage.ErrNotFound) + + initial := entity.DLQReconciliation{ID: "1", Queue: "dlq-snapshots", Source: source, BatchIDs: []string{"10"}, Version: 1} + require.NoError(t, store.Create(ctx, initial)) + contender := initial + contender.ID = "2" + require.ErrorIs(t, store.Create(ctx, contender), storage.ErrAlreadyExists) + winner, err := store.Get(ctx, source) + require.NoError(t, err) + assert.Equal(t, initial, winner) + + completed := initial + completed.BatchIDs = append(slices.Clone(initial.BatchIDs), "11") + completed.Completed = true + completed.Version = 99 + require.NoError(t, store.Update(ctx, completed, 1, 2)) + assert.Equal(t, int32(99), completed.Version) + assert.Equal(t, []string{"10"}, initial.BatchIDs) + require.ErrorIs(t, store.Update(ctx, initial, 1, 2), storage.ErrVersionMismatch) + require.ErrorIs(t, store.Update(ctx, contender, 2, 3), storage.ErrVersionMismatch) + + reloadedStore := s.forQueue(initial.Queue).GetDLQReconciliationStore() + persisted, err := reloadedStore.Get(ctx, source) + require.NoError(t, err) + completed.Version = 2 + assert.Equal(t, completed, persisted) +} + +func (s *StorageContractSuite) TestStorage_DLQReconciliationSourceIdentity() { + t := s.T() + base := entity.DLQReconciliation{ + Queue: "dlq-identity", Source: entity.DLQReconciliationSource{Topic: "speculate.final", PartitionKey: "partition", MessageID: "message"}, + BatchIDs: []string{}, Version: 1, + } + tests := []struct { + name string + change func(*entity.DLQReconciliation) + }{ + {name: "base", change: func(*entity.DLQReconciliation) {}}, + {name: "other queue", change: func(r *entity.DLQReconciliation) { r.Queue += "-other" }}, + {name: "queue case", change: func(r *entity.DLQReconciliation) { r.Queue = strings.ToUpper(r.Queue) }}, + {name: "queue trailing space", change: func(r *entity.DLQReconciliation) { r.Queue += " " }}, + {name: "other topic", change: func(r *entity.DLQReconciliation) { r.Source.Topic = "build.final" }}, + {name: "topic case", change: func(r *entity.DLQReconciliation) { r.Source.Topic = strings.ToUpper(r.Source.Topic) }}, + {name: "topic trailing space", change: func(r *entity.DLQReconciliation) { r.Source.Topic += " " }}, + {name: "partition case", change: func(r *entity.DLQReconciliation) { r.Source.PartitionKey = "Partition" }}, + {name: "partition trailing space", change: func(r *entity.DLQReconciliation) { r.Source.PartitionKey += " " }}, + {name: "message case", change: func(r *entity.DLQReconciliation) { r.Source.MessageID = "Message" }}, + {name: "message trailing space", change: func(r *entity.DLQReconciliation) { r.Source.MessageID += " " }}, + {name: "maximum key width", change: func(r *entity.DLQReconciliation) { + r.Queue = strings.Repeat("q", 255) + r.Source.Topic = strings.Repeat("t", 255) + r.Source.PartitionKey = strings.Repeat("🌳", 255) + r.Source.MessageID = strings.Repeat("🚀", 255) + }}, + } + records := make([]entity.DLQReconciliation, 0, len(tests)) + for i, tt := range tests { + record := base + record.ID = fmt.Sprint(i + 1) + tt.change(&record) + s.Run(tt.name, func() { + require.NoError(s.T(), s.forQueue(record.Queue).GetDLQReconciliationStore().Create(s.ctx, record)) + }) + records = append(records, record) + } + for _, record := range records { + got, err := s.forQueue(record.Queue).GetDLQReconciliationStore().Get(s.ctx, record.Source) + require.NoError(t, err) + assert.Equal(t, record, got) + } + otherQueue := s.forQueue("dlq-unrelated").GetDLQReconciliationStore() + _, err := otherQueue.Get(s.ctx, base.Source) + require.ErrorIs(t, err, storage.ErrNotFound) + require.Error(t, otherQueue.Create(s.ctx, records[0])) + require.Error(t, otherQueue.Update(s.ctx, records[0], 1, 2)) +}