amadan

carlos / platform Public

applog: chunk families and a generic JSONL codec, logs/ unchanged

Prefix/ChunkKey/EncodeChunk/ReadWindow keep their signatures and bytes;
FamilyPrefix, FamilyChunkKey, EncodeJSONL/DecodeJSONL and ReadChunks are
the seams the errors/ family (spec 2026-09-07 §3.2) builds on.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 9aa34f10d68e60fe0596384a6650b8eacc918cbf parent d37a072
4 files changed, +166 −35
  • internal/applog/chunk.go +57 −16
  • internal/applog/chunk_test.go +45 −0
  • internal/applog/read.go +35 −19
  • internal/applog/read_test.go +29 −0
diff --git a/internal/applog/chunk.go b/internal/applog/chunk.go
index 6603f8ab0..ef3a8b56b 100644
--- a/internal/applog/chunk.go
+++ b/internal/applog/chunk.go
@@ -33,21 +33,55 @@ func HourBefore(hour string, cutoff time.Time) bool {
return t.Add(time.Hour).Sub(cutoff.UTC()) <= 0 && HourOf(cutoff) != hour
}
-// Prefix is the app's whole log subtree — under release.AppPrefix on
-// purpose (spec §3): logs are the app's data, so the owner gate maps onto
-// the prefix and app trash/purge sweeps them for free.
-func Prefix(acct, app string) string { return release.AppPrefix(acct, app) + "logs/" }
+// Family names one chunk family under an app — the path segment after
+// apps/<sqid>/<app>/. logs/ is the original; errors/ and analytics/ are
+// the siblings the errors-and-analytics spec adds (§3.2). The key shape
+// below the family segment is identical for every family, which is what
+// lets one prune loop, one store-door grant and one reader serve all three.
+type Family string
-// BoxPrefix is one box's subtree within an app's logs — the unit of
+const (
+ FamilyLogs Family = "logs"
+ FamilyErrors Family = "errors"
+ FamilyAnalytics Family = "analytics"
+)
+
+// FamilyPrefix is the app's whole subtree for one family — under
+// release.AppPrefix on purpose (spec §3): it is the app's data, so the
+// owner gate maps onto the prefix and app trash/purge sweeps it for free.
+func FamilyPrefix(fam Family, acct, app string) string {
+ return release.AppPrefix(acct, app) + string(fam) + "/"
+}
+
+// FamilyBoxPrefix is one box's subtree within a family — the unit of
// write-side ownership: each box prunes only what it wrote.
-func BoxPrefix(acct, app, box string) string { return Prefix(acct, app) + box + "/" }
+func FamilyBoxPrefix(fam Family, acct, app, box string) string {
+ return FamilyPrefix(fam, acct, app) + box + "/"
+}
+
+// FamilyHourPrefix is one UTC hour within a box's subtree.
+func FamilyHourPrefix(fam Family, acct, app, box, hour string) string {
+ return FamilyBoxPrefix(fam, acct, app, box) + hour + "/"
+}
-// HourPrefix is one UTC hour within a box's subtree.
+// FamilyChunkKey names one flushed chunk in any family. See ChunkKey for
+// why seq and size live in the key.
+func FamilyChunkKey(fam Family, acct, app, box, hour string, seq uint64, size int) string {
+ return fmt.Sprintf("%s%016d-%d.jsonl.gz", FamilyHourPrefix(fam, acct, app, box, hour), seq, size)
+}
+
+// Prefix is the app's whole log subtree: FamilyPrefix for logs.
+func Prefix(acct, app string) string { return FamilyPrefix(FamilyLogs, acct, app) }
+
+// BoxPrefix is one box's subtree within an app's logs.
+func BoxPrefix(acct, app, box string) string { return FamilyBoxPrefix(FamilyLogs, acct, app, box) }
+
+// HourPrefix is one UTC hour within a box's log subtree.
func HourPrefix(acct, app, box, hour string) string {
- return BoxPrefix(acct, app, box) + hour + "/"
+ return FamilyHourPrefix(FamilyLogs, acct, app, box, hour)
}
-// ChunkKey names one flushed chunk: seq is the first record's per-box
+// ChunkKey names one flushed log chunk: seq is the first record's per-box
// sequence number, zero-padded (16 digits — seq is milliseconds-based, see
// NewSink) so lexical listing order is numeric order, and size is the
// object's own byte length. Size lives IN the key on purpose:
@@ -56,7 +90,7 @@ func HourPrefix(acct, app, box, hour string) string {
// costs every implementation. Chunks are written once and never appended
// or rewritten (spec §3).
func ChunkKey(acct, app, box, hour string, seq uint64, size int) string {
- return fmt.Sprintf("%s%016d-%d.jsonl.gz", HourPrefix(acct, app, box, hour), seq, size)
+ return FamilyChunkKey(FamilyLogs, acct, app, box, hour, seq, size)
}
// ChunkKeySize recovers the byte size a ChunkKey carries, false for any key
@@ -82,8 +116,9 @@ func ChunkKeySize(key string) (int64, bool) {
return n, true
}
-// EncodeChunk renders records as gzipped JSONL, one record per line.
-func EncodeChunk(recs []Record) ([]byte, error) {
+// EncodeJSONL renders any record shape as gzipped JSONL, one per line —
+// the chunk codec every family shares.
+func EncodeJSONL[T any](recs []T) ([]byte, error) {
var buf bytes.Buffer
zw := gzip.NewWriter(&buf)
enc := json.NewEncoder(zw)
@@ -98,20 +133,20 @@ func EncodeChunk(recs []Record) ([]byte, error) {
return buf.Bytes(), nil
}
-// DecodeChunk parses a chunk, skipping malformed lines rather than failing
+// DecodeJSONL parses a chunk, skipping malformed lines rather than failing
// the whole object — one corrupt line must not cost the member the rest of
// the hour. Corrupt gzip is an error: that is a broken object, not a line.
-func DecodeChunk(b []byte) ([]Record, error) {
+func DecodeJSONL[T any](b []byte) ([]T, error) {
zr, err := gzip.NewReader(bytes.NewReader(b))
if err != nil {
return nil, err
}
defer zr.Close()
- var out []Record
+ var out []T
sc := bufio.NewScanner(zr)
sc.Buffer(nil, 1<<20) // lines are capped well below this at the tee
for sc.Scan() {
- var r Record
+ var r T
if json.Unmarshal(sc.Bytes(), &r) != nil {
continue
}
@@ -122,3 +157,9 @@ func DecodeChunk(b []byte) ([]Record, error) {
}
return out, nil
}
+
+// EncodeChunk renders log records: EncodeJSONL over Record.
+func EncodeChunk(recs []Record) ([]byte, error) { return EncodeJSONL(recs) }
+
+// DecodeChunk parses a log chunk: DecodeJSONL over Record.
+func DecodeChunk(b []byte) ([]Record, error) { return DecodeJSONL[Record](b) }
diff --git a/internal/applog/chunk_test.go b/internal/applog/chunk_test.go
index 936aa983e..b257ebe4b 100644
--- a/internal/applog/chunk_test.go
+++ b/internal/applog/chunk_test.go
@@ -118,3 +118,48 @@ func TestHourBefore(t *testing.T) {
}
}
}
+
+func TestFamilyKeysMirrorLogs(t *testing.T) {
+ if got, want := FamilyPrefix(FamilyLogs, "bab", "web"), Prefix("bab", "web"); got != want {
+ t.Fatalf("FamilyPrefix(logs) = %q, Prefix = %q", got, want)
+ }
+ if got, want := FamilyChunkKey(FamilyLogs, "bab", "web", "box1", "2026-09-07-10", 42, 7), ChunkKey("bab", "web", "box1", "2026-09-07-10", 42, 7); got != want {
+ t.Fatalf("FamilyChunkKey(logs) = %q, ChunkKey = %q", got, want)
+ }
+ got := FamilyChunkKey(FamilyErrors, "bab", "web", "box1", "2026-09-07-10", 42, 7)
+ want := "apps/bab/web/errors/box1/2026-09-07-10/0000000000000042-7.jsonl.gz"
+ if got != want {
+ t.Fatalf("errors key = %q, want %q", got, want)
+ }
+ if n, ok := ChunkKeySize(got); !ok || n != 7 {
+ t.Fatalf("ChunkKeySize on an errors key = %d,%v", n, ok)
+ }
+}
+
+func TestEncodeJSONLRoundTripsAnyShape(t *testing.T) {
+ type rec struct {
+ A string `json:"a"`
+ N int `json:"n,omitempty"`
+ }
+ b, err := EncodeJSONL([]rec{{A: "x", N: 1}, {A: "y"}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ out, err := DecodeJSONL[rec](b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(out) != 2 || out[0].A != "x" || out[0].N != 1 || out[1].A != "y" || out[1].N != 0 {
+ t.Fatalf("round trip = %+v", out)
+ }
+ // EncodeChunk must be byte-identical to EncodeJSONL over the same
+ // records: the logs family's wire shape does not move.
+ recs := []Record{{TS: time.Date(2026, 9, 7, 10, 0, 0, 0, time.UTC), Msg: "hi", Seq: 1}}
+ a, _ := EncodeChunk(recs)
+ c, _ := EncodeJSONL(recs)
+ da, _ := DecodeChunk(a)
+ dc, _ := DecodeJSONL[Record](c)
+ if len(da) != 1 || len(dc) != 1 || da[0].Msg != dc[0].Msg || !da[0].TS.Equal(dc[0].TS) {
+ t.Fatalf("EncodeChunk and EncodeJSONL disagree: %+v vs %+v", da, dc)
+ }
+}
diff --git a/internal/applog/read.go b/internal/applog/read.go
index d0c41ea83..1cacfb07b 100644
--- a/internal/applog/read.go
+++ b/internal/applog/read.go
@@ -8,31 +8,29 @@ import (
"github.com/carlosframework/platform/internal/release"
)
-// ReadWindow assembles one app's merged timeline from the bucket: every
-// box's chunks whose hour directories intersect [f.Since, f.Until], decoded,
-// filtered through f, merged in (ts, box, seq) order. Hour pruning happens
-// at the PREFIX level — an hour outside the window is never listed, which
-// is what makes a 1h read over 7d of history cheap (spec §3).
-//
-// collectCap bounds memory against a pathological window: past it the
-// OLDEST records go (the tail is what a log reader came for), and the
-// second return says so. Store errors fail the read — a partial answer
-// that looks whole is the one thing this must never produce.
-func ReadWindow(ctx context.Context, s release.Store, acct, app string, f Filter, collectCap int) ([]Record, bool, error) {
- boxDirs, err := s.ListDirs(ctx, Prefix(acct, app))
+// ReadChunks walks every box's hour directories under prefix (a
+// FamilyPrefix) whose hours intersect [since, until], decodes each chunk as
+// T, keeps what passes keep, and returns one slice per box in listing
+// order. Hour pruning happens at the PREFIX level — an hour outside the
+// window is never listed, which is what makes a 1h read over 7d of history
+// cheap (spec §3). collectCap bounds memory per box: past it the OLDEST
+// records go and trimmed says so. Store errors fail the read — a partial
+// answer that looks whole is the one thing this must never produce.
+func ReadChunks[T any](ctx context.Context, s release.Store, prefix string, since, until time.Time, keep func(T) bool, collectCap int) ([][]T, bool, error) {
+ boxDirs, err := s.ListDirs(ctx, prefix)
if err != nil {
return nil, false, err
}
trimmed := false
- perBox := make([][]Record, 0, len(boxDirs))
+ perBox := make([][]T, 0, len(boxDirs))
for _, boxDir := range boxDirs {
hourDirs, err := s.ListDirs(ctx, boxDir)
if err != nil {
return nil, false, err
}
- var recs []Record
+ var recs []T
for _, hourDir := range hourDirs { // lexically sorted = chronological
- if !hourIntersects(hourNameOfDir(hourDir), f.Since, f.Until) {
+ if !hourIntersects(HourNameOfDir(hourDir), since, until) {
continue
}
keys, err := s.List(ctx, hourDir)
@@ -44,12 +42,12 @@ func ReadWindow(ctx context.Context, s release.Store, acct, app string, f Filter
if err != nil {
return nil, false, err
}
- chunk, err := DecodeChunk(body)
+ chunk, err := DecodeJSONL[T](body)
if err != nil {
return nil, false, err
}
for _, r := range chunk {
- if f.Match(r) {
+ if keep(r) {
recs = append(recs, r)
}
}
@@ -61,6 +59,24 @@ func ReadWindow(ctx context.Context, s release.Store, acct, app string, f Filter
}
perBox = append(perBox, recs)
}
+ return perBox, trimmed, nil
+}
+
+// ReadWindow assembles one app's merged timeline from the bucket: every
+// box's chunks whose hour directories intersect [f.Since, f.Until], decoded,
+// filtered through f, merged in (ts, box, seq) order. Hour pruning happens
+// at the PREFIX level — an hour outside the window is never listed, which
+// is what makes a 1h read over 7d of history cheap (spec §3).
+//
+// collectCap bounds memory against a pathological window: past it the
+// OLDEST records go (the tail is what a log reader came for), and the
+// second return says so. Store errors fail the read — a partial answer
+// that looks whole is the one thing this must never produce.
+func ReadWindow(ctx context.Context, s release.Store, acct, app string, f Filter, collectCap int) ([]Record, bool, error) {
+ perBox, trimmed, err := ReadChunks[Record](ctx, s, Prefix(acct, app), f.Since, 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:]
@@ -87,8 +103,8 @@ func hourIntersects(hour string, since, until time.Time) bool {
return true
}
-// hourNameOfDir extracts the final path segment of a "/"-terminated prefix.
-func hourNameOfDir(dir string) string {
+// HourNameOfDir extracts the final path segment of a "/"-terminated prefix.
+func HourNameOfDir(dir string) string {
trimmed := strings.TrimSuffix(dir, "/")
if i := strings.LastIndexByte(trimmed, '/'); i >= 0 {
return trimmed[i+1:]
diff --git a/internal/applog/read_test.go b/internal/applog/read_test.go
index 0d53482af..28f83b181 100644
--- a/internal/applog/read_test.go
+++ b/internal/applog/read_test.go
@@ -104,3 +104,32 @@ func TestHourIntersects(t *testing.T) {
t.Error("the window's own hour should intersect")
}
}
+
+func TestReadChunksWalksAnyFamily(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ type ev struct {
+ TS time.Time `json:"ts"`
+ Name string `json:"name"`
+ }
+ now := time.Date(2026, 9, 7, 10, 30, 0, 0, time.UTC)
+ body, _ := EncodeJSONL([]ev{{TS: now, Name: "boom"}, {TS: now.Add(time.Minute), Name: "skip"}})
+ key := FamilyChunkKey(FamilyErrors, "bab", "web", "box1", HourOf(now), 1, len(body))
+ if _, err := s.Put(ctx, key, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ perBox, trimmed, err := ReadChunks[ev](ctx, s, FamilyPrefix(FamilyErrors, "bab", "web"),
+ now.Add(-time.Hour), now.Add(time.Hour), func(e ev) bool { return e.Name != "skip" }, 100)
+ if err != nil || trimmed {
+ t.Fatalf("ReadChunks: %v trimmed=%v", err, trimmed)
+ }
+ if len(perBox) != 1 || len(perBox[0]) != 1 || perBox[0][0].Name != "boom" {
+ t.Fatalf("perBox = %+v", perBox)
+ }
+ // An hour outside the window is never listed.
+ perBox, _, err = ReadChunks[ev](ctx, s, FamilyPrefix(FamilyErrors, "bab", "web"),
+ now.Add(2*time.Hour), time.Time{}, func(ev) bool { return true }, 100)
+ if err != nil || len(perBox) != 1 || len(perBox[0]) != 0 {
+ t.Fatalf("out-of-window read = %+v, %v", perBox, err)
+ }
+}