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

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()