amadan

carlos / platform Public

apperr: coalescing sink and errors/ chunk read (spec §3.2)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 735b5afe1b4164dce06c88bce91bafc12b2f028e parent 4098a24
4 files changed, +423 −0
  • internal/apperr/read.go +122 −0
  • internal/apperr/read_test.go +51 −0
  • internal/apperr/sink.go +153 −0
  • internal/apperr/sink_test.go +97 −0
diff --git a/internal/apperr/read.go b/internal/apperr/read.go
new file mode 100644
index 000000000..d17608b7e
--- /dev/null
+++ b/internal/apperr/read.go
@@ -0,0 +1,122 @@
+package apperr
+
+import (
+ "context"
+ "sort"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Prefix is the app's errors/ subtree (spec §3.2): a sibling family of
+// logs/ with the same key shape below it.
+func Prefix(acct, app string) string { return applog.FamilyPrefix(applog.FamilyErrors, acct, app) }
+
+// BoxPrefix is one box's subtree within an app's errors.
+func BoxPrefix(acct, app, box string) string {
+ return applog.FamilyBoxPrefix(applog.FamilyErrors, acct, app, box)
+}
+
+// ChunkKey names one flushed errors chunk; see applog.ChunkKey.
+func ChunkKey(acct, app, box, hour string, seq uint64, size int) string {
+ return applog.FamilyChunkKey(applog.FamilyErrors, acct, app, box, hour, seq, size)
+}
+
+// HourOf is applog.HourOf, re-exported so shippers need one import.
+func HourOf(t time.Time) string { return applog.HourOf(t) }
+
+// EncodeChunk renders records as gzipped JSONL.
+func EncodeChunk(recs []Record) ([]byte, error) { return applog.EncodeJSONL(recs) }
+
+// DecodeChunk parses a chunk, skipping malformed lines.
+func DecodeChunk(b []byte) ([]Record, error) { return applog.DecodeJSONL[Record](b) }
+
+// Filter is the read-side match. A coalesced record belongs to a window
+// when [First, Last] overlaps [Since, Until] and is counted whole (spec
+// §3.2: a window edge is wrong by at most one flush interval).
+type Filter struct {
+ Since, Until time.Time
+ Host string
+ Source string
+ Release string
+ FP string
+}
+
+// Match reports whether r survives f.
+func (f Filter) Match(r Record) bool {
+ last, first := r.Last, r.First
+ if last.IsZero() {
+ last = r.TS
+ }
+ if first.IsZero() {
+ first = r.TS
+ }
+ if !f.Since.IsZero() && last.Before(f.Since) {
+ return false
+ }
+ if !f.Until.IsZero() && first.After(f.Until) {
+ return false
+ }
+ if f.Host != "" && r.Host != f.Host {
+ return false
+ }
+ if f.Source != "" && r.Source != f.Source {
+ return false
+ }
+ if f.Release != "" && r.Release != f.Release {
+ return false
+ }
+ if f.FP != "" && r.FP != f.FP {
+ return false
+ }
+ return true
+}
+
+// ReadWindow assembles one app's errors/ records across boxes: every hour
+// intersecting the window, filtered through f, merged by (Last, Box, Seq).
+// collectCap bounds memory; past it the oldest go and the bool says so.
+func ReadWindow(ctx context.Context, s release.Store, acct, app string, f Filter, collectCap int) ([]Record, bool, error) {
+ // A record's chunk sits under the hour of its FIRST occurrence (TS),
+ // but its Last can be much later — a group that started before the
+ // window opened and is still being added to at read time. ReadChunks
+ // prunes hour directories at listing time using the SAME since/until
+ // it is given, so listing with the exact f.Since would skip that
+ // chunk's hour entirely. Widen the LISTING window by one hour behind
+ // f.Since; f.Match (which tests [First, Last] against the true
+ // window) still does the exact filtering once the chunk is read.
+ listSince := f.Since
+ if !listSince.IsZero() {
+ listSince = listSince.Add(-time.Hour)
+ }
+ perBox, trimmed, err := applog.ReadChunks[Record](ctx, s, Prefix(acct, app), listSince, f.Until, f.Match, collectCap)
+ if err != nil {
+ return nil, false, err
+ }
+ merged := MergeRecords(perBox...)
+ if over := len(merged) - collectCap; over > 0 {
+ merged = merged[over:]
+ trimmed = true
+ }
+ return merged, trimmed, nil
+}
+
+// MergeRecords folds per-source slices into one timeline ordered by
+// (Last, Box, Seq), oldest first.
+func MergeRecords(slices ...[]Record) []Record {
+ var out []Record
+ for _, s := range slices {
+ out = append(out, s...)
+ }
+ sort.SliceStable(out, func(i, j int) bool {
+ a, b := out[i], out[j]
+ if !a.Last.Equal(b.Last) {
+ return a.Last.Before(b.Last)
+ }
+ if a.Box != b.Box {
+ return a.Box < b.Box
+ }
+ return a.Seq < b.Seq
+ })
+ return out
+}
diff --git a/internal/apperr/read_test.go b/internal/apperr/read_test.go
new file mode 100644
index 000000000..7a911432b
--- /dev/null
+++ b/internal/apperr/read_test.go
@@ -0,0 +1,51 @@
+package apperr
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestReadWindowMergesAndFilters(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ now := time.Date(2026, 9, 7, 10, 30, 0, 0, time.UTC)
+ put := func(box string, recs []Record) {
+ body, err := EncodeChunk(recs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := s.Put(ctx, ChunkKey("bab", "web", box, HourOf(recs[0].TS), recs[0].Seq, len(body)), body, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ put("box1", []Record{
+ {TS: now, Box: "box1", Host: "h1", Source: SourceEdge, Name: "a", FP: "fa", Count: 3, First: now.Add(-4 * time.Minute), Last: now, Seq: 1},
+ {TS: now, Box: "box1", Host: "h1", Source: SourceServer, Name: "b", FP: "fb", Count: 1, First: now, Last: now, Seq: 2},
+ })
+ put("box2", []Record{
+ {TS: now.Add(-3 * time.Hour), Box: "box2", Host: "h2", Source: SourceEdge, Name: "old", FP: "fo", Count: 1, First: now.Add(-3 * time.Hour), Last: now.Add(-3 * time.Hour), Seq: 1},
+ })
+ recs, trimmed, err := ReadWindow(ctx, s, "bab", "web", Filter{Since: now.Add(-time.Hour)}, 100)
+ if err != nil || trimmed {
+ t.Fatalf("read: %v %v", err, trimmed)
+ }
+ if len(recs) != 2 || recs[0].FP != "fa" || recs[1].FP != "fb" {
+ t.Fatalf("window = %+v", recs)
+ }
+ // A record whose [First, Last] overlaps the window is in, whole.
+ recs, _, _ = ReadWindow(ctx, s, "bab", "web", Filter{Since: now.Add(-2 * time.Minute), Until: now.Add(-time.Minute)}, 100)
+ if len(recs) != 1 || recs[0].FP != "fa" || recs[0].Count != 3 {
+ t.Fatalf("overlap = %+v", recs)
+ }
+ recs, _, _ = ReadWindow(ctx, s, "bab", "web", Filter{Since: now.Add(-time.Hour), Source: SourceServer}, 100)
+ if len(recs) != 1 || recs[0].FP != "fb" {
+ t.Fatalf("source filter = %+v", recs)
+ }
+ recs, _, _ = ReadWindow(ctx, s, "bab", "web", Filter{Since: now.Add(-4 * time.Hour), Host: "h2"}, 100)
+ if len(recs) != 1 || recs[0].FP != "fo" {
+ t.Fatalf("host filter = %+v", recs)
+ }
+}
diff --git a/internal/apperr/sink.go b/internal/apperr/sink.go
new file mode 100644
index 000000000..a02c282d1
--- /dev/null
+++ b/internal/apperr/sink.go
@@ -0,0 +1,153 @@
+package apperr
+
+import (
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+)
+
+// 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.
+const MaxGroupsPerAppHour = 1000
+
+// OverflowFP names the one row the excess beyond the cap folds into. Its
+// Count is occurrences, and the tab labels it as such.
+const OverflowFP = "_overflow"
+
+// Sink coalesces errors between ship-loop flushes: per app, one record per
+// (fp, host, release) carrying count/first/last and the first sample. An
+// error storm is the moment this must work — ten thousand identical panics
+// in a minute become one record. Safe for concurrent use; the edge, the
+// activator and the ship loop share one Sink per agent process.
+type Sink struct {
+ box string
+ now func() time.Time
+
+ mu sync.Mutex
+ apps map[applog.AppKey]*appBuf
+ seq uint64
+}
+
+type groupKey struct{ fp, host, release string }
+
+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
+}
+
+// NewSink returns a Sink stamping records with box and timestamps from now.
+// seq starts at UnixMilli for the reason applog.NewSink gives: chunk keys
+// embed the first record's seq, and a restart must sort after last boot.
+func NewSink(box string, now func() time.Time) *Sink {
+ return &Sink{box: box, now: now, apps: make(map[applog.AppKey]*appBuf), seq: uint64(now().UnixMilli())}
+}
+
+// Add buffers one occurrence for (acct, app). r is sanitised here (§3.1),
+// fingerprinted when it carries no FP, and folded into its group.
+func (s *Sink) Add(acct, app string, r Record) {
+ now := s.now().UTC()
+ r.Name = Sanitize(r.Name)
+ r.Path = SanitizePath(r.Path)
+ if r.Path == "/" && r.Status == 0 && r.Source != SourceEdge {
+ r.Path = ""
+ }
+ if len(r.Stack) > MaxStack {
+ r.Stack = r.Stack[:MaxStack]
+ }
+ r.Stack = Sanitize(r.Stack)
+ if len(r.Fields) > 0 {
+ f := make(map[string]string, len(r.Fields))
+ for k, v := range r.Fields {
+ f[k] = Sanitize(v)
+ }
+ r.Fields = f
+ }
+ if r.FP == "" {
+ r.FP = Fingerprint(r.Source, r.Kind, r.Name, r.Stack)
+ r.FPV = FPVersion
+ }
+ if len(r.FP) > 64 {
+ r.FP = r.FP[:64]
+ }
+
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ b := s.apps[applog.AppKey{Acct: acct, App: app}]
+ if b == nil {
+ b = &appBuf{groups: make(map[groupKey]*Record)}
+ s.apps[applog.AppKey{Acct: acct, App: app}] = b
+ }
+ if h := applog.HourOf(now); h != b.hour {
+ b.hour, b.admitted = h, 0
+ }
+ k := groupKey{r.FP, r.Host, r.Release}
+ if g := b.groups[k]; g != nil {
+ g.Count++
+ g.Last = now
+ return
+ }
+ if b.admitted >= MaxGroupsPerAppHour {
+ if b.overflow == nil {
+ o := r
+ o.FP = OverflowFP
+ o.TS, o.First, o.Last, o.Count = now, now, now, 0
+ o.Box = s.box
+ s.seq++
+ o.Seq = s.seq
+ b.overflow = &o
+ }
+ b.overflow.Count++
+ b.overflow.Last = now
+ return
+ }
+ b.admitted++
+ s.seq++
+ r.TS, r.First, r.Last, r.Count = now, now, now, 1
+ r.Box = s.box
+ r.Seq = s.seq
+ b.groups[k] = &r
+ b.order = append(b.order, k)
+}
+
+// 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.
+func (s *Sink) Drain() map[applog.AppKey][]Record {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ out := make(map[applog.AppKey][]Record)
+ for k, b := range s.apps {
+ if len(b.order) == 0 && b.overflow == nil {
+ continue
+ }
+ recs := make([]Record, 0, len(b.order)+1)
+ for _, gk := range b.order {
+ recs = append(recs, *b.groups[gk])
+ }
+ if b.overflow != nil {
+ recs = append(recs, *b.overflow)
+ }
+ out[k] = recs
+ b.groups = make(map[groupKey]*Record)
+ b.order = nil
+ b.overflow = nil
+ }
+ return out
+}
+
+// Known lists every app this sink has ever buffered for — the prune
+// worklist, as applog.Sink.Known is for logs.
+func (s *Sink) Known() []applog.AppKey {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ out := make([]applog.AppKey, 0, len(s.apps))
+ for k := range s.apps {
+ out = append(out, k)
+ }
+ return out
+}
diff --git a/internal/apperr/sink_test.go b/internal/apperr/sink_test.go
new file mode 100644
index 000000000..bffaf385c
--- /dev/null
+++ b/internal/apperr/sink_test.go
@@ -0,0 +1,97 @@
+package apperr
+
+import (
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/applog"
+)
+
+func TestSinkCoalescesByFingerprintHostRelease(t *testing.T) {
+ now := time.Date(2026, 9, 7, 10, 0, 0, 0, time.UTC)
+ clock := now
+ s := NewSink("box1", func() time.Time { return clock })
+ for i := 0; i < 10000; i++ {
+ clock = now.Add(time.Duration(i) * time.Millisecond)
+ s.Add("bab", "web", Record{Host: "h1", Source: SourceEdge, Kind: KindHTTP, Name: "500 GET /x", Status: 500, Release: "v1"})
+ }
+ s.Add("bab", "web", Record{Host: "h2", Source: SourceEdge, Kind: KindHTTP, Name: "500 GET /x", Status: 500, Release: "v1"})
+ out := s.Drain()
+ recs := out[applog.AppKey{Acct: "bab", App: "web"}]
+ if len(recs) != 2 {
+ t.Fatalf("records = %d, want 2 (one per host)", len(recs))
+ }
+ var h1 Record
+ for _, r := range recs {
+ if r.Host == "h1" {
+ h1 = r
+ }
+ }
+ if h1.Count != 10000 || !h1.First.Equal(now) || !h1.Last.Equal(now.Add(9999*time.Millisecond)) {
+ t.Fatalf("h1 = count %d first %v last %v", h1.Count, h1.First, h1.Last)
+ }
+ if h1.FP == "" || h1.FPV != FPVersion || h1.Box != "box1" || h1.Seq == 0 {
+ t.Fatalf("stamping: %+v", h1)
+ }
+ if len(s.Drain()) != 0 {
+ t.Fatal("second drain must be empty")
+ }
+}
+
+func TestSinkCapsGroupsPerAppPerHour(t *testing.T) {
+ now := time.Date(2026, 9, 7, 10, 0, 0, 0, time.UTC)
+ s := NewSink("box1", func() time.Time { return now })
+ for i := 0; i < MaxGroupsPerAppHour+5; i++ {
+ s.Add("bab", "web", Record{Source: SourceEdge, Kind: KindHTTP, Name: fmt.Sprintf("500 GET /p%d-x", i), FP: fmt.Sprintf("fp%d", i)})
+ }
+ s.Add("bab", "other", Record{Source: SourceEdge, Kind: KindHTTP, Name: "unaffected", FP: "z"})
+ out := s.Drain()
+ web := out[applog.AppKey{Acct: "bab", App: "web"}]
+ if len(web) != MaxGroupsPerAppHour+1 {
+ t.Fatalf("web records = %d, want cap+1 (the overflow row)", len(web))
+ }
+ var over *Record
+ for i := range web {
+ if web[i].FP == OverflowFP {
+ over = &web[i]
+ }
+ }
+ if over == nil || over.Count != 5 || over.Name != "500 GET /p1000-x" {
+ t.Fatalf("overflow = %+v", over)
+ }
+ if len(out[applog.AppKey{Acct: "bab", App: "other"}]) != 1 {
+ t.Fatal("a neighbour app must not be suppressed")
+ }
+ // The cap counts per hour: the next hour admits new groups again.
+ now = now.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)
+ }
+}
+
+func TestSinkSanitisesAndFingerprintsOnAdd(t *testing.T) {
+ now := time.Date(2026, 9, 7, 10, 0, 0, 0, time.UTC)
+ s := NewSink("box1", func() time.Time { return now })
+ long := make([]byte, MaxStack+100)
+ for i := range long {
+ long[i] = 'x'
+ }
+ s.Add("bab", "web", Record{Source: SourceServer, Kind: "RuntimeError", Name: "mail to paul@tito.io bounced",
+ Path: "/reset/8f3a9c1e2b4d6f708a9b0c1d2e3f4a5b?t=1", Stack: string(long),
+ Fields: map[string]string{"user": "paul@tito.io", "k": "v"}})
+ r := s.Drain()[applog.AppKey{Acct: "bab", App: "web"}][0]
+ if r.Name != "mail to <email> bounced" || r.Path != "/reset/:token" || r.Fields["user"] != "<email>" || r.Fields["k"] != "v" {
+ t.Fatalf("sanitised = %+v", r)
+ }
+ if len(r.Stack) != MaxStack {
+ t.Fatalf("stack len %d, want %d", len(r.Stack), MaxStack)
+ }
+ if r.FP != Fingerprint(SourceServer, "RuntimeError", "mail to <email> bounced", r.Stack) || r.FPV != FPVersion {
+ t.Fatalf("fp = %q", r.FP)
+ }
+ if r.Count != 1 || !r.First.Equal(now) || !r.Last.Equal(now) || !r.TS.Equal(now) {
+ t.Fatalf("stamps = %+v", r)
+ }
+}