feat(messagequeue): scale tenant reconciliation

## Summary

### Why?

Reconciling every configured tenant on every subscription tick multiplies idle MySQL work and can exhaust the database connection budget when one process serves 100 or more tenants.

### What?

- Prioritize recently active tenants across pipeline topics while sweeping idle tenants round-robin within one lease-duration window.
- Preserve tenant-scoped Vitess queries and per-tenant orphan-sweep progress while sharing activity across subscriptions.
- Bound each service's shared queue database pool with required QUEUE_MYSQL_MAX_OPEN_CONNECTIONS configuration and Compose defaults of 16.
- Keep the gateway log-consumer integration on its configured e2e-test-queue tenant.
- Add deterministic scheduler coverage for active priority, 100-tenant sweep completion, state preservation, failure isolation, and cancellation.

## Test Plan

✅ `./tool/bazel test //platform/extension/messagequeue/mysql:go_default_test //service/messagequeue:go_default_test --test_output=errors`
✅ `./tool/bazel test //test/integration/submitqueue/gateway:go_default_test --test_output=errors --strategy=TestRunner=local`
✅ `./tool/bazel test //test/e2e/submitqueue:go_default_test --test_output=errors --strategy=TestRunner=local`
✅ `./tool/bazel test //test/integration/extension/messagequeue/mysql/vitess:go_default_test --test_output=errors --strategy=TestRunner=local`
✅ `make fmt`

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

Author: 2 people

platform/extension/messagequeue/mysql/subscriber.go

@@ -113,6 +113,10 @@ type subscriber struct {
 	mu                 sync.RWMutex
 	closed             bool
 
+	// Activity is shared across topics so a tenant stays hot through pipeline handoffs.
+	tenantActivityMu  sync.Mutex
+	tenantActiveUntil map[string]time.Time
+
 	// Active subscriptions
 	subscriptions map[string]*subscription
 	subMu         sync.Mutex
@@ -160,6 +164,11 @@ type subscription struct {
 	// partition keys don't hold leases, offset rows, and polling workers forever.
 	// Only accessed by the single managePartitions goroutine — no locking needed.
 	drainedSince map[entityqueue.PartitionIdentity]time.Time
+
+	// Idle tenant sweeps advance independently for discovery and lease maintenance.
+	discoveryTenantCursor int
+	leaseTenantCursor     int
+	lastOrphanSweep       map[string]time.Time
 }
 
 func partitionKeysForTenant(partitions []entityqueue.PartitionIdentity, tenant string) []string {
@@ -181,6 +190,74 @@ func sortPartitionIdentities(partitions []entityqueue.PartitionIdentity) {
 	})
 }
 
+func selectTenantReconciliationBatch(
+	tenants []string,
+	activeTenants map[string]struct{},
+	cursor int,
+	tickIntervalMs int64,
+	sweepIntervalMs int64,
+) ([]string, int) {
+	selected := make([]string, 0, len(tenants))
+	idleCount := 0
+	for _, tenant := range tenants {
+		if _, active := activeTenants[tenant]; active {
+			selected = append(selected, tenant)
+			continue
+		}
+		idleCount++
+	}
+	if idleCount == 0 {
+		return selected, 0
+	}
+
+	ticksPerSweep := max(int((sweepIntervalMs+tickIntervalMs-1)/tickIntervalMs), 1)
+	idlePerTick := max((idleCount+ticksPerSweep-1)/ticksPerSweep, 1)
+	cursor %= len(tenants)
+	nextCursor := cursor
+	added := 0
+	for scanned := 0; scanned < len(tenants) && added < idlePerTick; scanned++ {
+		index := (cursor + scanned) % len(tenants)
+		nextCursor = (index + 1) % len(tenants)
+		if _, active := activeTenants[tenants[index]]; active {
+			continue
+		}
+		selected = append(selected, tenants[index])
+		added++
+	}
+	return selected, nextCursor
+}
+
+func (s *subscriber) activeTenantsForSubscription(sub *subscription, now time.Time) map[string]struct{} {
+	active := make(map[string]struct{})
+	s.tenantActivityMu.Lock()
+	for tenant, until := range s.tenantActiveUntil {
+		if until.After(now) {
+			active[tenant] = struct{}{}
+			continue
+		}
+		delete(s.tenantActiveUntil, tenant)
+	}
+	s.tenantActivityMu.Unlock()
+
+	sub.workersMu.Lock()
+	defer sub.workersMu.Unlock()
+	for _, partition := range sub.lastDiscoveredPartitions {
+		active[partition.Tenant] = struct{}{}
+	}
+	for partition := range sub.workers {
+		active[partition.Tenant] = struct{}{}
+	}
+	return active
+}
+
+func (s *subscriber) markTenantActive(tenant string, until time.Time) {
+	s.tenantActivityMu.Lock()
+	defer s.tenantActivityMu.Unlock()
+	if current := s.tenantActiveUntil[tenant]; current.Before(until) {
+		s.tenantActiveUntil[tenant] = until
+	}
+}
+
 type tenantOperationResult[T any] struct {
 	tenant string
 	value  T
@@ -501,6 +578,7 @@ func NewSubscriber(logger *zap.SugaredLogger, scope tally.Scope, messageStore me
 		deliveryStateStore: deliveryStateStore,
 		tenants:            tenants,
 		shutdownTimeout:    subscriptionShutdownTimeout,
+		tenantActiveUntil:  make(map[string]time.Time, len(tenants)),
 		subscriptions:      make(map[string]*subscription),
 	}
 }
