gateway: faster mass reconnect (Phases 1 + 3)

[bensonc]

Summary

Implements Phases 1 and 3 of the faster mass-reconnect design. Phase 2 (drain-based stabilize signal) is deferred — the divergence + sweep work in Phase 3 attacks the same bottleneck more directly.

Mass identify-storm reconnect of ~1024 shards goes from ~75 min today to Discord's identify-throughput floor (~5–10 min) in steady state, once the DB has been populated and most guilds skip backfill via the divergence check. Until that steady state is reached, the structural changes from Phase 1 still let ops dial up identify-stabilize concurrency to land in the 8–16 min range with default config.

Phase 1 — lock-shrink + decoupled stabilize gate

• WS dial + HELLO moved before identify mutex acquire.
• Identify mutex held only for IDENTIFY_PACING_SECONDS (default 10s) + 5s grace, regardless of READY arrival. Slow READYs no longer wedge the bucket.
• 60s stabilize wait moved to a per-process weighted semaphore. Capacity = IDENTIFY_STABILIZE_CONCURRENCY (default 1). Hold = IDENTIFY_STABILIZE_SECONDS (default 60). Skipped when shouldProcessMembers() is false.
• Etcd lease TTL now reflects the new (~25s) hold time. Crashed shards free the bucket faster.

Phase 3a — divergence check at GUILD_CREATE

• handler.GuildCreate reads cached member count once via existing GetGuildMemberCount and returns it alongside Discord's member_count in EventPayload.
• gatewayws.shouldRequestMembers skips Request Guild Members when |discord - cached| / discord <= BACKFILL_DIVERGENCE_RATIO (default 0.05). Cold guilds (cached == 0) always request.
• Steady-state mass reconnect on a populated DB skips most member backfill → no chunk burst → stabilize semaphore drains immediately.

Phase 3b — bounded-rate cursor sweep

• New state.DB.GetGuildIDsAfter(after, limit) paged guild enumeration. Postgres uses the existing PK index; FoundationDB uses a SelectorRange from FirstGreaterThan(guild_key(after)). Never loads more than limit ids into memory.
• internal/manager/sweep.go runs one goroutine per process. Pages guilds, dispatches Request Guild Members to the owning shard for ids within this process's shard range, throttled by golang.org/x/time/rate.
• Cursor persisted to etcd at /gateway/sweep_cursor/<name> — survives restarts, no schema migration.
• Wraps to 0 when end of guild data reached, with a brief sleep to avoid spinning on an empty DB.

New env vars (all optional, defaults preserve current behavior)

Var	Default	Purpose
IDENTIFY_PACING_SECONDS	10	Identify mutex hold time after IDENTIFY
IDENTIFY_STABILIZE_CONCURRENCY	1	Concurrent shards in backfill window
IDENTIFY_STABILIZE_SECONDS	60	Per-shard stabilize hold
IDENTIFY_STABILIZE_SECONDS_MAX	2x duration	Stabilize safety bound
BACKFILL_DIVERGENCE_RATIO	0.05	Skip request when divergence <= this
BACKFILL_SWEEP_ENABLED	true	Set false to disable the background sweep
BACKFILL_SWEEP_REQUESTS_PER_SECOND	12	Global sweep dispatch rate
BACKFILL_SWEEP_BATCH	200	Max ids loaded per sweep query

Out of scope

• Phase 2 (drain-based stabilize signal). Skipped because Phase 3 makes the fixed 60s wait the right knob to keep — most shards trigger zero or few backfills at steady state.
• Auto-discovering Discord max_concurrency to expand from 16 buckets.
• Replacing etcd identify mutex with in-process semaphore (separate cleanup).

Test plan

