amadan

carlos / platform Public

apperr: the hourly cap counts distinct keys, not re-admissions after each drain

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 355aaa3a8f8da7d1091d2e8de213cbcaca19ef73 parent 21fde9b
2 files changed, +65 −9
  • internal/apperr/sink.go +14 −9
  • internal/apperr/sink_test.go +51 −0
diff --git a/internal/apperr/sink.go b/internal/apperr/sink.go
index a02c282d1..55e80a85e 100644
--- a/internal/apperr/sink.go
+++ b/internal/apperr/sink.go
@@ -8,8 +8,11 @@ import (
)
// MaxGroupsPerAppHour caps coalesced records per app per UTC hour, counting
-// (fp, host, release) combinations (spec §3.2). Per app, not per box: one
-// tenant's storm must not suppress a neighbour's.
+// distinct (fp, host, release) combinations admitted within the hour, across
+// every drain (spec §3.2) — not admissions since the last drain. A key once
+// admitted this hour stays admitted; only genuinely new keys count against
+// the cap. Per app, not per box: one tenant's storm must not suppress a
+// neighbour's.
const MaxGroupsPerAppHour = 1000
// OverflowFP names the one row the excess beyond the cap folds into. Its
@@ -36,8 +39,8 @@ type appBuf struct {
groups map[groupKey]*Record
order []groupKey // insertion order, for a deterministic drain
overflow *Record
- hour string // the UTC hour `admitted` counts for
- admitted int // distinct groups admitted this hour, cap counter
+ hour string // the UTC hour `seen` counts for
+ seen map[groupKey]bool // keys admitted this hour, across every drain — the cap counter; Drain must not touch it
}
// NewSink returns a Sink stamping records with box and timestamps from now.
@@ -83,7 +86,7 @@ func (s *Sink) Add(acct, app string, r Record) {
s.apps[applog.AppKey{Acct: acct, App: app}] = b
}
if h := applog.HourOf(now); h != b.hour {
- b.hour, b.admitted = h, 0
+ b.hour, b.seen = h, make(map[groupKey]bool)
}
k := groupKey{r.FP, r.Host, r.Release}
if g := b.groups[k]; g != nil {
@@ -91,7 +94,7 @@ func (s *Sink) Add(acct, app string, r Record) {
g.Last = now
return
}
- if b.admitted >= MaxGroupsPerAppHour {
+ if !b.seen[k] && len(b.seen) >= MaxGroupsPerAppHour {
if b.overflow == nil {
o := r
o.FP = OverflowFP
@@ -105,7 +108,7 @@ func (s *Sink) Add(acct, app string, r Record) {
b.overflow.Last = now
return
}
- b.admitted++
+ b.seen[k] = true
s.seq++
r.TS, r.First, r.Last, r.Count = now, now, now, 1
r.Box = s.box
@@ -115,8 +118,10 @@ func (s *Sink) Add(acct, app string, r Record) {
}
// Drain returns everything buffered, keyed by app, and resets the groups
-// (the per-hour cap counter and seq persist — they are the hour's and the
-// process's, not the batch's). Empty apps are omitted.
+// (the per-hour cap counter `seen` and seq persist — they are the hour's and
+// the process's, not the batch's; a key seen in an earlier drain this hour
+// stays admitted without counting against the cap again). Empty apps are
+// omitted.
func (s *Sink) Drain() map[applog.AppKey][]Record {
s.mu.Lock()
defer s.mu.Unlock()
diff --git a/internal/apperr/sink_test.go b/internal/apperr/sink_test.go
index bffaf385c..60dc22930 100644
--- a/internal/apperr/sink_test.go
+++ b/internal/apperr/sink_test.go
@@ -95,3 +95,54 @@ func TestSinkSanitisesAndFingerprintsOnAdd(t *testing.T) {
t.Fatalf("stamps = %+v", r)
}
}
+
+// TestSinkCapCountsDistinctKeysAcrossDrains replays the ship loop: a 10 s
+// drain, a handful of distinct errors recurring in every tick. Six distinct
+// errors must never overflow, however many ticks the hour has.
+func TestSinkCapCountsDistinctKeysAcrossDrains(t *testing.T) {
+ base := time.Date(2026, 9, 7, 10, 0, 0, 0, time.UTC)
+ clock := base
+ s := NewSink("box1", func() time.Time { return clock })
+ for tick := 0; tick < 360; tick++ {
+ clock = base.Add(time.Duration(tick*10) * time.Second)
+ for i := 0; i < 6; i++ {
+ s.Add("bab", "web", Record{Source: SourceEdge, Kind: KindHTTP, Name: fmt.Sprintf("500 GET /p%d", i), Host: "h1", Release: "v1"})
+ }
+ for _, recs := range s.Drain() {
+ for _, r := range recs {
+ if r.FP == OverflowFP {
+ t.Fatalf("overflow at tick %d with only 6 distinct errors", tick)
+ }
+ }
+ if len(recs) != 6 {
+ t.Fatalf("tick %d drained %d records, want 6", tick, len(recs))
+ }
+ }
+ }
+ // The cap still binds on genuinely distinct keys within the hour, and
+ // a key already seen this hour is admitted even after the cap is hit.
+ clock = base.Add(30 * time.Minute)
+ for i := 0; i < MaxGroupsPerAppHour+5; i++ {
+ s.Add("bab", "web", Record{Source: SourceEdge, Kind: KindHTTP, Name: "x", FP: fmt.Sprintf("d%d", i)})
+ }
+ s.Add("bab", "web", Record{Source: SourceEdge, Kind: KindHTTP, Name: "500 GET /p0", Host: "h1", Release: "v1"}) // seen this hour: admitted, not overflow
+ out := s.Drain()[applog.AppKey{Acct: "bab", App: "web"}]
+ var over, p0 int
+ for _, r := range out {
+ if r.FP == OverflowFP {
+ over = r.Count
+ }
+ if r.Name == "500 GET /p0" && r.FP != OverflowFP {
+ p0++
+ }
+ }
+ if over != 11 || p0 != 1 { // 6 recurring + 994 new = 1000; 11 new beyond it overflow; p0 was seen
+ t.Fatalf("overflow count = %d (want 11), p0 admitted = %d (want 1)", over, p0)
+ }
+ // A new hour starts the count over.
+ clock = base.Add(time.Hour)
+ s.Add("bab", "web", Record{Source: SourceEdge, Kind: KindHTTP, Name: "fresh", FP: "fresh"})
+ if recs := s.Drain()[applog.AppKey{Acct: "bab", App: "web"}]; len(recs) != 1 || recs[0].FP != "fresh" {
+ t.Fatalf("next hour = %+v", recs)
+ }
+}