diff --git a/docs/releases.md b/docs/releases.md index e026969e..b78844a5 100644 --- a/docs/releases.md +++ b/docs/releases.md @@ -2302,6 +2302,7 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release * Daily limits now reset on a rolling 24-hour period per account or IP address instead of for everyone at midnight UTC: each visitor's period starts at its own time of day, which spreads the resets over the day. The account page and the `/v1/account` API (`stats.reset_at`) show when the visitor's counts reset * Visitors without a tier have a daily e-mail limit again (24h / `visitor-email-limit-replenish`), now that it is shared between servers and survives restarts * The daily reset job is gone; the `stats_*` columns of the user table are zeroed by the upgrade and no longer read or written +* The rate limit state of a visitor is forgotten about a minute after it has fully recovered (all buckets refilled, no attachment bandwidth used in its current daily period) instead of after a day without use, which frees the memory of one-off visitors such as floods of new IPs. The `visitors` stat and the `ntfy_visitors_total` metric therefore count the visitors of roughly the last minute, not of the last day ([#2126](https://github.com/binwiederhier/ntfy/pull/2126)) **Bug fixes + maintenance:** diff --git a/limit/service.go b/limit/service.go index f1ee6f9b..6b019adb 100644 --- a/limit/service.go +++ b/limit/service.go @@ -56,6 +56,10 @@ const ( // than a period is never current again pruneInterval = time.Hour pruneAfter = 25 * time.Hour + + // settledIdle is how long a key whose buckets are full again must go unused before Prune + // forgets its buckets, so a key in steady use is not dropped and recreated over and over + settledIdle = time.Minute ) // Service decides on usage from memory; see the package comment @@ -63,7 +67,7 @@ type Service struct { conf *Config now func() time.Time // Injectable clock, for tests counters [shardCount]*counterShard // Daily counts, kept until their period ends - buckets [shardCount]*bucketShard // Token buckets, per node, evicted when idle and full + buckets [shardCount]*bucketShard // Token buckets, per node, forgotten when settled (see Prune) store *Store // Nil: memory only watermark int64 // Time of the last read, unix seconds; the sync worker's only prunedAt int64 // Time of the last prune, unix seconds; the sync worker's only @@ -182,8 +186,7 @@ func (s *Service) Tokens(key string, rule Rule) int64 { if !ok { return rule.Burst // A fresh bucket is full; no need to create it to say so } - b := &e.buckets[rule.Kind] - b.refill(now, &rule) + b := e.refillNoLock(&rule, now) return int64(math.Max(0, math.Floor(b.tokens))) } @@ -194,11 +197,14 @@ func (s *Service) Refund(key string, kind Kind, n int64) { start := periodStart(key, s.now()) i := dailyIndex(kind) if i < 0 { + if kind != KindBandwidth { + return // Not tallied + } bs := s.buckets[shardIndex(key)] bs.mu.Lock() defer bs.mu.Unlock() if e, ok := bs.entries[key]; ok && e.periodStart >= start { - e.counts[kind] = max(0, e.counts[kind]-n) + e.bandwidth = max(0, e.bandwidth-n) } return } @@ -211,14 +217,15 @@ func (s *Service) Refund(key string, kind Kind, n int64) { } // Usage returns the key's counts in its current period, from memory: the daily counts (all -// nodes' as of the last sync, plus this node's since), and this node's units of the other kinds +// nodes' as of the last sync, plus this node's since), and this node's bandwidth. The other kinds +// are not tallied, so that their use does not keep the key for the rest of its period (see Prune). func (s *Service) Usage(key string) Usage { start := periodStart(key, s.now()) var u Usage bs := s.buckets[shardIndex(key)] bs.mu.Lock() if e, ok := bs.entries[key]; ok && e.periodStart >= start { - u.counts = e.counts + u.counts[KindBandwidth] = e.bandwidth } bs.mu.Unlock() cs := s.counters[shardIndex(key)] @@ -237,17 +244,17 @@ func (s *Service) ResetAt(key string) time.Time { return periodEnd(key, s.now()) } -// Prune forgets the buckets of keys that have not been used for longer than idle and are full -// again (a forgotten key comes back with full buckets), and the daily counts of periods that -// have ended. It returns how many keys' buckets it forgot. -func (s *Service) Prune(idle time.Duration) int { +// Prune forgets the buckets of keys that are settled (see bucketEntry.settledNoLock) and have not +// been used for settledIdle, and the daily counts of periods that have ended. It returns how many +// keys' buckets it forgot. +func (s *Service) Prune() int { now := s.now() - cutoff := now.Add(-idle).UnixNano() + cutoff := now.Add(-settledIdle).UnixNano() pruned := 0 for _, bs := range s.buckets { bs.mu.Lock() for key, e := range bs.entries { - if e.seen < cutoff && now.UnixNano() >= e.full { + if e.seen < cutoff && e.settledNoLock(key, now) { delete(bs.entries, key) pruned++ } @@ -468,8 +475,7 @@ func (s *Service) takeTokens(key string, rule *Rule, n int64, now time.Time) Dec e.countNoLock(key, rule.Kind, n, now) return Decision{OK: true} } - b := &e.buckets[rule.Kind] - b.refill(now, rule) + b := e.refillNoLock(rule, now) if n > rule.Burst { return Decision{} // Can never fit, so there is no time to wait for } else if b.tokens < float64(n) { @@ -500,7 +506,9 @@ func (s *Service) reset(key string, kind Kind) { bs.mu.Lock() if e, ok := bs.entries[key]; ok { e.buckets[kind].tokens = float64(e.buckets[kind].burst) - e.counts[kind] = 0 + if kind == KindBandwidth { + e.bandwidth = 0 + } } bs.mu.Unlock() i := dailyIndex(kind) @@ -539,17 +547,38 @@ func (bs *bucketShard) entryNoLock(key string, now time.Time) *bucketEntry { return e } -// countNoLock adds n units of a kind without a counter to the entry's counts of the current -// period, starting them over if the period has ended. The caller must hold the shard's lock. +// countNoLock adds n bytes of bandwidth to the entry's tally of the current period, starting it +// over if the period has ended; other kinds are counted in the key's counter, or not at all. The +// caller must hold the shard's lock. func (e *bucketEntry) countNoLock(key string, kind Kind, n int64, now time.Time) { - if dailyIndex(kind) >= 0 { - return // Counted in the key's counter + if kind != KindBandwidth { + return } if start := periodStart(key, now); e.periodStart < start { - e.counts = [numKinds]int64{} + e.bandwidth = 0 e.periodStart = start } - e.counts[kind] += n + e.bandwidth += n +} + +// settledNoLock returns true if forgetting the entry changes nothing: every bucket is full again +// (a forgotten key comes back with full buckets), and there is no bandwidth tally in the key's +// current period. The caller must hold the shard's lock. +func (e *bucketEntry) settledNoLock(key string, now time.Time) bool { + if now.UnixNano() < e.full { + return false + } + return e.bandwidth == 0 || e.periodStart < periodStart(key, now) +} + +// refillNoLock refills the entry's bucket for the rule and notes when it is full again: the rule +// may refill slower than the one the bucket was last charged with, also on a denied call. The +// caller must hold the shard's lock. +func (e *bucketEntry) refillNoLock(rule *Rule, now time.Time) *bucket { + b := &e.buckets[rule.Kind] + b.refill(now, rule) + e.noteFull(b, rule) + return b } // noteFull moves the time from which every bucket is full again to when this bucket is full, if diff --git a/limit/service_test.go b/limit/service_test.go index 4250d455..4b6cc6ce 100644 --- a/limit/service_test.go +++ b/limit/service_test.go @@ -213,9 +213,9 @@ func TestService_Allow_DailyChangeAppliesAtOnce(t *testing.T) { func TestService_Allow_Unlimited(t *testing.T) { s, _ := newTestServiceWithClock(t) for i := 0; i < 1000; i++ { - require.True(t, s.Allow("k", Rule{Kind: KindTopics}, 1).OK) + require.True(t, s.Allow("k", Rule{Kind: KindBandwidth}, 1).OK) } - require.Equal(t, int64(1000), s.Usage("k").Count(KindTopics)) + require.Equal(t, int64(1000), s.Usage("k").Count(KindBandwidth)) } func TestService_Allow_NeverRefills(t *testing.T) { @@ -282,7 +282,7 @@ func TestService_Prune(t *testing.T) { clock.advance(10 * time.Minute) require.True(t, s.Allow("fresh", rule, 1).OK) require.Equal(t, 2, s.Size()) - s.Prune(5 * time.Minute) + s.Prune() require.Equal(t, 1, s.Size()) require.True(t, s.Allow("old", rule, 1).OK) // Forgotten: back with a full bucket require.False(t, s.Allow("fresh", rule, 1).OK) // Kept @@ -300,11 +300,11 @@ func TestService_Prune_KeepsKeyUntilBucketsRefill(t *testing.T) { } require.True(t, s.Allow("key", rule, 1).OK) clock.advance(10 * time.Minute) - s.Prune(5 * time.Minute) + s.Prune() require.Equal(t, 1, s.Size()) require.False(t, s.Allow("key", rule, 1).OK) clock.advance(time.Hour) - s.Prune(5 * time.Minute) + s.Prune() require.Equal(t, 0, s.Size()) require.True(t, s.Allow("key", rule, 1).OK) } @@ -318,15 +318,159 @@ func TestService_Prune_DropsEndedPeriods(t *testing.T) { } require.True(t, s.Allow("k", rule, 4).OK) clock.advance(15 * time.Hour) - s.Prune(time.Minute) + s.Prune() require.Equal(t, 0, s.Size()) require.Equal(t, int64(4), s.Usage("k").Count(KindMessages)) require.Len(t, s.counters[shardIndex("k")].counters, 1) clock.advance(time.Hour) - s.Prune(time.Minute) + s.Prune() require.Empty(t, s.counters[shardIndex("k")].counters) } +func TestService_Prune_ForgetsSettledKeyAfterAMinute(t *testing.T) { + // A one-shot key (e.g. one of a flood of IPs) is forgotten once its buckets are full again and + // it went unused for settledIdle, not after a day + s, clock := newTestServiceWithClock(t) + rule := Rule{ + Kind: KindRequests, + Burst: 10, + Refill: time.Second, + } + require.True(t, s.Allow("k", rule, 1).OK) + clock.advance(settledIdle / 2) + require.Zero(t, s.Prune()) + require.Equal(t, 1, s.Size()) // Full again, but used too recently + clock.advance(settledIdle) + require.Equal(t, 1, s.Prune()) + require.Equal(t, 0, s.Size()) + require.Equal(t, rule.Burst, s.Tokens("k", rule)) +} + +func TestService_Prune_NoRefillDoesNotHoldKey(t *testing.T) { + // A bucket without refill is never full again, so it does not hold the key: like any settled + // key, it is forgotten after settledIdle and comes back with a full bucket + s, clock := newTestServiceWithClock(t) + rule := Rule{ + Kind: KindAccount, + Burst: 1, + } + require.True(t, s.Allow("k", rule, 1).OK) + require.False(t, s.Allow("k", rule, 1).OK) + clock.advance(2 * settledIdle) + require.Equal(t, 1, s.Prune()) + require.True(t, s.Allow("k", rule, 1).OK) +} + +func TestService_Prune_ForgettingIsARefillAtPruneTime(t *testing.T) { + // Forgetting a settled key is the same as refilling its buckets when it is forgotten: a + // wall clock stepped back afterwards finds the same tokens either way + clock := &testClock{t: time.Date(2026, 10, 25, 12, 0, 0, 0, time.UTC)} + pruned, kept := newService(&Config{}, clock.now), newService(&Config{}, clock.now) + rule := Rule{ + Kind: KindRequests, + Burst: 10, + Refill: time.Second, + } + for _, s := range []*Service{pruned, kept} { + require.True(t, s.Allow("k", rule, 10).OK) + } + clock.advance(time.Hour) + require.Equal(t, 1, pruned.Prune()) + require.Equal(t, rule.Burst, kept.Tokens("k", rule)) + clock.advance(-59 * time.Minute) // 5 s after the bucket was emptied + for _, s := range []*Service{pruned, kept} { + require.Equal(t, rule.Burst, s.Tokens("k", rule)) + require.True(t, s.Allow("k", rule, 10).OK) + require.False(t, s.Allow("k", rule, 1).OK) + } +} + +func TestService_Prune_DeniedSlowerRuleHoldsKey(t *testing.T) { + // A bucket charged with a fast rule and then denied with a slower one (a tier downgrade, a + // config change) is full again only at the slower rate: the key must be kept until then, or + // it comes back with a full bucket + s, clock := newTestServiceWithClock(t) + fast := Rule{ + Kind: KindRequests, + Burst: 2, + Refill: time.Second, + } + slow := Rule{ + Kind: KindRequests, + Burst: 2, + Refill: time.Hour, + } + require.True(t, s.Allow("k", fast, 2).OK) + clock.advance(time.Second) + require.False(t, s.Allow("k", slow, 2).OK) + clock.advance(2 * settledIdle) + require.Zero(t, s.Prune()) + require.Equal(t, 1, s.Size()) + require.False(t, s.Allow("k", slow, 1).OK) +} + +func TestService_Prune_TokensWithSlowerRuleHoldsKey(t *testing.T) { + // Tokens refills too, so a slower rule read there moves the time the bucket is full again + s, clock := newTestServiceWithClock(t) + fast := Rule{ + Kind: KindRequests, + Burst: 2, + Refill: time.Second, + } + slow := Rule{ + Kind: KindRequests, + Burst: 2, + Refill: time.Hour, + } + require.True(t, s.Allow("k", fast, 2).OK) + clock.advance(time.Second) + require.Zero(t, s.Tokens("k", slow)) + clock.advance(2 * settledIdle) + require.Zero(t, s.Prune()) + require.Zero(t, s.Tokens("k", slow)) +} + +func TestService_Prune_BandwidthTallyHoldsKeyUntilPeriodEnd(t *testing.T) { + // A big bandwidth bucket refills a small download in milliseconds, but the bytes stay in the + // key's usage until its period ends, so the key is kept until then + s, clock := newTestServiceWithClock(t) + burst := int64(15 * 1024 * 1024 * 1024) + rule := Rule{ + Kind: KindBandwidth, + Burst: burst, + Refill: oneDay / time.Duration(burst), + } + require.True(t, s.Allow("k", rule, 1024*1024).OK) + clock.advance(2 * settledIdle) + require.Zero(t, s.Prune()) + require.Equal(t, int64(1024*1024), s.Usage("k").Count(KindBandwidth)) + require.Equal(t, int64(1024*1024), s.Limiter("k", rule).Value()) + clock.t = s.ResetAt("k") + require.Equal(t, 1, s.Prune()) + require.Equal(t, Usage{}, s.Usage("k")) +} + +func TestService_Prune_LimiterViewAcrossPrune(t *testing.T) { + // A limiter view (e.g. a reader metering an upload) refers to the key, not to its entry: bytes + // it charged keep the entry, a revert finds them, and after a prune it charges a fresh entry + s, clock := newTestServiceWithClock(t) + rule := Rule{ + Kind: KindBandwidth, + Burst: 1000, + Refill: time.Millisecond, + } + l := s.Limiter("k", rule) + require.True(t, l.AllowN(100)) + clock.advance(2 * settledIdle) + require.Zero(t, s.Prune()) + require.True(t, l.AllowN(-100)) // Revert: the upload failed + require.Zero(t, l.Value()) + require.Equal(t, 1, s.Prune()) + require.True(t, l.AllowN(50)) + require.Equal(t, int64(50), l.Value()) + require.Equal(t, rule.Burst-50, s.Tokens("k", rule)) +} + func TestService_Allow_Concurrent(t *testing.T) { s := New(&Config{}) rule := Rule{ @@ -420,11 +564,11 @@ func TestService_TouchAndSeen(t *testing.T) { require.Equal(t, clock.now(), s.Seen("k")) clock.advance(10 * time.Minute) s.Touch("k") - s.Prune(5 * time.Minute) + s.Prune() require.Equal(t, 1, s.Size()) // Touched, so not idle require.Equal(t, clock.now(), s.Seen("k")) clock.advance(10 * time.Minute) - s.Prune(5 * time.Minute) + s.Prune() require.True(t, s.Seen("k").IsZero()) } @@ -497,13 +641,13 @@ func TestService_Prune_ConcurrentAllow(t *testing.T) { defer wg.Done() <-start for i := 0; i < 100; i++ { - s.Prune(time.Hour) + s.Prune() } }() close(start) wg.Wait() require.Zero(t, admitted.Load()) - require.Equal(t, int64(1), s.Usage("hot").Count(rule.Kind)) + require.Zero(t, s.Tokens("hot", rule)) require.Equal(t, 1, s.Size()) } @@ -777,7 +921,7 @@ func TestService_Persist_ReadBackKeyIsNotUsedHere(t *testing.T) { // Once the period ended and the buckets are idle and full, the key is gone clock.advance(25 * time.Hour) - b.Prune(time.Hour) + b.Prune() require.Equal(t, 0, b.Size()) require.Equal(t, Usage{}, b.Usage("user:u_1")) }) @@ -984,6 +1128,7 @@ const ( benchIdleKeys = 400_000 benchPublisherKeys = 50_000 benchPromotedKeys = 5_000 + benchFloodKeys = 500_000 ) var ( @@ -1058,3 +1203,33 @@ func BenchmarkService_Memory(b *testing.B) { runtime.KeepAlive(s) } } + +func BenchmarkService_MemoryFlood(b *testing.B) { + // A flood of one-shot keys (one request each, as from new IPs): what the buckets hold right + // after it, and after a prune a little later + for i := 0; i < b.N; i++ { + var base, flooded, pruned runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&base) + clock := &testClock{t: time.Date(2026, 10, 25, 12, 0, 0, 0, time.UTC)} + s := newService(&Config{}, clock.now) + for j := 0; j < benchFloodKeys; j++ { + key := fmt.Sprintf("ip:10.%d.%d.%d", j>>16, (j>>8)&255, j&255) + s.Touch(key) + s.Allow(key, benchRequestRule, 1) + } + runtime.GC() + runtime.ReadMemStats(&flooded) + keys := s.Size() + clock.advance(2 * settledIdle) + s.Prune() + runtime.GC() + runtime.ReadMemStats(&pruned) + b.ReportMetric(float64(keys), "keys") + b.ReportMetric(float64(flooded.HeapAlloc-base.HeapAlloc)/float64(benchFloodKeys), "B/key") + b.ReportMetric(float64(flooded.HeapAlloc-base.HeapAlloc)/1e6, "MB-flooded") + b.ReportMetric(float64(s.Size()), "keys-pruned") + b.ReportMetric(float64(int64(pruned.HeapAlloc)-int64(base.HeapAlloc))/1e6, "MB-pruned") + runtime.KeepAlive(s) + } +} diff --git a/limit/types.go b/limit/types.go index f89af1d4..757bec57 100644 --- a/limit/types.go +++ b/limit/types.go @@ -48,7 +48,7 @@ type Decision struct { RetryAfter time.Duration } -// Usage is a key's daily counts in its current period +// Usage is a key's counts in its current period: the daily counts, and this node's bandwidth type Usage struct { counts [numKinds]int64 } @@ -75,10 +75,10 @@ type counterShard struct { // bucketEntry is one key's token buckets type bucketEntry struct { buckets [numKinds]bucket - counts [numKinds]int64 // Units used in the current period of the kinds without a counter (bandwidth, ...), memory only - periodStart int64 // The period of counts (see periodStart) - seen int64 // Last use, unix nanoseconds - full int64 // When every bucket is full again, unix nanoseconds + bandwidth int64 // Bytes used in the current period, memory only; the only kind without a counter that is tallied + periodStart int64 // The period of bandwidth (see periodStart) + seen int64 // Last use, unix nanoseconds + full int64 // When every bucket is full again, unix nanoseconds } // bucketShard is one independently locked part of the token buckets @@ -145,7 +145,7 @@ func (l *limiter) AllowN(n int64) bool { return l.s.Allow(l.key, l.rule, n).OK } -// Value returns the key's daily count of the kind (0 for a kind without one) +// Value returns the key's count of the kind in its current period (see Usage) func (l *limiter) Value() int64 { return l.s.Usage(l.key).Count(l.rule.Kind) } diff --git a/limit/types_test.go b/limit/types_test.go index ec0817ab..63466cad 100644 --- a/limit/types_test.go +++ b/limit/types_test.go @@ -30,10 +30,16 @@ func TestUsage_Count(t *testing.T) { Burst: 100, Refill: time.Second, }, 42).OK) + require.True(t, s.Allow("k", Rule{ + Kind: KindRequests, + Burst: 100, + Refill: time.Second, + }, 1).OK) u := s.Usage("k") require.Equal(t, int64(3), u.Count(KindMessages)) require.Equal(t, int64(42), u.Count(KindBandwidth)) // In memory only, but counted require.Equal(t, int64(0), u.Count(KindEmails)) + require.Equal(t, int64(0), u.Count(KindRequests)) // Not tallied, so that it does not hold the key (see Prune) } func TestLimiter_Value_Bandwidth(t *testing.T) { diff --git a/server/server_manager.go b/server/server_manager.go index e55ae33e..39d82078 100644 --- a/server/server_manager.go +++ b/server/server_manager.go @@ -115,7 +115,7 @@ func (s *Server) pruneVisitors() { log. Tag(tagManager). Timing(func() { - pruned = s.limiter.Prune(visitorExpungeAfter) // Visitors are per request; only their keys' buckets and counts stay + pruned = s.limiter.Prune() // Visitors are per request; only their keys' buckets and counts stay, until they are settled }). Field("pruned_keys", pruned). Debug("Finished deleting stale visitor keys") diff --git a/server/server_test.go b/server/server_test.go index cb901196..136b75a0 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -3446,9 +3446,8 @@ func TestServer_SubscriberRateLimiting_VisitorExpiration(t *testing.T) { require.Equal(t, int64(0), publisher.Stats().Messages) require.Equal(t, int64(1), s.topics["upAAAAAAAAAAAA"].rateVisitor.Stats().Messages) - // Expire visitor: its subscriber has been gone for a day, and its key is pruned after a day without use + // Expire visitor: its subscriber has been gone for a day ageRateVisitor(s.topics["upAAAAAAAAAAAA"], 25*time.Hour) - s.limiter.Prune(-time.Hour) // Everything is idle for longer than -1h require.True(t, s.topics["upAAAAAAAAAAAA"].rateVisitorStaleNoLock()) // Publish message again, observe that rateVisitor is not used anymore and is reset diff --git a/server/visitor.go b/server/visitor.go index fdfa58bb..b0e82628 100644 --- a/server/visitor.go +++ b/server/visitor.go @@ -16,10 +16,8 @@ const ( // oneDay is an approximation of a day as a time.Duration oneDay = 24 * time.Hour - // visitorExpungeAfter is how long a limiter key may go unused before its buckets and counts are - // forgotten (see Server.pruneVisitors), and how long a topic keeps a rate visitor whose subscriber - // is gone. This number has to be very high to prevent e-mail abuse, but it doesn't really affect - // the other limits anyway, since they are replenished faster (typically). + // visitorExpungeAfter is how long a topic keeps a rate visitor whose subscriber is gone. The + // limiter forgets a key on its own, once forgetting it changes nothing (see limit.Service.Prune). visitorExpungeAfter = oneDay // promoteRate is the divisor of the daily message cap (cap / promoteRate, i.e. 10%) from which diff --git a/server/visitor_test.go b/server/visitor_test.go index a89a6c4b..a77b17d4 100644 --- a/server/visitor_test.go +++ b/server/visitor_test.go @@ -638,8 +638,8 @@ func TestVisitor_TierWithoutEmailsAdmitsNone(t *testing.T) { func TestVisitor_RateVisitorLivenessIsTheSubscribers(t *testing.T) { // A topic keeps its rate visitor for as long as the subscriber is around (its keepalive), not - // for as long as its limiter key is in use: the limiter forgetting the key must not drop it, - // and publishers charging the key must not keep it + // for as long as its limiter key is in use: publishers charging the key must not keep it. The + // limiter forgetting the key cannot drop it either, since the topic never asks the limiter. c := newTestConfig(t, "") c.VisitorSubscriberRateLimiting = true s := newTestServer(t, c) @@ -647,7 +647,6 @@ func TestVisitor_RateVisitorLivenessIsTheSubscribers(t *testing.T) { topics, err := s.topicsFromIDs(v, "mytopic") require.Nil(t, err) s.setRateVisitors(httptest.NewRequest("GET", "/mytopic/json", nil), v, topics) - s.limiter.Prune(0) // The key is gone, the subscriber is not require.NotNil(t, topics[0].RateVisitor()) ageRateVisitor(topics[0], 2*visitorExpungeAfter) // The subscriber is gone require.True(t, s.limiter.Allow(v.key, v.messageRule(), 1).OK) // A publisher charges the key