• go build ./... clean
• go test ./internal/gatewayws/... ./internal/manager/... ./discord/... ./handler/... clean
• Deploy to non-prod cluster; confirm log lines: identify lock acquired wait=..., stabilize semaphore acquired/released held=..., sweep batch dispatched ...
• Force a mass reconnect; confirm reconnect completes faster than today even with default config
• Bump IDENTIFY_STABILIZE_CONCURRENCY=4, observe state.DB batcher queue depth — if stable, ratchet further
• Confirm sweep wraps after a full cycle; validate cursor persists across restarts via etcd

🤖 Generated with Claude Code

[gemini-code-assist]

Code Review

This pull request implements the first phase of the 'Faster Gateway Mass Reconnect' design, aimed at reducing shard re-identification time by decoupling Discord's identify pacing from database stabilization. Key changes include moving the WebSocket handshake outside the identify lock, releasing the lock after a fixed pacing window, and introducing a cross-shard semaphore to manage backfill concurrency. Review feedback identifies critical race conditions in background goroutines where session state and contexts are captured by pointer, which could lead to premature lock releases or failures during session restarts.

[gemini-code-assist]

The background goroutine in scheduleIdentifyLockRelease captures the Session pointer (s), which leads to a race condition if the session restarts. When Open is called again (e.g., after a connection failure), s.ctx and s.identifyMu are overwritten. The goroutine from the previous attempt will then be waiting on the new session's context and potentially releasing the new session's identify lock prematurely.

Additionally, the comment mentions context.Background to survive unwinding, but the implementation uses s.ctx. If s.ctx is cancelled, s.releaseIdentifyLock() will fail because it also uses the cancelled s.ctx for the etcd Unlock call.

You should capture the context, mutex, and logger locally before starting the goroutine, and use a background context with a timeout for the actual unlock operation.

func (s *Session) scheduleIdentifyLockRelease() {
	pacing := s.identifyPacing
	if pacing <= 0 {
		pacing = IdentifyWaitTime
	}
	hold := pacing + readyGrace
	
	// Capture session state to avoid races if the session restarts.
	ctx, mu, log := s.ctx, s.identifyMu, s.log

	go func() {
		t := time.NewTimer(hold)
		defer t.Stop()
		select {
		case <-t.C:
		case <-ctx.Done():
		}
		if mu == nil || mu.Key() == "" {
			return
		}
		if mu.IsOwner().Result != etcdserverpb.Compare_EQUAL {
			return
		}

		// Use a background context for the unlock to ensure it completes
		// even if the session context has been cancelled.
		unlockCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()

		log.Info(ctx, "release identify lock after pacing", slog.F("key", mu.Key()))
		if err := mu.Unlock(unlockCtx); err != nil {
			log.Error(ctx, "release identify lock after pacing", slog.Error(err))
		}
	}()
}

[gemini-code-assist]

Similar to scheduleIdentifyLockRelease, holdStabilize captures the Session pointer and accesses s.ctx inside a background goroutine. If the session restarts and s.ctx is reassigned, the goroutine will wait on the wrong context.

Capture s.ctx locally at the start of the function to ensure the goroutine remains bound to the session attempt that started it.

func (s *Session) holdStabilize() {
	if s.stabilizeSem == nil || s.stabilizeDuration <= 0 {
		return
	}

	ctx, log := s.ctx, s.log
	max := s.stabilizeMaxDuration
	if max <= 0 {
		max = 2 * s.stabilizeDuration
	}

	acquireStart := time.Now()
	select {
	case s.stabilizeSem <- struct{}{}:
	case <-ctx.Done():
		return
	}
	acquired := time.Now()
	log.Info(ctx, "stabilize semaphore acquired",
		slog.F("wait", acquired.Sub(acquireStart).String()))

	defer func() {
		select {
		case <-s.stabilizeSem:
		default:
		}
		log.Info(ctx, "stabilize semaphore released",
			slog.F("held", time.Since(acquired).String()))
	}()

	hold := min(s.stabilizeDuration, max)
	t := time.NewTimer(hold)
	defer t.Stop()
	select {
	case <-t.C:
	case <-ctx.Done():
	}
}