feat(queue/sql): add factory and documentation

Add Factory implementation (sql.go) for SQL queue that provides Publisher
and Subscriber instances with proper lifecycle management.

Renamed NewFactory to NewQueue for clarity.

Add comprehensive unit tests (sql_test.go) for factory covering:
- Success cases with all params and nil metrics scope
- Error cases: nil DB, nil Logger, invalid config, ping failure
- Publisher and Subscriber singleton behavior
- Close lifecycle and idempotency (multiple errors combined)
- Full integration test

Tests use table-driven approach for consistency.

Add concise README documenting SQL queue implementation, usage patterns,
and configuration options. Removed verbose metric details and examples.

Fixed critical issues:
- Close() now releases lock before calling subscriber/publisher Close()
  to prevent deadlocks
- Close() combines all errors instead of only returning first error

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: behinddwalls

extensions/queue/sql/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "offset_store.go",
         "partition_lease_store.go",
         "publisher.go",
+        "sql.go",
         "stores.go",
         "subscriber.go",
         "validation.go",
@@ -34,6 +35,7 @@ go_test(
         "offset_store_test.go",
         "partition_lease_store_test.go",
         "publisher_test.go",
+        "sql_test.go",
         "subscriber_test.go",
     ],
     embed = [":sql"],

extensions/queue/sql/README.md

@@ -0,0 +1,82 @@
+# SQL Queue Implementation
+
+MySQL-based distributed queue with partition leasing, visibility timeout, and at-least-once delivery.
+
+## Key Features
+
+- **Partition leasing** - Workers coordinate via database leases with automatic failover
+- **Visibility timeout** - Messages retry automatically if worker crashes
+- **At-least-once delivery** - Offset tracking for crash recovery
+- **Dead letter queue** - Failed messages moved to DLQ after max retries
+
+## Quick Start
+
+```go
+import (
+    "database/sql"
+    _ "github.com/go-sql-driver/mysql"
+    queueSQL "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, _ := queueSQL.NewQueue(queueSQL.Params{
+    DB:     db,
+    Logger: logger,
+    Config: queueSQL.DefaultConfig("orchestrator", "worker-1"),
+})
+defer q.Close()
+
+// Publish
+msg := queue.NewMessage("msg-id", []byte(`{"data": "value"}`))
+msg.PartitionKey = "repo-123"  // Required for ordering
+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
+        continue
+    }
+    delivery.Ack(ctx)
+}
+```
+
+## Configuration
+
+```go
+config := queueSQL.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.Retry.MaxAttempts = 3                       // Max retries before DLQ
+```
+
+## How It Works
+
+**Partition Leasing:**
+1. Workers discover partitions from messages table
+2. Workers acquire leases (one worker per partition)
+3. Stale leases can be stolen by other workers
+
+**Message Flow:**
+1. Fetch visible messages (invisible_until <= now)
+2. Process message
+3. Ack: DELETE message, UPDATE offset
+4. Nack: Message becomes visible after timeout
+5. 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
+
+## Partition Ordering
+
+Messages with same `PartitionKey` are processed in order by a single worker.
+
+## Distributed Processing
+
+Multiple workers in the same consumer group share partitions. Workers in different consumer groups consume independently.

extensions/queue/sql/config.go

@@ -6,7 +6,7 @@ import (
 )
 
 // Config holds configuration for SQL-based queue.
