feat(storage): add DLQ reconciliation persistence groundwork [runtime BLOCKED]

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",

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
+}

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",

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.

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
+}

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",

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",

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
+}