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

To implement the SQL-based queue subscriber with partition leasing for distributed consumption and offset tracking for reliable message processing.

- Add MessageStore methods for subscriber: Delete, FetchByOffset, MoveToDLQ, SetVisibilityTimeout
- Add OffsetStore interface for offset tracking and message acknowledgment
- Add PartitionLeaseStore interface for distributed partition leasing
- Implement Subscriber with partition leasing, polling, and retry/DLQ logic
- Add gomock-based tests for Subscriber
- Regenerate gomock mocks for all 3 store interfaces
- Change SetVisibilityTimeout to use int64 (milliseconds) instead of time.Duration
- Add MessageRow struct for representing fetched message rows
- Use external test package pattern (sql_test) to avoid circular dependencies

- Unit tests pass: `bazel test //extensions/queue/sql:sql_test`
- All tests use gomock exclusively, no manual mocks
- Publisher uses defer for RUnlock (cleaner and safer)

Author: behinddwalls

extensions/queue/sql/BUILD.bazel

@@ -6,12 +6,14 @@ 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",
         "@com_github_uber_go_tally_v4//:tally",
         "@org_uber_go_zap//:zap",
     ],
@@ -21,11 +23,36 @@ go_test(
     name = "sql_test",
     srcs = [
         "config_test.go",
-        "publisher_test.go",
     ],
-    data = ["//extensions/queue/sql/schema"],
     embed = [":sql"],
     deps = [
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+    ],
+)
+
+go_test(
+    name = "sql_publisher_test",
+    srcs = ["publisher_test.go"],
+    deps = [
+        ":sql",
+        "//entities/queue",
+        "//extensions/queue",
+        "//extensions/queue/sql/mocks",
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+        "@com_github_uber_go_tally_v4//:tally",
+        "@org_uber_go_mock//gomock",
+        "@org_uber_go_zap//zaptest",
+    ],
+)
+
+go_test(
+    name = "sql_subscriber_test",
+    srcs = ["subscriber_test.go"],
+    data = ["//extensions/queue/sql/schema"],
+    deps = [
+        ":sql",
         "//entities/queue",
         "//extensions/queue",
         "//extensions/queue/sql/mocks",

extensions/queue/sql/mocks/BUILD.bazel

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

extensions/queue/sql/mocks/mock_stores.go

@@ -20,7 +20,7 @@ type publisher struct {
 	closed       bool
 }
 
-// NewPublisher creates a publisher with the given configuration and dependencies
+// NewPublisher creates a publisher for testing with a custom message store
 func NewPublisher(config Config, logger *zap.SugaredLogger, metrics tally.Scope, messageStore MessageStore) *publisher {
 	return &publisher{
 		config:       config,

extensions/queue/sql/publisher.go

@@ -1,17 +1,19 @@
-package sql
+package sql_test
 
 import (
 	"context"
 	"fmt"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally/v4"
 	"go.uber.org/mock/gomock"
 	"go.uber.org/zap/zaptest"
 
 	"github.com/uber/submitqueue/entities/queue"
 	extqueue "github.com/uber/submitqueue/extensions/queue"
+	"github.com/uber/submitqueue/extensions/queue/sql"
 	"github.com/uber/submitqueue/extensions/queue/sql/mocks"
 )
 
@@ -20,9 +22,10 @@ const fixedTimestamp = int64(1234567890000) // Fixed timestamp for test repeatab
 func setupPublisherTest(t *testing.T, mockStore *mocks.MockMessageStore) extqueue.Publisher {
 	t.Helper()
 
-	config := DefaultConfig("test-consumer", "test-worker")
+	config := sql.DefaultConfig("test-consumer", "test-worker")
 
-	return NewPublisher(config,
+	return sql.NewPublisher(
+		config,
 		zaptest.NewLogger(t).Sugar().Named("publisher"),
 		tally.NoopScope.SubScope("publisher"),
 		mockStore,