mirror of
https://github.com/logos-messaging/logos-messaging-go.git
synced 2026-07-18 23:30:01 +00:00
fix(filter): correct inverted idle-subscription cleanup check (#1314)
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
f3fc70002f
commit
b007066aa5
@ -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)
|
||||
}
|
||||
|
||||
|
||||
@ -107,16 +107,20 @@ func TestRemoveBogus(t *testing.T) {
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
// TestCleanup verifies the idle-subscription cleanup lifecycle: a subscriber
|
||||
// that keeps getting refreshed (e.g. via pings) survives the periodic cleanup
|
||||
// while active, and is removed only once it goes idle past the timeout. This
|
||||
// also guards against a regression where the idle check was inverted, deleting
|
||||
// active subscribers while keeping idle ones forever.
|
||||
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)
|
||||
|
||||
subs.Set(peerId, PUBSUB_TOPIC, []string{"topic1", "topic2"})
|
||||
|
||||
hasSubs := subs.Has(peerId)
|
||||
@ -125,10 +129,18 @@ func TestCleanup(t *testing.T) {
|
||||
_, exists := subs.Get(peerId)
|
||||
require.True(t, exists)
|
||||
|
||||
time.Sleep(2 * time.Second)
|
||||
// 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")
|
||||
}
|
||||
|
||||
hasSubs = subs.Has(peerId)
|
||||
require.False(t, hasSubs)
|
||||
// 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")
|
||||
|
||||
_, exists = subs.Get(peerId)
|
||||
require.False(t, exists)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user