feat(queue/sql): add subscriber with partition leasing and offset tracking

- Implement Subscriber interface with partition-based message polling
- Add partition lease management for distributed workers
- Add offset tracking per partition for consumption progress
- Extend MessageStore interface with FetchByOffset, SetVisibilityTimeout, MoveToDLQ
- Add OffsetStore interface for offset management
- Add PartitionLeaseStore interface for partition leasing
- Generate mocks for all three store interfaces
- Add standardized metric tags (topic + partition_key) across all operations
- Add comprehensive test coverage for subscription, ack/nack, and error handling

Author: behinddwalls

extensions/queue/sql/BUILD.bazel

@@ -4,15 +4,18 @@ go_library(
     name = "sql",
     srcs = [
         "config.go",
+        "errors.go",
         "mock_stores.go",
         "publisher.go",
         "stores.go",
+        "subscriber.go",
         "validation.go",
     ],
     importpath = "github.com/uber/submitqueue/extensions/queue/sql",
     visibility = ["//visibility:public"],
     deps = [
         "//entities/queue",
+        "//extensions/queue",
         "@com_github_uber_go_tally_v4//:tally",
         "@org_uber_go_mock//gomock",
         "@org_uber_go_zap//:zap",
@@ -24,6 +27,7 @@ go_test(
     srcs = [
         "config_test.go",
         "publisher_test.go",
+        "subscriber_test.go",
     ],
     embed = [":sql"],
     deps = [

extensions/queue/sql/errors.go

@@ -0,0 +1,12 @@
+package sql
+
+import "fmt"
+
+// ErrAlreadyAcknowledged is returned when attempting to ack/nack a delivery that was already processed
+type ErrAlreadyAcknowledged struct {
+	DeliveryID string
+}
+
+func (e *ErrAlreadyAcknowledged) Error() string {
+	return fmt.Sprintf("delivery %s already acknowledged or nacked", e.DeliveryID)
+}

extensions/queue/sql/mock_stores.go

@@ -8,8 +8,80 @@ import (
 	"github.com/uber/submitqueue/entities/queue"
 )
 
+// StoreConfig contains configuration needed by store implementations
+
+// MessageRow represents a row from the messages table
+type MessageRow struct {
+	// Offset is the auto-incrementing sequence number for message ordering within a partition
+	Offset int64
+	// ID is the unique message identifier
+	ID string
+	// Payload is the message body in bytes
+	Payload []byte
+	// Metadata contains key-value pairs for message attributes
+	Metadata map[string]string
+	// PartitionKey determines which partition this message belongs to for ordering guarantees
+	PartitionKey string
+	// RetryCount tracks how many times this message has been retried on the current topic
+	RetryCount int
+	// PublishedAt is the Unix timestamp in milliseconds when message was published
+	PublishedAt int64
+}
+
 // MessageStore handles message table operations
 type MessageStore interface {
 	// Insert inserts messages into the topic table
 	Insert(ctx context.Context, topic string, messages []queue.Message) error
+
+	// Delete deletes a message by ID
+	Delete(ctx context.Context, topic string, messageID string) error
+
+	// FetchByOffset fetches messages with offset > currentOffset for a specific partition
+	// Only fetches visible messages (invisible_until <= now)
+	// Atomically sets invisible_until and increments retry_count
+	FetchByOffset(ctx context.Context, topic string, partitionKey string, currentOffset int64, limit int) ([]MessageRow, error)
+
+	// MoveToDLQ moves a message to the dead letter queue
+	MoveToDLQ(ctx context.Context, topic string, messageID string, failureCount int, lastError string) error
+
+	// SetVisibilityTimeout sets the invisible_until timestamp for a message
+	// visibilityTimeoutMillis: milliseconds from now to hide the message
+	// If visibilityTimeoutMillis is 0, makes the message visible immediately
+	// If visibilityTimeoutMillis > 0, makes the message invisible until now + visibilityTimeoutMillis
+	SetVisibilityTimeout(ctx context.Context, topic string, messageID string, visibilityTimeoutMillis int64) error
+}
+
+// OffsetStore handles offset table operations for per-partition offset tracking
+type OffsetStore interface {
+	// Initialize creates an offset entry for a topic+partition if it doesn't exist
+	Initialize(ctx context.Context, topic string, partitionKey string) error
+
+	// GetAckedOffset returns the current acked offset for a topic+partition
+	GetAckedOffset(ctx context.Context, topic string, partitionKey string) (int64, error)
+
+	// UpdateAckedOffset updates the offset_acked for a topic+partition (only if new offset is greater)
+	UpdateAckedOffset(ctx context.Context, topic string, partitionKey string, offset int64) error
+
+	// AckMessage atomically deletes a message and updates the acked offset
+	AckMessage(ctx context.Context, topic string, partitionKey string, messageID string, offset int64, messageStore MessageStore) error
+}
+
+// PartitionLeaseStore handles partition lease operations
+type PartitionLeaseStore interface {
+	// TryAcquireLease attempts to acquire or renew a lease for a partition
+	// Returns true if lease is acquired/owned by this worker
+	TryAcquireLease(ctx context.Context, topic string, partitionKey string) (bool, error)
+
+	// RenewLease renews the lease for a partition owned by this worker
+	RenewLease(ctx context.Context, topic string, partitionKey string) error
+
+	// ReleaseLease releases the lease for a partition owned by this worker
+	ReleaseLease(ctx context.Context, topic string, partitionKey string) error
+
+	// GetLeasedPartitions returns all partitions currently leased by this worker
+	GetLeasedPartitions(ctx context.Context, topic string) ([]string, error)
+
+	// DiscoverAndAcquirePartitions discovers partitions from messages table and tries to acquire leases
+	// Returns the number of new leases acquired
+	DiscoverAndAcquirePartitions(ctx context.Context, topic string) (int, error)
 }