carlos / platform Public

analytics: requeue once per delta, count malformed events, throttle salt retries

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 28de8d368ca0543fea9dc32ba5f3fc46bbefe4db parent bbe7593
3 files changed, +161 −64
  • internal/analytics/day.go +6 −0
  • internal/analytics/meter.go +62 −24
  • internal/analytics/meter_test.go +93 −40
diff --git a/internal/analytics/day.go b/internal/analytics/day.go
index 0ae9ab2fd..d016afc8f 100644
--- a/internal/analytics/day.go
+++ b/internal/analytics/day.go
@@ -61,6 +61,12 @@ type Delta struct {
Events *TopK
Dropped Dropped
Restarted bool
+
+ // requeued marks a delta that has already been through one failed
+ // flush (internal/analytics.Meter.Requeue). Not serialised: it never
+ // reaches the day file, only decides whether a second Requeue keeps
+ // or drops the delta.
+ requeued bool
}
// Prefix is the app's analytics/ subtree.
diff --git a/internal/analytics/meter.go b/internal/analytics/meter.go
index 77889c762..4455f818e 100644
--- a/internal/analytics/meter.go
+++ b/internal/analytics/meter.go
@@ -82,6 +82,11 @@ func EventName(s string) (string, bool) {
return s, true
}
+// saltRetryDelay bounds how often a failing DaySalt is retried. Visitors
+// go uncounted while it fails, but Observe must never do that I/O (or log)
+// on every request — one attempt, and one log line, per window.
+const saltRetryDelay = 30 * time.Second
+
// Meter accumulates per-app deltas between flushes. One per edge process;
// Observe sits on the serving path (one mutex, no I/O except the first
// salt read of a day).
@@ -90,30 +95,39 @@ type Meter struct {
saltDir string
now func() time.Time
- mu sync.Mutex
- cur map[AppKey]map[string]*Delta // app → date → delta being built
- pending map[AppKey][]Delta // re-queued after a failed flush (one window)
- seq int64
- saltDay string
- salt []byte
- started string // date the process started: Restarted is set on that day's first delta
+ mu sync.Mutex
+ cur map[AppKey]map[string]*Delta // app → date → delta being built
+ pending map[AppKey][]Delta // re-queued after a failed flush, each marked once
+ dropPending map[AppKey]int64 // deltas dropped by a Requeue that found an already-marked one, awaiting the next delta to land on
+ seq int64
+ saltDay string
+ salt []byte
+ saltFailDay string // date DaySalt last failed for, "" once it next succeeds
+ saltFailAt time.Time // when it failed; not retried until saltRetryDelay has passed
+ started string // date the process started: Restarted is set on the very first delta created after start
}
// NewMeter returns a meter whose salts live in saltDir (the box data dir).
func NewMeter(saltDir, box string, now func() time.Time) *Meter {
return &Meter{box: box, saltDir: saltDir, now: now, cur: map[AppKey]map[string]*Delta{},
- pending: map[AppKey][]Delta{}, seq: now().UnixMilli(), started: usage.DayOf(now())}
+ pending: map[AppKey][]Delta{}, dropPending: map[AppKey]int64{},
+ seq: now().UnixMilli(), started: usage.DayOf(now())}
}
func (m *Meter) saltFor(date string) []byte {
if m.saltDay == date && m.salt != nil {
return m.salt
}
+ if m.saltFailDay == date && m.now().Sub(m.saltFailAt) < saltRetryDelay {
+ return nil // still within the backoff window: don't retry, don't log again
+ }
s, err := DaySalt(m.saltDir, date)
if err != nil {
+ m.saltFailDay, m.saltFailAt = date, m.now()
log.Printf("analytics: day salt: %v (visitors not counted until it works)", err)
return nil
}
+ m.saltFailDay = ""
m.saltDay, m.salt = date, s
return s
}
@@ -126,9 +140,13 @@ func (m *Meter) delta(k AppKey, date string) *Delta {
}
d := days[date]
if d == nil {
+ restarted := date == m.started
d = &Delta{Date: date, Box: m.box, Hours: map[string]HourRow{}, Visitors: &HLL{},
- Pages: NewTopK(), Referers: NewTopK(), Events: NewTopK(), Restarted: date == m.started}
+ Pages: NewTopK(), Referers: NewTopK(), Events: NewTopK(), Restarted: restarted}
days[date] = d
+ if restarted {
+ m.started = "" // the note belongs to the first delta created after start, not every delta of the day
+ }
}
return d
}
@@ -163,18 +181,25 @@ func (m *Meter) Observe(o Obs) {
}
// Track counts one named event (the doors call it; nothing does in step 2).
+// An invalid name is not silently discarded: it counts against the app's
+// current-day Dropped.A.Malformed so the honesty block sees it.
func (m *Meter) Track(acct, app, name string, when time.Time) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ d := m.delta(AppKey{Acct: acct, App: app}, usage.DayOf(when))
n, ok := EventName(name)
if !ok {
+ d.Dropped.A.Malformed++
return
}
- m.mu.Lock()
- defer m.mu.Unlock()
- m.delta(AppKey{Acct: acct, App: app}, usage.DayOf(when)).Events.Add(n, 1)
+ d.Events.Add(n, 1)
}
// Drain returns everything buffered — re-queued deltas first, then the
// current ones, each stamped with a fresh box-monotonic seq — and resets.
+// A window an earlier Requeue had to drop (because it was already marked
+// re-queued when Requeue saw it again) is not silently lost: its count is
+// added to Dropped.Flush on the first delta this drain emits for that app.
func (m *Meter) Drain() map[AppKey][]Delta {
m.mu.Lock()
defer m.mu.Unlock()
@@ -191,29 +216,42 @@ func (m *Meter) Drain() map[AppKey][]Delta {
}
}
m.cur = map[AppKey]map[string]*Delta{}
+ for k, ds := range out {
+ if len(ds) == 0 {
+ continue
+ }
+ if dropped := m.dropPending[k]; dropped > 0 {
+ ds[0].Dropped.Flush += dropped
+ delete(m.dropPending, k)
+ }
+ }
if m.started != "" && usage.DayOf(m.now()) != m.started {
m.started = "" // the restart note is only for the day it happened
}
return out
}
-// Requeue holds a failed flush for the next drain. One window survives: a
-// window already sitting in pending when Requeue is called again (i.e. no
-// successful Drain emptied it in between) is dropped and its delta count
-// is added to Dropped.Flush on the newest delta of the incoming batch.
-// Seq is left untouched — a re-queued delta keeps the seq it was drained
-// with, so a replay that actually landed is a no-op at the day file.
+// Requeue holds a failed flush for the next drain. Each delta may survive
+// one failed flush: Requeue marks it (an unexported field on Delta, never
+// serialised) and holds it in pending. A delta Requeue is handed that is
+// already marked — the flush carrying it forward also failed, with no
+// successful Drain landing it in between — is dropped instead, and the
+// drop is counted per app in dropPending rather than kept, so pending
+// never grows past one window's worth of history no matter how many
+// flushes in a row fail. Seq is left untouched — a re-queued delta keeps
+// the seq it was drained with, so a replay that actually landed is a
+// no-op at the day file.
func (m *Meter) Requeue(failed map[AppKey][]Delta) {
m.mu.Lock()
defer m.mu.Unlock()
for k, ds := range failed {
- if n := len(m.pending[k]); n > 0 {
- dropped := int64(n)
- m.pending[k] = nil
- if len(ds) > 0 {
- ds[len(ds)-1].Dropped.Flush += dropped
+ for _, d := range ds {
+ if d.requeued {
+ m.dropPending[k]++
+ continue
}
+ d.requeued = true
+ m.pending[k] = append(m.pending[k], d)
}
- m.pending[k] = append(m.pending[k], ds...)
}
}
diff --git a/internal/analytics/meter_test.go b/internal/analytics/meter_test.go
index a6445bb6d..da7abf4a5 100644
--- a/internal/analytics/meter_test.go
+++ b/internal/analytics/meter_test.go
@@ -1,6 +1,7 @@
package analytics
import (
+ "path/filepath"
"testing"
"time"
)
@@ -77,6 +78,9 @@ func TestMeterCountsAndDrains(t *testing.T) {
if e := d.Events.Fold(); e["signup"] != 1 || len(e) != 1 {
t.Fatalf("events = %v", e)
}
+ if d.Dropped.A.Malformed != 1 {
+ t.Fatalf("bad event name must count Dropped.A.Malformed, not vanish: %+v", d.Dropped)
+ }
if len(m.Drain()) != 0 {
t.Fatal("second drain must be empty")
}
@@ -88,60 +92,109 @@ func TestMeterCountsAndDrains(t *testing.T) {
}
}
-// TestMeterRequeueSurvivesOnce pins down the exact requeue semantics:
-//
-// 1. Requeue(first) after a failed flush: first's delta is held in
-// pending — nothing was pending before, so nothing is dropped.
-// 2. A normal Drain (no Requeue in between, so pending got emptied by
-// the Drain call itself) returns the held delta ahead of the
-// newly-current one, each keeping its own Seq — the re-queued delta
-// keeps the seq it was drained with (so a replay that actually landed
-// is a no-op at the day file, via Day.Merge's seq gate), while the
-// current delta gets a fresh, higher seq.
-// 3. Requeue(second) holds both of second's deltas in pending. Because
-// Drain had just cleared pending, this call finds nothing already
-// there and drops nothing yet.
-// 4. A second Requeue call — this time with a freshly built delta,
-// fired with NO Drain in between — finds second's two deltas still
-// sitting in pending from step 3. That whole older window is dropped;
-// its count (2) is added to Dropped.Flush on the newest delta in the
-// incoming batch, and only the incoming batch is kept in pending.
-// 5. A Drain() with nothing newly observed returns exactly that one
-// surviving, Dropped.Flush-stamped delta: the older window is gone.
+// TestMeterRequeueSurvivesOnce exercises the production caller's actual
+// sequence — Drain, flush, Requeue on failure, repeat — where pending is
+// always empty at the moment Requeue is called (the prior Drain just
+// cleared it). The "already re-queued once" state has to live on the
+// delta itself, not on pending's emptiness, or nothing is ever dropped
+// and pending grows without bound across repeated failures. Each delta
+// may survive exactly one failed flush: a second failure for the same
+// delta drops it and counts the drop (once the delta is marked, a second
+// Requeue drops it) on the next delta that reaches the app's bucket —
+// bounded forever at two windows per app no matter how many flushes in a
+// row fail.
func TestMeterRequeueSurvivesOnce(t *testing.T) {
now := time.Date(2026, 9, 7, 9, 30, 0, 0, time.UTC)
m := NewMeter(t.TempDir(), "box1", func() time.Time { return now })
k := AppKey{Acct: "bab", App: "web"}
m.Observe(obs("GET", 200, "text/html"))
- first := m.Drain()
- m.Requeue(first) // flush failed
+ d1 := m.Drain()
+ m.Requeue(d1) // flush failed: d1's delta is marked and held
+
+ m.Observe(obs("GET", 200, "text/html"))
+ d2 := m.Drain() // = d1's delta (marked, own seq) + the new one (fresh seq)
+ if ds := d2[k]; len(ds) != 2 || ds[0].Seq >= ds[1].Seq {
+ t.Fatalf("d2 must be the requeued delta then the new one, in seq order: %+v", ds)
+ }
+ m.Requeue(d2) // failed again: d1's delta is already marked, so it drops; d2's new delta is marked and held
m.Observe(obs("GET", 200, "text/html"))
- second := m.Drain()
- ds := second[k]
- if len(ds) != 2 || ds[0].Seq >= ds[1].Seq {
- t.Fatalf("requeued delta must come back first with its own seq: %+v", ds)
+ d3 := m.Drain()[k]
+ if len(d3) != 2 {
+ t.Fatalf("d1's window must be gone, leaving d2's surviving delta plus the new one: %+v", d3)
+ }
+ if d3[0].Dropped.Flush != 1 {
+ t.Fatalf("the drop must be reported on the delta that reaches the bucket: %+v", d3)
+ }
+ if d3[0].Seq != d2[k][1].Seq {
+ t.Fatalf("the surviving delta must keep the seq it had in d2, not be re-stamped: got %d, want %d", d3[0].Seq, d2[k][1].Seq)
}
- if ds[0].Seq != first[k][0].Seq {
- t.Fatalf("requeued delta must keep its original seq: got %d, want %d", ds[0].Seq, first[k][0].Seq)
+ if d3[1].Seq <= d3[0].Seq {
+ t.Fatalf("the new delta must get a fresh, higher seq: %+v", d3)
+ }
+
+ // A third failure in a row: bounded forever at two windows per app.
+ m.Requeue(map[AppKey][]Delta{k: d3})
+ m.Observe(obs("GET", 200, "text/html"))
+ d4 := m.Drain()[k]
+ if len(d4) != 2 || d4[0].Dropped.Flush != 1 {
+ t.Fatalf("repeated failures must stay bounded at two windows, each drop reported once: %+v", d4)
+ }
+}
+
+// TestMeterTrackMalformedCountsDropped: an invalid event name must not
+// vanish silently — it counts against the app's current-day
+// Dropped.A.Malformed.
+func TestMeterTrackMalformedCountsDropped(t *testing.T) {
+ now := time.Date(2026, 9, 7, 9, 30, 0, 0, time.UTC)
+ m := NewMeter(t.TempDir(), "box1", func() time.Time { return now })
+ m.Track("bab", "web", "not a valid name!", now)
+ m.Track("bab", "web", "also/not valid", now)
+ d := m.Drain()[AppKey{Acct: "bab", App: "web"}][0]
+ if d.Dropped.A.Malformed != 2 {
+ t.Fatalf("Dropped.A.Malformed = %d, want 2", d.Dropped.A.Malformed)
}
+}
- m.Requeue(second) // failed again: pending was empty, so nothing drops yet
+// TestMeterSaltFailureIsThrottled: a persistently failing DaySalt must not
+// be retried on every request — pageviews still count, but visitors stay
+// uncounted while it fails, without blocking or panicking on the
+// serving path.
+func TestMeterSaltFailureIsThrottled(t *testing.T) {
+ now := time.Date(2026, 9, 7, 9, 30, 0, 0, time.UTC)
+ saltDir := filepath.Join(t.TempDir(), "missing", "nested") // DaySalt can't create its file here
+ m := NewMeter(saltDir, "box1", func() time.Time { return now })
+ m.Observe(obs("GET", 200, "text/html")) // salt lookup fails
+ m.Observe(obs("GET", 200, "text/html")) // within the backoff window: not retried
+ now = now.Add(31 * time.Second)
+ m.Observe(obs("GET", 200, "text/html")) // backoff elapsed: retried (and fails again, saltDir still missing)
+ d := m.Drain()[AppKey{Acct: "bab", App: "web"}][0]
+ if d.Hours["09"].PV != 3 {
+ t.Fatalf("pageviews must still count while the salt fails: %+v", d.Hours)
+ }
+ if c := d.Visitors.Count(); c != 0 {
+ t.Fatalf("visitors must stay uncounted while the salt fails, got %.2f", c)
+ }
+}
- // A further failure before any successful Drain: pending still holds
- // second's two deltas from the call above, so this Requeue drops that
- // whole older window onto the newest delta of the incoming batch.
- handBuilt := Delta{Date: "2026-09-07", Box: "box1", Seq: 999,
- Hours: map[string]HourRow{}, Visitors: &HLL{}, Pages: NewTopK(), Referers: NewTopK(), Events: NewTopK()}
- m.Requeue(map[AppKey][]Delta{k: {handBuilt}})
+// TestMeterRestartedOnlyOnFirstDelta: Restarted marks the very first
+// delta created after the meter starts, not every app-day's first
+// delta.
+func TestMeterRestartedOnlyOnFirstDelta(t *testing.T) {
+ now := time.Date(2026, 9, 7, 9, 30, 0, 0, time.UTC)
+ m := NewMeter(t.TempDir(), "box1", func() time.Time { return now })
+ m.Observe(obs("GET", 200, "text/html")) // first delta created since start: bab/web
+ o := obs("GET", 200, "text/html")
+ o.Acct, o.App = "other", "app2"
+ m.Observe(o) // a different app's first delta of the day, but not the meter's first since start
- third := m.Drain()[k]
- if len(third) != 1 || third[0].Dropped.Flush != 2 {
- t.Fatalf("requeue-while-pending must drop the older window and count both its deltas: %+v", third)
+ out := m.Drain()
+ if !out[AppKey{Acct: "bab", App: "web"}][0].Restarted {
+ t.Fatal("the very first delta created after start must carry Restarted")
}
- if third[0].Seq != handBuilt.Seq {
- t.Fatalf("surviving window must be the most recently re-queued one: %+v", third)
+ if out[AppKey{Acct: "other", App: "app2"}][0].Restarted {
+ t.Fatal("only the meter's first delta since start should carry Restarted, not every app's")
}
}