-// DB connection, logger, and metrics are passed separately to NewFactory.
+// DB connection, logger, and metrics are passed separately to NewQueue.
 type Config struct {
 	// ConsumerGroup identifies this consumer for offset tracking (required)
 	ConsumerGroup string

extensions/queue/sql/sql.go

@@ -0,0 +1,171 @@
+package sql
+
+import (
+	"database/sql"
+	"fmt"
+	"sync"
+
+	"github.com/uber-go/tally/v4"
+	"go.uber.org/zap"
+
+	"github.com/uber/submitqueue/extensions/queue"
+)
+
+type factory struct {
+	db          *sql.DB
+	config      Config
+	logger      *zap.SugaredLogger
+	metrics     tally.Scope
+	messageStore messageStore
+	offsetStore  offsetStore
+	leaseStore   partitionLeaseStore
+	publisher   queue.Publisher
+	subscriber  queue.Subscriber
+	mu          sync.RWMutex
+	closed      bool
+}
+
+// Params holds dependencies for creating a SQL queue factory
+type Params struct {
+	// DB is the database connection (required)
+	DB *sql.DB
+
+	// Logger for debugging and observability (required)
+	Logger *zap.Logger
+
+	// MetricsScope for metrics collection (optional, uses noop if nil)
+	MetricsScope tally.Scope
+
+	// Config holds queue configuration
+	Config Config
+}
+
+// NewQueue creates a new SQL-based queue factory
+func NewQueue(params Params) (queue.Queue, error) {
+	if params.DB == nil {
+		return nil, fmt.Errorf("DB is required")
+	}
+	if params.Logger == nil {
+		return nil, fmt.Errorf("Logger is required")
+	}
+
+	if err := params.Config.Validate(); err != nil {
+		return nil, fmt.Errorf("invalid config: %w", err)
+	}
+
+	// Use noop scope if not provided
+	metricsScope := params.MetricsScope
+	if metricsScope == nil {
+		metricsScope = tally.NoopScope
+	}
+
+	// Test connection
+	if err := params.DB.Ping(); err != nil {
+		return nil, fmt.Errorf("failed to ping database: %w", err)
+	}
+
+	logger := params.Logger.Sugar().Named("queue.sql")
+	logger.Infow("created SQL queue factory",
+		"consumer_group", params.Config.ConsumerGroup,
+		"worker_id", params.Config.WorkerID,
+		"poll_interval", params.Config.PollInterval,
+		"batch_size", params.Config.BatchSize,
+	)
+
+	// Create stores
+	messageStore := newMessageStore(params.DB, params.Config, params.Logger, metricsScope)
+	offsetStore := newOffsetStore(params.DB, params.Config, params.Logger, metricsScope)
+	leaseStore := newPartitionLeaseStore(params.DB, params.Config, params.Logger, metricsScope)
+
+	f := &factory{
+		db:          params.DB,
+		config:      params.Config,
+		logger:      logger,
+		metrics:     metricsScope.SubScope("queue"),
+		messageStore: messageStore,
+		offsetStore:  offsetStore,
+		leaseStore:   leaseStore,
+	}
+
+	return f, nil
+}
+
+// Publisher returns a Publisher instance (singleton)
+func (f *factory) Publisher() queue.Publisher {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	if f.publisher == nil {
+		f.publisher = &publisher{
+			config:      f.config,
+			logger:      f.logger.Named("publisher"),
+			metrics:     f.metrics.SubScope("publisher"),
+			messageStore: f.messageStore,
+		}
+	}
+
+	return f.publisher
+}
+
+// Subscriber returns a Subscriber instance (singleton)
+func (f *factory) Subscriber() queue.Subscriber {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	if f.subscriber == nil {
+		f.subscriber = NewSubscriber(
+			f.config,
+			f.logger.Named("subscriber"),
+			f.metrics.SubScope("subscriber"),
+			f.messageStore,
+			f.offsetStore,
+			f.leaseStore,
+		)
+	}
+
+	return f.subscriber
+}
+
+// Close shuts down the factory and all associated resources
+func (f *factory) Close() error {
+	// Mark as closed and grab references under lock
+	f.mu.Lock()
+	if f.closed {
+		f.mu.Unlock()
+		return nil
+	}
+	f.closed = true
+	sub := f.subscriber
+	pub := f.publisher
+	f.mu.Unlock()
+
+	// Close subscriber and publisher outside the lock to avoid deadlocks
+	var errs []error
+
+	if sub != nil {
+		if err := sub.Close(); err != nil {
+			errs = append(errs, fmt.Errorf("subscriber close failed: %w", err))
+		}
+	}
+
+	if pub != nil {
+		if err := pub.Close(); err != nil {
+			errs = append(errs, fmt.Errorf("publisher close failed: %w", err))
+		}
+	}
+
+	// Return all errors if any occurred
+	if len(errs) == 0 {
+		return nil
+	}
+
+	// Combine all errors into a single descriptive message
+	errMsg := "multiple close errors: "
+	for i, err := range errs {
+		if i > 0 {
+			errMsg += "; "
+		}
+		errMsg += err.Error()
+	}
+	return fmt.Errorf(errMsg)
+}