From bad12e44b65847abf70c9972b44db8a51e9b569c Mon Sep 17 00:00:00 2001 From: Igor Sirotin Date: Mon, 6 Jul 2026 09:40:12 +0100 Subject: [PATCH] fix(filter): correct inverted idle-subscription cleanup check The periodic cleanup in SubscribersMap.cleanUp deleted subscribers whose elapsed time since last activity was *less* than the idle timeout, and kept those that had gone idle. This is backwards: active subscribers (that keep their subscription alive via pings, which call Refresh) were being dropped within a cleanup cycle, while genuinely idle subscribers were never reaped and leaked forever. Flip the comparison so a subscriber is removed only once it has been idle for longer than the timeout. Also rewrite TestCleanup (which previously encoded the buggy behavior) and add TestCleanupKeepsActiveSubscribers, which fails on the old inverted check. Co-Authored-By: Claude Opus 4.8 (1M context) --- waku/v2/protocol/filter/subscribers_map.go | 2 +- .../protocol/filter/subscribers_map_test.go | 40 +++++++++++++++++-- 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/waku/v2/protocol/filter/subscribers_map.go b/waku/v2/protocol/filter/subscribers_map.go index b539357e..fe32cb32 100644 --- a/waku/v2/protocol/filter/subscribers_map.go +++ b/waku/v2/protocol/filter/subscribers_map.go @@ -250,7 +250,7 @@ func (sub *SubscribersMap) cleanUp(ctx context.Context, cleanupInterval time.Dur sub.Lock() for peerID, lastSeen := range sub.lastSeen { elapsedTime := time.Since(lastSeen) - if elapsedTime < sub.timeout { + if elapsedTime > sub.timeout { _ = sub.deleteAll(peerID) } diff --git a/waku/v2/protocol/filter/subscribers_map_test.go b/waku/v2/protocol/filter/subscribers_map_test.go index 7b7dd74b..0ccdbb34 100644 --- a/waku/v2/protocol/filter/subscribers_map_test.go +++ b/waku/v2/protocol/filter/subscribers_map_test.go @@ -107,13 +107,15 @@ func TestRemoveBogus(t *testing.T) { require.Error(t, err) } +// TestCleanup verifies that a subscriber which has not been refreshed within +// the idle timeout is removed by the periodic cleanup. func TestCleanup(t *testing.T) { - subs := NewSubscribersMap(2 * time.Second) + subs := NewSubscribersMap(1 * time.Second) ctx, cancel := context.WithCancel(context.Background()) defer cancel() - go subs.cleanUp(ctx, 500*time.Millisecond) + go subs.cleanUp(ctx, 100*time.Millisecond) peerId := createPeerID(t) @@ -125,7 +127,8 @@ func TestCleanup(t *testing.T) { _, exists := subs.Get(peerId) require.True(t, exists) - time.Sleep(2 * time.Second) + // Let the subscriber go idle past the timeout; it must be cleaned up. + time.Sleep(1500 * time.Millisecond) hasSubs = subs.Has(peerId) require.False(t, hasSubs) @@ -133,3 +136,34 @@ func TestCleanup(t *testing.T) { _, exists = subs.Get(peerId) require.False(t, exists) } + +// TestCleanupKeepsActiveSubscribers verifies that a subscriber which keeps +// getting refreshed (e.g. via pings) survives the periodic cleanup, and is +// only removed once it goes idle past the timeout. This guards against a +// regression where the idle check was inverted and active subscribers were +// deleted while idle ones were kept forever. +func TestCleanupKeepsActiveSubscribers(t *testing.T) { + timeout := 1 * time.Second + subs := NewSubscribersMap(timeout) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + go subs.cleanUp(ctx, 100*time.Millisecond) + + peerId := createPeerID(t) + subs.Set(peerId, PUBSUB_TOPIC, []string{"topic1", "topic2"}) + + // Keep the subscriber active for well over the timeout by refreshing it + // periodically. It must never be cleaned up while active. + for i := 0; i < 15; i++ { + time.Sleep(100 * time.Millisecond) + subs.Refresh(peerId) + require.True(t, subs.Has(peerId), "active subscriber was removed while being refreshed") + } + + // Stop refreshing; once idle past the timeout it must be removed. + require.Eventually(t, func() bool { + return !subs.Has(peerId) + }, 3*time.Second, 100*time.Millisecond, "idle subscriber was not cleaned up") +}