@@ -613,12 +691,13 @@ func (s *subscriber) Subscribe(ctx context.Context, topic string, config extqueu
 	// effective for a caller whose ctx never completes.
 	subCtx, cancel := context.WithCancel(ctx)
 	sub := &subscription{
-		topic:      topic,
-		config:     config,
-		deliveryCh: make(chan extqueue.Delivery, config.BatchSize*2),
-		cancelFunc: cancel,
-		done:       make(chan struct{}),
-		workers:    make(map[entityqueue.PartitionIdentity]*partitionWorker),
+		topic:           topic,
+		config:          config,
+		deliveryCh:      make(chan extqueue.Delivery, config.BatchSize*2),
+		cancelFunc:      cancel,
+		done:            make(chan struct{}),
+		workers:         make(map[entityqueue.PartitionIdentity]*partitionWorker),
+		lastOrphanSweep: make(map[string]time.Time, len(s.tenants)),
 	}
 
 	s.subscriptions[subKey] = sub
@@ -668,26 +747,36 @@ func (s *subscriber) managePartitions(ctx context.Context, sub *subscription) {
 		"subscriber_name", cfg.SubscriberName,
 	}
 
-	discoveryTicker := time.NewTicker(time.Duration(cfg.PartitionDiscoveryIntervalMs) * time.Millisecond)
+	discoveryInterval := time.Duration(cfg.PartitionDiscoveryIntervalMs) * time.Millisecond
+	discoveryTicker := time.NewTicker(discoveryInterval)
 	defer discoveryTicker.Stop()
 
-	leaseTicker := time.NewTicker(time.Duration(cfg.LeaseRenewalIntervalMs) * time.Millisecond)
+	leaseInterval := time.Duration(cfg.LeaseRenewalIntervalMs) * time.Millisecond
+	leaseTicker := time.NewTicker(leaseInterval)
 	defer leaseTicker.Stop()
+	tenantSweepIntervalMs := cfg.LeaseDurationMs
 
 	// Orphan sweep pacing: an uncapped acquisition pass runs every two lease
 	// durations. Two lease durations is past every transient window in the
 	// protocol — an expiring lease, a crashed peer's heartbeat going stale —
 	// so anything still unleased at sweep time is genuinely unclaimed.
 	orphanSweepInterval := 2 * time.Duration(cfg.LeaseDurationMs) * time.Millisecond
-	lastOrphanSweep := time.Now()
-
-	// Send initial heartbeat so this subscriber is immediately visible to
-	// ActiveSubscribers. Without this, other subscribers compute incorrect
-	// fair shares until the first leaseTicker fires.
-	// Initial heartbeat failure is non-fatal — the next leaseTicker fires within
-	// LeaseRenewalIntervalMs and retries.
-	tenantLeaseTimeout := time.Duration(cfg.LeaseRenewalIntervalMs) * time.Millisecond
-	for _, result := range runTenantOperations(ctx, s.tenants, tenantLeaseTimeout, func(tenantCtx context.Context, tenant string) (struct{}, error) {
+	for _, tenant := range s.tenants {
+		sub.lastOrphanSweep[tenant] = time.Now()
+	}
+
+	// Idle tenants are spread across one lease-duration window; active tenants
+	// join every pass so their heartbeats and leases retain the normal cadence.
+	tenantLeaseTimeout := leaseInterval
+	initialTenants, nextLeaseCursor := selectTenantReconciliationBatch(
+		s.tenants,
+		s.activeTenantsForSubscription(sub, time.Now()),
+		sub.leaseTenantCursor,
+		cfg.LeaseRenewalIntervalMs,
+		tenantSweepIntervalMs,
+	)
+	sub.leaseTenantCursor = nextLeaseCursor
+	for _, result := range runTenantOperations(ctx, initialTenants, tenantLeaseTimeout, func(tenantCtx context.Context, tenant string) (struct{}, error) {
 		return struct{}{}, s.sendHeartbeat(tenantCtx, sub, tenant)
 	}) {
 		if result.err != nil {
@@ -719,7 +808,15 @@ func (s *subscriber) managePartitions(ctx context.Context, sub *subscription) {
 			return
 
 		case <-leaseTicker.C:
-			runTenantOperations(ctx, s.tenants, tenantLeaseTimeout, func(tenantCtx context.Context, tenant string) (struct{}, error) {
+			leaseTenants, nextCursor := selectTenantReconciliationBatch(
+				s.tenants,
+				s.activeTenantsForSubscription(sub, time.Now()),
+				sub.leaseTenantCursor,
+				cfg.LeaseRenewalIntervalMs,
+				tenantSweepIntervalMs,
+			)
+			sub.leaseTenantCursor = nextCursor
+			runTenantOperations(ctx, leaseTenants, tenantLeaseTimeout, func(tenantCtx context.Context, tenant string) (struct{}, error) {
 				tenantFields := append(logFields, "tenant", tenant)
 				// Fetch leased partitions once for this tenant tick — shared by
 				// rebalance and renewLeases to avoid redundant queries.
@@ -786,11 +883,23 @@ func (s *subscriber) managePartitions(ctx context.Context, sub *subscription) {
 			// subscriber that heartbeats without acquiring) is picked up by
 			// whichever subscriber sweeps first. An over-cap grab is shed at
 			// the next rebalance once a peer has spare capacity to take it.
-			uncapped := time.Since(lastOrphanSweep) >= orphanSweepInterval
-			if uncapped {
-				lastOrphanSweep = time.Now()
+			discoveryTenants, nextCursor := selectTenantReconciliationBatch(
+				s.tenants,
+				s.activeTenantsForSubscription(sub, time.Now()),
+				sub.discoveryTenantCursor,
+				cfg.PartitionDiscoveryIntervalMs,
+				tenantSweepIntervalMs,
+			)
+			sub.discoveryTenantCursor = nextCursor
+			uncappedTenants := make(map[string]struct{})
+			now := time.Now()
+			for _, tenant := range discoveryTenants {
+				if now.Sub(sub.lastOrphanSweep[tenant]) >= orphanSweepInterval {
+					uncappedTenants[tenant] = struct{}{}
+					sub.lastOrphanSweep[tenant] = now
+				}
 			}
-			if err := s.discoverAndReconcileWorkers(ctx, sub, uncapped); err != nil {
+			if err := s.discoverAndReconcileWorkers(ctx, sub, discoveryTenants, uncappedTenants); err != nil {
 				s.logger.Errorw("partition discovery failed, will retry on next tick", append(logFields, "error", err)...)
 			}
 			s.emitSignal(SignalPartitionUpdate)
@@ -800,20 +909,33 @@ func (s *subscriber) managePartitions(ctx context.Context, sub *subscription) {
 
 // discoverAndReconcileWorkers discovers new partitions and reconciles workers.
 // Uses fair share to limit how many partitions this subscriber acquires;
-// uncapped skips the fair-share cap entirely (the orphan sweep).
-func (s *subscriber) discoverAndReconcileWorkers(ctx context.Context, sub *subscription, uncapped bool) error {
-	if len(s.tenants) == 0 {
+// tenants in uncappedTenants skip the fair-share cap (the orphan sweep).
+func (s *subscriber) discoverAndReconcileWorkers(
+	ctx context.Context,
+	sub *subscription,
+	tenants []string,
+	uncappedTenants map[string]struct{},
+) error {
+	if len(tenants) == 0 {
 		return nil
 	}
 
 	cfg := sub.config
 
 	sub.workersMu.Lock()
 	cachedDiscovered := append([]entityqueue.PartitionIdentity(nil), sub.lastDiscoveredPartitions...)
+	cachedLeased := make([]entityqueue.PartitionIdentity, 0, len(sub.workers))
+	for partition := range sub.workers {
+		cachedLeased = append(cachedLeased, partition)
+	}
 	sub.workersMu.Unlock()
 
-	discoveredByTenant := make(map[string][]string, len(s.tenants))
-	leasedByTenant := make(map[string][]string, len(s.tenants))
+	attemptedTenants := make(map[string]struct{}, len(tenants))
+	for _, tenant := range tenants {
+		attemptedTenants[tenant] = struct{}{}
+	}
+	discoveredByTenant := make(map[string][]string, len(tenants))
+	leasedByTenant := make(map[string][]string, len(tenants))
 	var discoveryErrs []error
 
 	type tenantDiscovery struct {
@@ -824,7 +946,7 @@ func (s *subscriber) discoverAndReconcileWorkers(ctx context.Context, sub *subsc
 		time.Duration(cfg.PartitionDiscoveryIntervalMs)*time.Millisecond,
 		time.Duration(cfg.LeaseRenewalIntervalMs)*time.Millisecond,
 	)
-	results := runTenantOperations(ctx, s.tenants, discoveryTimeout, func(tenantCtx context.Context, tenant string) (tenantDiscovery, error) {
+	results := runTenantOperations(ctx, tenants, discoveryTimeout, func(tenantCtx context.Context, tenant string) (tenantDiscovery, error) {
 		leasedPartitions, err := s.leaseStore.GetLeasedPartitions(tenantCtx, tenant, sub.topic, cfg.SubscriberName, cfg.ConsumerGroup)
 		if err != nil {
 			return tenantDiscovery{}, fmt.Errorf("get leased partitions tenant=%s: %w", tenant, err)
@@ -833,7 +955,7 @@ func (s *subscriber) discoverAndReconcileWorkers(ctx context.Context, sub *subsc
 		cachedForTenant := partitionKeysForTenant(cachedDiscovered, tenant)
 
 		maxPartitions := 0
-		if !uncapped {
+		if _, uncapped := uncappedTenants[tenant]; !uncapped {
 			maxPartitions, err = s.fairShareCap(tenantCtx, sub, tenant, leasedPartitions, cachedForTenant)
 			if err != nil {
 				return tenantDiscovery{}, fmt.Errorf("compute fair share cap tenant=%s: %w", tenant, err)
@@ -849,14 +971,28 @@ func (s *subscriber) discoverAndReconcileWorkers(ctx context.Context, sub *subsc
 		if err != nil {
 			return tenantDiscovery{}, fmt.Errorf("get leased partitions after acquire tenant=%s: %w", tenant, err)
 		}
+		if len(discoveredPartitions) > 0 || len(leasedPartitions) > 0 {
+			if err := s.sendHeartbeat(tenantCtx, sub, tenant); err != nil {
+				s.logger.Errorw("heartbeat on tenant activation failed",
+					"tenant", tenant,
+					"topic", sub.topic,
+					"consumer_group", cfg.ConsumerGroup,
+					"error", err,
+				)
+			}
+		}
 		return tenantDiscovery{discovered: discoveredPartitions, leased: leasedPartitions}, nil
 	})
 
+	activeUntil := time.Now().Add(time.Duration(cfg.LeaseDurationMs) * time.Millisecond)
 	for _, result := range results {
 		if result.err != nil {
 			discoveryErrs = append(discoveryErrs, result.err)
 			continue
 		}
+		if len(result.value.discovered) > 0 || len(result.value.leased) > 0 {
+			s.markTenantActive(result.tenant, activeUntil)
+		}
 		discoveredByTenant[result.tenant] = result.value.discovered
 		leasedByTenant[result.tenant] = result.value.leased
 	}
@@ -869,6 +1005,23 @@ func (s *subscriber) discoverAndReconcileWorkers(ctx context.Context, sub *subsc
 	var expired []entityqueue.PartitionIdentity
 
 	for _, tenant := range s.tenants {
+		if _, attempted := attemptedTenants[tenant]; !attempted {
+			for _, pk := range partitionKeysForTenant(cachedDiscovered, tenant) {
+				allDiscovered = append(allDiscovered, entityqueue.PartitionIdentity{Tenant: tenant, PartitionKey: pk})
+			}
+			for _, partition := range cachedLeased {
+				if partition.Tenant == tenant {
+					allLeased = append(allLeased, partition)
+				}
+			}
+			for partition, since := range sub.drainedSince {
+				if partition.Tenant == tenant {
+					nextDrainedSince[partition] = since
+				}
+			}
+			continue
+		}
+
 		discoveredPartitions, succeeded := discoveredByTenant[tenant]
 		if !succeeded {
 			// Unconfirmed leases must not keep workers polling: a peer can acquire after expiry.