feat(queue/sql): add factory, E2E tests, and documentation

Integrates publisher and subscriber with Queue factory interface.

- Factory creates publisher and subscriber singletons
- Lifecycle management with Close() that cleans up both
- Comprehensive README with usage examples and architecture docs
- End-to-end tests covering publish/subscribe, partitioning, retries, DLQ
- Test coverage for message age tracking and metadata persistence

Test Plan:
- E2E single message publish and consume tested
- Multiple message batch processing verified
- Partition ordering confirmed
- Retry and DLQ flow validated
- Concurrent worker coordination tested
- Message age and metadata persistence verified

Author: behinddwalls

extensions/queue/sql/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "config.go",
         "constants.go",
+        "factory.go",
         "message_store.go",
         "offset_store.go",
         "partition_lease_store.go",
@@ -45,8 +46,36 @@ go_test(
     deps = [
         "//entities/queue",
         "//extensions/queue",
+        "@com_github_dolthub_go_mysql_server//:go-mysql-server",
+        "@com_github_dolthub_go_mysql_server//memory",
+        "@com_github_dolthub_go_mysql_server//server",
+        "@com_github_dolthub_go_mysql_server//sql",
+        "@com_github_dolthub_vitess//go/mysql",
+        "@com_github_go_sql_driver_mysql//:mysql",
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+        "@com_github_uber_go_tally_v4//:tally",
+        "@org_uber_go_zap//:zap",
+    ],
+)
+
+go_test(
+    name = "e2e_test",
+    srcs = ["e2e_test.go"],
+    data = ["//extensions/queue/sql/schema"],
+    embed = [":sql"],
+    deps = [
+        "//entities/queue",
+        "//extensions/queue",
+        "@com_github_dolthub_go_mysql_server//:go-mysql-server",
+        "@com_github_dolthub_go_mysql_server//memory",
+        "@com_github_dolthub_go_mysql_server//server",
+        "@com_github_dolthub_go_mysql_server//sql",
+        "@com_github_dolthub_vitess//go/mysql",
         "@com_github_go_sql_driver_mysql//:mysql",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
+        "@com_github_uber_go_tally_v4//:tally",
+        "@org_uber_go_zap//:zap",
     ],
 )

extensions/queue/sql/README.md

