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

Implements distributed SQL queue subscriber with partition-based load balancing,
offset tracking, and retry/DLQ support.

## Core Features

- **Partition Leasing**: Workers acquire exclusive 30s leases on partitions,
  auto-renewed every 10s. Failed workers release via expiry for automatic rebalancing.
- **Offset Tracking**: Per-partition offsets with optimistic locking. Ack deletes
  message + updates offset atomically. Prevents duplicate processing on restarts.
- **Retry/DLQ**: Persistent retry counter, visibility timeout for crash recovery,
  automatic DLQ move when MaxAttempts exceeded.
- **Event-Driven Architecture**: One goroutine per topic with dual timers
  (poll: 100ms, lease renewal: 10s). Dynamic partition discovery and rebalancing.

## Implementation

**Store Interfaces** (in `extensions/queue/sql/store/`):
- `MessageStore`: FetchByOffset, MoveToDLQ, SetVisibilityTimeout
- `OffsetStore`: Initialize, GetAckedOffset, UpdateAckedOffset, AckMessage
- `PartitionLeaseStore`: DiscoverAndAcquirePartitions, RenewLease, ReleaseLease

**Subscriber Design**:
- `managePartitions()` event loop: poll partitions, renew leases, handle shutdown
- Buffered delivery channel (BatchSize*2) for throughput
- `sqlDelivery` implements Ack/Nack/ExtendVisibilityTimeout
- Standardized metrics with partition-level tags (topic + partition_key)

**Polling Flow**:
1. Discover and acquire new partitions
2. Fetch messages from owned partitions (beyond last acked offset)
3. Check retry limit → DLQ if exceeded
4. Send to delivery channel for consumer processing

## Breaking Changes

- `SetVisibilityTimeout`: Changed from `time.Duration` to `int64` (milliseconds)
- Created `store/` subpackage for interfaces (breaks circular dependency with mocks)
- Added `MessageRow` struct for fetched message representation

## Known Limitations

1. Blocking channel send prevents lease renewal if consumer slow (can cause duplicates)
2. Worker lifetime tied to Subscribe() context (dies if context cancelled)
3. No backpressure metrics or channel depth monitoring
4. Sequential partition processing (one slow partition blocks others)

Follow-up PRs will add timeout-based sends, channel metrics, and concurrent polling.

## Testing

- Unit tests with gomock for all store interfaces
- External test package (`sql_test`) avoids circular dependencies
- Tests: subscription lifecycle, multiple topics, graceful shutdown, close idempotency

## Verification

```bash
bazel test //extensions/queue/sql:sql_test  # PASSED
```

Author: behinddwalls

extensions/queue/sql/BUILD.bazel

@@ -6,12 +6,15 @@ go_library(
         "config.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",
+        "//extensions/queue/sql/store",
         "@com_github_uber_go_tally_v4//:tally",
         "@org_uber_go_zap//:zap",
     ],
@@ -22,8 +25,8 @@ go_test(
     srcs = [
         "config_test.go",
         "publisher_test.go",
+        "subscriber_test.go",
     ],
-    data = ["//extensions/queue/sql/schema"],
     embed = [":sql"],
     deps = [
         "//entities/queue",

extensions/queue/sql/mocks/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//entities/queue",
+        "//extensions/queue/sql/store",
         "@org_uber_go_mock//gomock",
     ],
 )

extensions/queue/sql/mocks/mock_stores.go

@@ -0,0 +1,11 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "store",
+    srcs = ["stores.go"],
+    importpath = "github.com/uber/submitqueue/extensions/queue/sql/store",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//entities/queue",
+    ],
+)

extensions/queue/sql/store/BUILD.bazel

@@ -0,0 +1,78 @@
+package store
+
+//go:generate mockgen -source=stores.go -destination=../mocks/mock_stores.go -package=mocks
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/entities/queue"
+)
+
+// MessageRow represents a row from the messages table
+type MessageRow struct {
+	Offset       int64
+	ID           string
+	Payload      []byte
+	Metadata     map[string]string
+	PartitionKey string
+	RetryCount   int
+	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)
+}