feat(messagequeue): supervisor ticks use tenant IN-list (#702)

## Summary

### Why?

Each subscription tick fanned out one goroutine and one `WHERE tenant =
?` round-trip per configured tenant. Tenants that hash onto the same
Vitess shard still paid N Go timeouts and N queries. Partial-read
failures in that new shared path also dropped poll workers or skipped
lease renew, so one listing error could stall every tenant until the
next successful tick.

### What?

Replace supervisor reads and same-shape writes with `WHERE tenant IN
(MQ_TENANTS)`. Go still groups rows by tenant and applies fair-share per
`(tenant, topic)`. `TryAcquireLease` and per-partition `ReleaseLease`
stay row-local. Poll workers and publish SQL are unchanged.

If discovery or the leased-list read fails, cached discovery is kept and
unconfirmed workers stop. If GetAllLeases or ActiveSubscribers fails
after leases are known, skip acquire and reconcile from this tick's
leased list. If the lease tick cannot list owned partitions or active
peers, skip rebalance and still renew, heartbeat, and purge.

Remove the superseded single-tenant supervisor store methods and their
test-only callers so the private interfaces expose only the production
paths.

## Test Plan

- ✅ `make check-mocks && make check-gazelle && make check-tidy`
- ✅ `./tool/bazel test
//platform/extension/messagequeue/mysql:go_default_test
--test_output=errors`
- ✅ `./tool/bazel test
//test/integration/extension/messagequeue/mysql:go_default_test
--test_output=errors`
- ✅ `./tool/bazel test
//test/integration/extension/messagequeue/mysql/vitess:go_default_test
--test_output=errors --strategy=TestRunner=local`

Co-authored-by: Cursor <cursoragent@cursor.com>

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: behinddwalls

doc/rfc/messagequeue-tenant-sharding.md

@@ -69,12 +69,12 @@ Today partition discovery runs `SELECT DISTINCT partition_key FROM queue_message
 The subscriber takes an explicit configured tenant list from `MQ_TENANTS`. Consumer processes reject an empty list at startup; Stovepipe also rejects ingest requests for names outside the list. Discovery becomes:
 
 ```sql
-SELECT DISTINCT partition_key FROM queue_messages
-WHERE tenant = ? AND topic = ?
-ORDER BY partition_key
+SELECT DISTINCT tenant, partition_key FROM queue_messages
+WHERE tenant IN (MQ_TENANTS) AND topic = ?
+ORDER BY tenant, partition_key
 ```
 
-Fair-share, orphan sweep, and idle-lease release run per `(tenant, topic)`, not across all tenants on a topic. Discovery and shutdown attempt every configured tenant and aggregate errors so one unavailable shard does not block unrelated tenants.
+vtgate scatters only to shards that own those vindex values. Fair-share, orphan sweep, and idle-lease release still run per `(tenant, topic)` after grouping the result set in Go. One unavailable serving shard fails the tick for every listed tenant; the next interval retries. Poll workers stay scoped to leased `(tenant, partition_key)` rows. Discovery never uses an unscoped `WHERE topic = ?` predicate on Vitess.
 
 ## Publish
 

platform/extension/messagequeue/mysql/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "delivery_state_store.go",
         "errors.go",
         "identifier.go",
+        "inlist.go",
         "message_store.go",
         "mock_stores.go",
         "offset_store.go",
@@ -35,6 +36,7 @@ go_test(
     name = "go_default_test",
     srcs = [
         "delivery_state_store_test.go",
+        "inlist_test.go",
         "message_store_test.go",
         "offset_store_test.go",
         "partition_lease_store_test.go",

platform/extension/messagequeue/mysql/delivery_state_store.go

@@ -246,22 +246,21 @@ func (s *sqldeliveryStateStore) AdvanceWatermark(ctx context.Context, consumerGr
 	}
 
 	// Batch-fetch delivery state for the provided offsets.
-	placeholders := make([]byte, 0, len(offsets)*2-1)
+	placeholders, ok := inListPlaceholders(len(offsets))
+	if !ok {
+		return currentWatermark, nil
+	}
 	args := make([]interface{}, 0, 4+len(offsets))
 	args = append(args, tenant, consumerGroup, topic, partitionKey)
-	for i, offset := range offsets {
-		if i > 0 {
-			placeholders = append(placeholders, ',')
-		}
-		placeholders = append(placeholders, '?')
+	for _, offset := range offsets {
 		args = append(args, offset)
 	}
 
 	rows, err := s.db.QueryContext(ctx, fmt.Sprintf(`
 		SELECT message_offset, acked FROM %s
 		WHERE tenant = ? AND consumer_group = ? AND topic = ? AND partition_key = ?
 		AND message_offset IN (%s)
-	`, DeliveryStateTableName, string(placeholders)), args...)
+	`, DeliveryStateTableName, placeholders), args...)
 	if err != nil {
 		return currentWatermark, fmt.Errorf("query delivery state for watermark tenant=%s topic=%s partition=%s: %w", tenant, topic, partitionKey, err)
 	}