@@ -0,0 +1,146 @@
+# SQL Queue Implementation
+
+MySQL-based distributed queue with partition leasing, visibility timeout, and at-least-once delivery.
+
+## Key Features
+
+- **Single-table design** - All topics in unified tables
+- **Partition leasing** - Workers coordinate via database-native leases with automatic failover
+- **Visibility timeout** - Messages retry automatically if worker crashes or hangs
+- **Persistent retry tracking** - `retry_count` survives crashes, prevents infinite retries
+- **At-least-once delivery** - Crash recovery with offset tracking
+- **Dead letter queue** - Failed messages moved to DLQ after max retries
+
+## Quick Start
+
+```go
+import (
+    "database/sql"
+    _ "github.com/go-sql-driver/mysql"
+    "github.com/uber/submitqueue/extensions/queue/sql"
+    "github.com/uber/submitqueue/entities/queue"
+)
+
+// Setup
+db, _ := sql.Open("mysql", "user:pass@tcp(localhost:3306)/db")
+q, _ := sql.NewFactory(sql.Params{
+    DB:           db,
+    Logger:       logger,
+    MetricsScope: metricsScope,
+    Config:       sql.DefaultConfig("orchestrator", "worker-1"),
+})
+defer q.Close()
+
+// Publish
+msg := queue.NewMessage("msg-id", []byte(`{"data": "value"}`))
+msg.PartitionKey = "repo-123"  // Required
+q.Publisher().Publish(ctx, "merge_events", msg)
+
+// Subscribe
+deliveryCh, _ := q.Subscriber().Subscribe(ctx, "merge_events")
+for delivery := range deliveryCh {
+    if err := process(delivery.Message()); err != nil {
+        delivery.Nack(ctx, 0)  // Retry immediately (0 milliseconds)
+        continue
+    }
+    delivery.Ack(ctx)
+}
+```
+
+## Configuration
+
+```go
+config := sql.DefaultConfig("consumer-group", "worker-id")
+config.PollInterval = 50 * time.Millisecond        // Poll frequency
+config.BatchSize = 20                              // Messages per poll
+config.VisibilityTimeout = 60 * time.Second        // Retry delay
+config.LeaseRenewalInterval = 10 * time.Second     // Lease renewal frequency
+config.LeaseDuration = 30 * time.Second            // Lease validity
+config.Retry.MaxAttempts = 3                       // Max retries before DLQ
+```
+
+## Architecture
+
+### Tables (Single-Table Design)
+
+See full schema: [schema/queue/mysql/schema.sql](../../../schema/queue/mysql/schema.sql)
+
+- **queue_messages** - All messages (topic, partition_key, retry_count, invisible_until)
+- **queue_partition_leases** - Worker coordination (consumer_group, topic, partition_key, leased_by)
+- **queue_offsets** - Consumption progress (consumer_group, topic, partition_key, offset_acked)
+- **queue_dlq** - Failed messages (topic, partition_key, failure_count, last_error)
+
+### How It Works
+
+**Partition Leasing:**
+1. Workers discover partitions from messages table
+2. Workers acquire leases (one worker per partition)
+3. Workers renew leases periodically
+4. Stale leases can be stolen by other workers
+
+**Message Flow:**
+1. Fetch visible messages (invisible_until <= now)
+2. Atomically UPDATE invisible_until and INCREMENT retry_count
+3. Process message
+4. Ack: DELETE message, UPDATE offset_acked
+5. Nack: Do nothing, message becomes visible after timeout
+6. If retry_count >= MaxAttempts: Move to DLQ
+
+**Crash Recovery:**
+- Messages become visible after visibility timeout
+- Other workers steal stale leases
+- Resume from last acked offset
+- Persistent retry_count prevents infinite retries
+
+## Distributed Processing
+
+```go
+// Multiple workers in same consumer group - partition distribution
+q1, _ := sql.NewFactory(sql.Params{
+    Config: sql.DefaultConfig("orchestrator", "worker-1"),
+    // ...
+})
+q2, _ := sql.NewFactory(sql.Params{
+    Config: sql.DefaultConfig("orchestrator", "worker-2"),
+    // ...
+})
+// worker-1 leases partition "repo-1"
+// worker-2 leases partition "repo-2"
+// Automatic failover if worker crashes
+
+// Different consumer group - independent consumption
+q3, _ := sql.NewFactory(sql.Params{
+    Config: sql.DefaultConfig("speculator", "worker-1"),
+    // ...
+})
+```
+
+## Partition Ordering
+
+Messages with same `PartitionKey` are processed in order by single worker:
+
+```go
+msg1 := queue.NewMessage("1", []byte("first"))
+msg1.PartitionKey = "repo-123"
+
+msg2 := queue.NewMessage("2", []byte("second"))
+msg2.PartitionKey = "repo-123"  // Waits for msg1 to be acked
+
+publisher := q.Publisher()
+publisher.Publish(ctx, "events", msg1)
+publisher.Publish(ctx, "events", msg2)
+```
+
+## Metrics (via tally)
+
+**Publisher:** `messages_published`, `publish_errors`, `publish_latency`
+**Subscriber:** `messages_received`, `messages_acked`, `messages_nacked`, `messages_moved_to_dlq`, `message_age`, `poll_latency`, `leases_acquired`
+
+All tagged with `topic`.
+
+## Limitations
+
+- Polling-based (configurable latency via PollInterval)
+- Single MySQL instance throughput
+- Lease renewal overhead
+- Visibility timeout delay before retry