diff --git a/pkg/vmcp/health/monitor.go b/pkg/vmcp/health/monitor.go index bca767421f..41b99a98ea 100644 --- a/pkg/vmcp/health/monitor.go +++ b/pkg/vmcp/health/monitor.go @@ -264,6 +264,7 @@ func NewMonitor( // Create status tracker with circuit breaker configuration // The status tracker will lazily initialize circuit breakers as needed statusTracker := newStatusTracker(config.UnhealthyThreshold, config.CircuitBreaker) + statusTracker.checkInterval = config.CheckInterval // The client (directly or via the telemetry decorator) optionally reports the // negotiated MCP revision for the status read-model; nil when unsupported. @@ -458,6 +459,10 @@ func (m *Monitor) UpdateBackends(newBackends []vmcp.Backend) { } } + // Prune expired tombstones to bound removedBackends (covers churn where + // old IDs are never re-checked via isRemoved). + m.statusTracker.pruneExpiredRemovedBackends() + if membershipChanged { m.changes.notify() } @@ -502,6 +507,9 @@ func (m *Monitor) monitorBackend(ctx context.Context, backend *vmcp.Backend, isI // performHealthCheck performs a single health check for a backend and updates status. func (m *Monitor) performHealthCheck(ctx context.Context, backend *vmcp.Backend) { + // Opportunistically prune expired tombstones to bound removedBackends + // even when UpdateBackends is not called for a long period. + m.statusTracker.pruneExpiredRemovedBackends() slog.Debug("performing health check for backend", "backend", backend.Name, "url", backend.BaseURL) // Check if circuit breaker allows health check diff --git a/pkg/vmcp/health/regression_5860_tombstone_test.go b/pkg/vmcp/health/regression_5860_tombstone_test.go new file mode 100644 index 0000000000..0d1bf616fe --- /dev/null +++ b/pkg/vmcp/health/regression_5860_tombstone_test.go @@ -0,0 +1,151 @@ +// SPDX-FileCopyrightText: Copyright 2026 Stacklok, Inc. +// SPDX-License-Identifier: Apache-2.0 + +package health + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + + "github.com/stacklok/toolhive/pkg/vmcp" + "github.com/stacklok/toolhive/pkg/vmcp/mocks" +) + +// TestRemovedBackends_Bounded_5860 verifies that removedBackends tombstones +// are bounded and expire after TTL. On the leaky base, removedBackends grows +// without bound as distinct backend IDs churn. +func TestRemovedBackends_Bounded_5860(t *testing.T) { + t.Parallel() + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockClient := mocks.NewMockBackendClient(ctrl) + mockClient.EXPECT(). + ListCapabilities(gomock.Any(), gomock.Any()). + Return(&vmcp.CapabilityList{}, nil). + AnyTimes() + + // Use short check interval so TTL (2*interval) is short for test. + config := MonitorConfig{ + CheckInterval: 20 * time.Millisecond, + UnhealthyThreshold: 1, + Timeout: 10 * time.Millisecond, + } + monitor, err := NewMonitor(mockClient, nil, config) + require.NoError(t, err) + + ctx := context.Background() + require.NoError(t, monitor.Start(ctx)) + defer func() { _ = monitor.Stop() }() + + // Churn 50 distinct backends: each added then removed. + for i := 0; i < 50; i++ { + id := fmt.Sprintf("backend-%d", i) + b := vmcp.Backend{ID: id, Name: id, BaseURL: "http://example.com"} + monitor.UpdateBackends([]vmcp.Backend{b}) + monitor.UpdateBackends([]vmcp.Backend{}) + } + + monitor.statusTracker.mu.RLock() + initial := len(monitor.statusTracker.removedBackends) + monitor.statusTracker.mu.RUnlock() + require.Equal(t, 50, initial, "precondition: churn should create 50 tombstones") + + // Wait for TTL (2*CheckInterval = 40ms) plus margin, so tombstones should expire. + time.Sleep(60 * time.Millisecond) + + // Trigger pruning via UpdateBackends (after fix, this prunes expired entries). + monitor.UpdateBackends([]vmcp.Backend{}) + + monitor.statusTracker.mu.RLock() + after := len(monitor.statusTracker.removedBackends) + monitor.statusTracker.mu.RUnlock() + + assert.Less(t, after, 10, + "removedBackends must be bounded after TTL expiry (leak if still 50); after=%d", after) +} + +// TestRemovedBackends_IsRemoved_Expiry_5860 verifies that isRemoved +// tombstone expires after TTL and allows re-recording health checks. +// On base, isRemoved never expires, so a removed backend stays ignored forever. +func TestRemovedBackends_IsRemoved_Expiry_5860(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3, nil) + // Use short interval so TTL = 2*interval = 40ms is fast for test. + tracker.checkInterval = 20 * time.Millisecond + backendID := "backend-expire-test" + + // Simulate removal. + tracker.RemoveBackend(backendID) + + // Immediately, isRemoved should be true (race protection). + tracker.mu.RLock() + isRemovedNow := tracker.isRemoved(backendID) + tracker.mu.RUnlock() + require.True(t, isRemovedNow, "immediately after RemoveBackend, isRemoved must be true") + + // Simulate expiry by moving the stored timestamp to the past beyond TTL. + // On fixed version (map[string]time.Time) this makes isRemoved prune and return false. + // On base (map[string]bool) there is no timestamp — entry stays true forever, + // so this test FAILs on base, proving the leak. + tracker.mu.Lock() + tracker.removedBackends[backendID] = time.Now().Add(-100 * time.Millisecond) + tracker.mu.Unlock() + + tracker.mu.Lock() + isRemovedAfter := tracker.isRemoved(backendID) + // After expiry, entry must be pruned, so map should not contain the ID. + _, stillExists := tracker.removedBackends[backendID] + tracker.mu.Unlock() + + assert.False(t, isRemovedAfter, "isRemoved must be false after TTL expiry (tombstone should expire)") + assert.False(t, stillExists, "expired tombstone must be pruned from map") + + // Verify that after expiry, health recording is no longer suppressed: + // RecordSuccess should create a new state instead of being ignored. + changed := tracker.RecordSuccess(backendID, "backend-expire-test", "healthy") + // On fixed version, RecordSuccess will not be ignored and will create state. + // We verify state exists. + tracker.mu.RLock() + _, exists := tracker.states[backendID] + tracker.mu.RUnlock() + assert.True(t, exists, "after tombstone expiry, RecordSuccess must create state (not be ignored)") + _ = changed // suppress unused warning; advertisability flip is not asserted here +} + +// TestRemovedBackends_PruneExpired_5860 verifies that pruneExpired removes old tombstones. +func TestRemovedBackends_PruneExpired_5860(t *testing.T) { + t.Parallel() + + tracker := newStatusTracker(3, nil) + tracker.checkInterval = 20 * time.Millisecond + // Insert 3 tombstones with old timestamps. + past := time.Now().Add(-100 * time.Millisecond) + tracker.mu.Lock() + tracker.removedBackends["old-1"] = past + tracker.removedBackends["old-2"] = past + tracker.removedBackends["recent"] = time.Now() + tracker.mu.Unlock() + + tracker.pruneExpiredRemovedBackends() + + tracker.mu.RLock() + _, old1Exists := tracker.removedBackends["old-1"] + _, old2Exists := tracker.removedBackends["old-2"] + _, recentExists := tracker.removedBackends["recent"] + count := len(tracker.removedBackends) + tracker.mu.RUnlock() + + assert.False(t, old1Exists, "expired old-1 must be pruned") + assert.False(t, old2Exists, "expired old-2 must be pruned") + assert.True(t, recentExists, "recent must not be pruned") + assert.Equal(t, 1, count, "only recent should remain") +} diff --git a/pkg/vmcp/health/status.go b/pkg/vmcp/health/status.go index 019eb6020a..447c8e5961 100644 --- a/pkg/vmcp/health/status.go +++ b/pkg/vmcp/health/status.go @@ -51,7 +51,9 @@ type statusTracker struct { // removedBackends tracks backends that were explicitly removed to prevent // race conditions where in-flight health checks re-create removed backends. - removedBackends map[string]bool + // The value is the removal timestamp; entries expire after removedTTL() + // (2*checkInterval or 5m default) to bound the map. + removedBackends map[string]time.Time // unhealthyThreshold is the number of consecutive failures before marking unhealthy. unhealthyThreshold int @@ -59,6 +61,10 @@ type statusTracker struct { // circuitBreakerConfig contains circuit breaker configuration. // nil means circuit breaker is disabled. circuitBreakerConfig *CircuitBreakerConfig + + // checkInterval is the health check interval used to compute the + // removedBackend TTL (2*interval). Zero means use default 5m. + checkInterval time.Duration } // newStatusTracker creates a new status tracker. @@ -77,16 +83,58 @@ func newStatusTracker(unhealthyThreshold int, circuitBreakerConfig *CircuitBreak return &statusTracker{ states: make(map[string]*backendHealthState), - removedBackends: make(map[string]bool), + removedBackends: make(map[string]time.Time), unhealthyThreshold: unhealthyThreshold, circuitBreakerConfig: circuitBreakerConfig, } } +// removedTTL returns the TTL for removedBackend tombstones. +// When checkInterval is set (via Monitor), TTL is 2*interval; otherwise 5m default. +// The 2*interval factor bounds the map while preserving the race-protection window: +// in-flight health checks that started before RemoveBackend can still be +// in-flight for up to Timeout (10s) + network latency; 2 ticks (60s at default +// 30s interval) comfortably covers this window without retaining tombstones +// indefinitely under ID churn. Zero interval (direct newStatusTracker in tests) +// falls back to 5m to avoid premature expiry. +func (t *statusTracker) removedTTL() time.Duration { + if t.checkInterval > 0 { + return 2 * t.checkInterval + } + return 5 * time.Minute +} + +// pruneExpiredRemovedBackendsLocked deletes tombstones older than TTL. +// Must be called with t.mu held (exclusive). +func (t *statusTracker) pruneExpiredRemovedBackendsLocked(now time.Time) { + ttl := t.removedTTL() + for id, ts := range t.removedBackends { + if now.Sub(ts) > ttl { + delete(t.removedBackends, id) + } + } +} + +// pruneExpiredRemovedBackends deletes tombstones older than TTL. +// It acquires the lock, so it can be called without holding it. +func (t *statusTracker) pruneExpiredRemovedBackends() { + t.mu.Lock() + defer t.mu.Unlock() + t.pruneExpiredRemovedBackendsLocked(time.Now()) +} + // isRemoved checks if a backend has been explicitly removed. -// Must be called with lock held. +// Must be called with lock held. Expired tombstones are lazily pruned. func (t *statusTracker) isRemoved(backendID string) bool { - return t.removedBackends[backendID] + ts, ok := t.removedBackends[backendID] + if !ok { + return false + } + if time.Since(ts) > t.removedTTL() { + delete(t.removedBackends, backendID) + return false + } + return true } // getOrCreateState retrieves an existing backend state or creates a new one with the specified initial values. @@ -454,13 +502,16 @@ func (t *statusTracker) IsHealthy(backendID string) bool { // RemoveBackend removes a backend from the status tracker. // The backend is marked as removed to prevent race conditions where in-flight -// health checks might try to re-create the backend state. +// health checks might try to re-create the backend state. The tombstone +// expires after removedTTL() to bound the map. func (t *statusTracker) RemoveBackend(backendID string) { t.mu.Lock() defer t.mu.Unlock() delete(t.states, backendID) - t.removedBackends[backendID] = true + // Prune expired tombstones before adding the new one to keep the map bounded. + t.pruneExpiredRemovedBackendsLocked(time.Now()) + t.removedBackends[backendID] = time.Now() } // ClearRemovedFlag clears the "removed" flag for a backend. diff --git a/pkg/vmcp/server/regression_5860_health_resync_test.go b/pkg/vmcp/server/regression_5860_health_resync_test.go new file mode 100644 index 0000000000..1a45e5744e --- /dev/null +++ b/pkg/vmcp/server/regression_5860_health_resync_test.go @@ -0,0 +1,157 @@ +// SPDX-FileCopyrightText: Copyright 2026 Stacklok, Inc. +// SPDX-License-Identifier: Apache-2.0 + +package server + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + vmcpsession "github.com/stacklok/toolhive/pkg/vmcp/session" +) + +// selectiveSessionManager is a per-ID liveness stub for regression test. +type selectiveSessionManager struct { + mu sync.RWMutex + alive map[string]bool +} + +func (m *selectiveSessionManager) GetMultiSession(_ context.Context, sessionID string) (vmcpsession.MultiSession, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + ok := m.alive[sessionID] + return nil, ok +} +func (m *selectiveSessionManager) Generate() string { panic("Generate unexpected") } +func (m *selectiveSessionManager) Validate(string) (bool, error) { + panic("Validate unexpected") +} +func (m *selectiveSessionManager) Terminate(string) (bool, error) { return false, nil } +func (m *selectiveSessionManager) CreateSession(context.Context, string, vmcpsession.ListChangedSink) (vmcpsession.MultiSession, error) { + panic("CreateSession unexpected") +} +func (m *selectiveSessionManager) DecorateSession(string, func(vmcpsession.MultiSession) vmcpsession.MultiSession) error { + panic("DecorateSession unexpected") +} +func (m *selectiveSessionManager) NotifyBackendExpired(string, string, map[string]string) {} + +// TestHealthResyncRegistry_BoundedAfterSessionExpiry_5860 verifies that +// resyncSessionsOnBackendHealthChange prunes dead sessions synchronously and +// only triggers live sessions. On the leaky base, dead workers are retained +// until the next async liveness guard runs, and all workers are triggered. +func TestHealthResyncRegistry_BoundedAfterSessionExpiry_5860(t *testing.T) { + t.Parallel() + + // Track which workers were actually triggered. + var mu sync.Mutex + triggered := make(map[string]int) + var totalTriggers atomic.Int32 + + makeWorker := func(id string) *listChangedResyncWorker { + return &listChangedResyncWorker{ + baseCtx: context.Background(), + run: func(ctx context.Context, purge bool) { + mu.Lock() + triggered[id]++ + mu.Unlock() + totalTriggers.Add(1) + }, + } + } + + alive := map[string]bool{ + "live-1": true, + "live-2": true, + // dead-1, dead-2 remain false (expired) + } + mgr := &selectiveSessionManager{alive: alive} + srv := &Server{ + core: &fakeCore{}, + vmcpSessionMgr: mgr, + resyncBaseCtx: context.Background(), + } + + // Add 4 workers: 2 live, 2 dead. + for _, id := range []string{"live-1", "live-2", "dead-1", "dead-2"} { + srv.healthResync.add(id, makeWorker(id)) + } + require.Len(t, srv.healthResync.snapshot(), 4, "precondition: registry holds all sessions") + + // Fan-out: should synchronously prune dead entries and only trigger live. + srv.resyncSessionsOnBackendHealthChange(1) + + // Give triggered workers a moment to run (they are async via trigger). + // Use Eventually to wait for expected live triggers. + require.Eventually(t, func() bool { return totalTriggers.Load() >= 2 }, + 2*time.Second, 10*time.Millisecond, "live workers must be triggered") + + // Allow a short window for any (incorrect) dead triggers to surface. + // On base, dead workers will also have been triggered, so total will be 4. + time.Sleep(100 * time.Millisecond) + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + // If dead were triggered, triggered map will contain them. + _, dead1 := triggered["dead-1"] + _, dead2 := triggered["dead-2"] + return dead1 || dead2 || len(triggered) == 2 + }, 500*time.Millisecond, 10*time.Millisecond, "wait for trigger set to settle") + + mu.Lock() + defer mu.Unlock() + + // Invariant: |workers| == |liveSessions| bounded. + assert.Equal(t, 2, len(srv.healthResync.snapshot()), + "registry must be bounded to live sessions after fan-out (dead pruned synchronously)") + assert.Equal(t, 2, len(triggered), + "only live sessions must be triggered (dead must be filtered at fan-out)") + assert.Contains(t, triggered, "live-1") + assert.Contains(t, triggered, "live-2") + assert.NotContains(t, triggered, "dead-1", "dead session must not be triggered") + assert.NotContains(t, triggered, "dead-2", "dead session must not be triggered") +} + +// TestHealthResyncRegistry_FilterReducesTriggers_5860 is a second guard that +// ensures filtering happens BEFORE trigger, not just via async liveness prune. +func TestHealthResyncRegistry_FilterReducesTriggers_5860(t *testing.T) { + t.Parallel() + + alive := map[string]bool{"live": true} + mgr := &selectiveSessionManager{alive: alive} + srv := &Server{ + core: &fakeCore{}, + vmcpSessionMgr: mgr, + resyncBaseCtx: context.Background(), + } + + var liveTriggers atomic.Int32 + var deadTriggers atomic.Int32 + + liveWorker := &listChangedResyncWorker{ + baseCtx: context.Background(), + run: func(context.Context, bool) { liveTriggers.Add(1) }, + } + deadWorker := &listChangedResyncWorker{ + baseCtx: context.Background(), + run: func(context.Context, bool) { deadTriggers.Add(1) }, + } + + srv.healthResync.add("live", liveWorker) + srv.healthResync.add("dead", deadWorker) + + srv.resyncSessionsOnBackendHealthChange(42) + + require.Eventually(t, func() bool { return liveTriggers.Load() == 1 }, + 2*time.Second, 10*time.Millisecond, "live must be triggered exactly once") + // Dead must never be triggered; wait a bit to ensure no late trigger. + // On base, dead will be triggered (incorrect). Use Sleep then assert. + time.Sleep(100 * time.Millisecond) + assert.Equal(t, int32(0), deadTriggers.Load(), "dead session must not be triggered at fan-out") + assert.Equal(t, 1, len(srv.healthResync.snapshot()), "registry must contain only live after fan-out") +} diff --git a/pkg/vmcp/server/serve_health_resync.go b/pkg/vmcp/server/serve_health_resync.go index e8b6663c88..1667433856 100644 --- a/pkg/vmcp/server/serve_health_resync.go +++ b/pkg/vmcp/server/serve_health_resync.go @@ -4,6 +4,7 @@ package server import ( + "context" "log/slog" "sync" @@ -38,16 +39,15 @@ import ( // enabled only — optimizer-mode sessions are never registered because the // fan-out is a no-op for them in PR1, and with health monitoring disabled // there is no OnChange subscriber, so nothing would ever trigger the fan-out -// or run the lazy prune below) and removed eagerly on every termination -// path the server observes — registration failure, binding-failure -// termination, and SDK-initiated termination (HTTP DELETE), the last via +// or run the prune) and removed eagerly on every termination path the server +// observes — registration failure, binding-failure termination, and +// SDK-initiated termination (HTTP DELETE), the last via // pruneOnTerminateSessionIDManager. Sessions that end without any Terminate -// call (TTL expiry) are pruned lazily by runListChangedResync when a -// triggered resync finds them gone. Between fan-out events the registry can -// therefore still hold entries for expired sessions; each such entry retains -// the worker closure (the SDK ClientSession and the registration-time -// identity + forwarded headers), is skipped harmlessly by the worker's -// liveness guard, and is pruned on the next trigger. +// call (TTL expiry) are pruned eagerly at fan-out by +// resyncSessionsOnBackendHealthChange (which filters dead sessions before +// triggering) and lazily by runListChangedResync when a triggered resync +// finds them gone. The fan-out filter ensures |workers| == |liveSessions| +// and bounds the registry between health-change events. type healthResyncRegistry struct { mu sync.Mutex workers map[string]*listChangedResyncWorker @@ -91,15 +91,15 @@ func (r *healthResyncRegistry) remove(sessionID string) { delete(r.workers, sessionID) } -// snapshot returns the currently registered workers. The copy lets callers -// trigger workers without holding the registry lock (trigger may start a -// goroutine that re-enters remove via the liveness prune). -func (r *healthResyncRegistry) snapshot() []*listChangedResyncWorker { +// snapshot returns the currently registered workers keyed by session ID. +// The copy lets callers trigger workers without holding the registry lock +// (trigger may start a goroutine that re-enters remove via the liveness prune). +func (r *healthResyncRegistry) snapshot() map[string]*listChangedResyncWorker { r.mu.Lock() defer r.mu.Unlock() - out := make([]*listChangedResyncWorker, 0, len(r.workers)) - for _, w := range r.workers { - out = append(out, w) + out := make(map[string]*listChangedResyncWorker, len(r.workers)) + for id, w := range r.workers { + out[id] = w } return out } @@ -146,7 +146,19 @@ func (s *Server) resyncSessionsOnBackendHealthChange(generation uint64) { workers := s.healthResync.snapshot() slog.Debug("backend health change: triggering tools resync for live sessions", "generation", generation, "sessions", len(workers)) - for _, w := range workers { + // Filter dead sessions synchronously before triggering to keep the + // registry bounded (|workers| == |liveSessions|) and avoid unnecessary + // work. The snapshot is already a copy, so GetMultiSession I/O does not + // hold the registry lock. Removal is done via the registry's mutex. + ctx := s.resyncBaseCtx + if ctx == nil { + ctx = context.Background() + } + for sessionID, w := range workers { + if _, ok := s.vmcpSessionMgr.GetMultiSession(ctx, sessionID); !ok { + s.healthResync.remove(sessionID) + continue + } w.trigger(false) } } diff --git a/pkg/vmcp/server/serve_health_resync_test.go b/pkg/vmcp/server/serve_health_resync_test.go index 356484a8ae..2b35cc52c0 100644 --- a/pkg/vmcp/server/serve_health_resync_test.go +++ b/pkg/vmcp/server/serve_health_resync_test.go @@ -14,6 +14,7 @@ import ( "github.com/stacklok/toolhive-core/mcpcompat/mcp" "github.com/stacklok/toolhive-core/mcpcompat/server" + "github.com/stacklok/toolhive/pkg/auth" "github.com/stacklok/toolhive/pkg/vmcp" "github.com/stacklok/toolhive/pkg/vmcp/health" "github.com/stacklok/toolhive/pkg/vmcp/optimizer" @@ -100,22 +101,38 @@ func (f *fakeToolsSession) setToolsCalls() int { return f.setSessionToolsCalls } -// gatedSessionManager is a stubSessionManager whose GetMultiSession blocks on -// gate (a closed channel unblocks all callers), letting the coalescing test -// deterministically hold the first resync in flight while further deliveries -// arrive. +// gatedSessionManager is a stubSessionManager whose GetMultiSession is +// counted but not blocking. The coalescing test now blocks on the backend +// sweep (ListTools) via gatedFakeCore, not on GetMultiSession, because the +// fan-out liveness filter (#5860) also calls GetMultiSession synchronously +// and would otherwise block the burst delivery. type gatedSessionManager struct { stubSessionManager - gate chan struct{} calls atomic.Int32 } -func (m *gatedSessionManager) GetMultiSession(context.Context, string) (vmcpsession.MultiSession, bool) { +func (m *gatedSessionManager) GetMultiSession(_ context.Context, _ string) (vmcpsession.MultiSession, bool) { m.calls.Add(1) - <-m.gate return nil, true } +// gatedFakeCore blocks the first ListTools call on gate, letting the +// coalescing test hold the first resync's backend sweep in flight. +type gatedFakeCore struct { + *fakeCore + gate chan struct{} + calls atomic.Int32 +} + +func (c *gatedFakeCore) ListTools(ctx context.Context, id *auth.Identity) ([]vmcp.Tool, error) { + // Only the first ListTools should block; the follow-up should proceed + // after gate is closed. + if c.calls.Add(1) == 1 { + <-c.gate + } + return c.fakeCore.ListTools(ctx, id) +} + // TestResyncSessionsOnBackendHealthChange_CoalescesBurst verifies a burst of // health-change deliveries collapses into the in-flight resync plus exactly // one follow-up run (the per-session worker's dirty-flag coalescing), not one @@ -124,9 +141,11 @@ func TestResyncSessionsOnBackendHealthChange_CoalescesBurst(t *testing.T) { t.Parallel() fc := &fakeCore{tools: []vmcp.Tool{{Name: "t"}}} - mgr := &gatedSessionManager{stubSessionManager: stubSessionManager{alive: true}, gate: make(chan struct{})} + gate := make(chan struct{}) + gfc := &gatedFakeCore{fakeCore: fc, gate: gate} + mgr := &gatedSessionManager{stubSessionManager: stubSessionManager{alive: true}} srv := &Server{ - core: fc, + core: gfc, vmcpSessionMgr: mgr, resyncBaseCtx: context.Background(), } @@ -134,9 +153,9 @@ func TestResyncSessionsOnBackendHealthChange_CoalescesBurst(t *testing.T) { _, toolsWorker := srv.buildListChangedSink("sess-1", sess, nil, nil) srv.healthResync.add("sess-1", toolsWorker) - // First delivery starts the worker; it blocks inside the liveness guard. + // First delivery starts the worker; it blocks inside ListTools. srv.resyncSessionsOnBackendHealthChange(1) - require.Eventually(t, func() bool { return mgr.calls.Load() == 1 }, + require.Eventually(t, func() bool { return gfc.calls.Load() == 1 }, 2*time.Second, 10*time.Millisecond, "first resync must be in flight") // Nine more deliveries arrive while the first resync is blocked: they must @@ -144,7 +163,7 @@ func TestResyncSessionsOnBackendHealthChange_CoalescesBurst(t *testing.T) { for gen := uint64(2); gen <= 10; gen++ { srv.resyncSessionsOnBackendHealthChange(gen) } - close(mgr.gate) + close(gate) // The blocked run completes and exactly one follow-up run drains the // coalesced deliveries: two re-derivations total, never ten.