platform/extension/messagequeue/mysql/inlist.go

@@ -0,0 +1,32 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mysql
+
+import "strings"
+
+// inListPlaceholders returns "?,?,…" of length n. n < 1 is a no-op query.
+func inListPlaceholders(n int) (string, bool) {
+	if n < 1 {
+		return "", false
+	}
+	return strings.Repeat(",?", n)[1:], true
+}
+
+func appendStrings(args []any, values []string) []any {
+	for _, value := range values {
+		args = append(args, value)
+	}
+	return args
+}

platform/extension/messagequeue/mysql/inlist_test.go

@@ -0,0 +1,39 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mysql
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestInListPlaceholders(t *testing.T) {
+	tests := []struct {
+		n      int
+		want   string
+		wantOK bool
+	}{
+		{n: 0},
+		{n: -1},
+		{n: 1, want: "?", wantOK: true},
+		{n: 3, want: "?,?,?", wantOK: true},
+	}
+	for _, tt := range tests {
+		got, ok := inListPlaceholders(tt.n)
+		assert.Equal(t, tt.wantOK, ok)
+		assert.Equal(t, tt.want, got)
+	}
+}

platform/extension/messagequeue/mysql/message_store.go

@@ -144,22 +144,6 @@ func (s *sqlmessageStore) Insert(ctx context.Context, tenant string, topic strin
 	return nil
 }
 
-// Delete deletes a message by tenant, topic, partition key, and ID
-func (s *sqlmessageStore) Delete(ctx context.Context, tenant string, topic string, partitionKey string, messageID string) (retErr error) {
-	op := metrics.Begin(s.scope, "delete", metrics.StorageLatencyBuckets, metrics.NewTag("topic", topic))
-	defer func() { op.Complete(retErr) }()
-
-	_, err := s.db.ExecContext(ctx, fmt.Sprintf(`
-		DELETE FROM %s WHERE tenant = ? AND topic = ? AND partition_key = ? AND id = ?
-	`, MessagesTableName), tenant, topic, partitionKey, messageID)
-
-	if err != nil {
-		return fmt.Errorf("delete message tenant=%s topic=%s partition=%s message=%s: %w", tenant, topic, partitionKey, messageID, err)
-	}
-
-	return nil
-}
-
 // FetchByOffset fetches messages with offset > currentOffset for a specific partition.
 // Messages are fetched from the immutable log; no per-message mutation occurs.
 func (s *sqlmessageStore) FetchByOffset(ctx context.Context, tenant string, topic string, partitionKey string, currentOffset int64, limit int) (_ []messageRow, retErr error) {

platform/extension/messagequeue/mysql/message_store_test.go

@@ -131,24 +131,6 @@ func TestMessageStore_Insert(t *testing.T) {
 	}
 }
 
-func TestMessageStore_Delete(t *testing.T) {
-	db, mock, store := setupmessageStoreTest(t)
-	defer db.Close()
-
-	ctx := context.Background()
-	topic := "test_topic"
-	partitionKey := "part1"
-	messageID := "msg1"
-
-	mock.ExpectExec("DELETE FROM queue_messages").
-		WithArgs(testTenant, topic, partitionKey, messageID).
-		WillReturnResult(sqlmock.NewResult(0, 1))
-
-	err := store.Delete(ctx, testTenant, topic, partitionKey, messageID)
-	require.NoError(t, err)
-	require.NoError(t, mock.ExpectationsWereMet())
-}
-
 func TestMessageStore_FetchByOffset(t *testing.T) {
 	db, mock, store := setupmessageStoreTest(t)
 	defer db.Close()