amadan

carlos / platform Public

applog: fetch chunks 16 at a time, assemble in serial order

The Errors tab took ~5 s of its ~7 s reading a day of titogo's log chunks
one GET at a time (to find error-level lines; there were none). ReadChunks
now lists box and hour directories and fetches chunks readConcurrency (16)
at a time. The answer is assembled strictly in listing order, so it is
exactly what the serial walk returned; fetches run at most 16 ahead of the
chunk being assembled, so memory stays bounded; a failed listing or chunk
still fails the whole read.

Every reader of the chunk families gets it: the Logs tab and `carlos logs`,
the Errors (and, once #417 lands, Availability) reads.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Bill Horsman pushed by bill@keymail.dev d4d218614b13204f839c1fdc342cd373ddf76f2a parent b5b0a53
2 files changed, +238 −25
  • internal/applog/read.go +148 −25
  • internal/applog/read_test.go +90 −0

CI passed — run details

diff --git a/internal/applog/read.go b/internal/applog/read.go
index 1cacfb07b..355dea8a0 100644
--- a/internal/applog/read.go
+++ b/internal/applog/read.go
@@ -3,11 +3,20 @@ package applog
import (
"context"
"strings"
+ "sync"
"time"
"github.com/carlosframework/platform/internal/release"
)
+// readConcurrency bounds one read's in-flight bucket calls: listings in
+// the first half, chunk GETs in the second. A day of an app's logs is
+// hundreds of chunks, and fetched one at a time that was the whole of the
+// Errors tab's ~5 s (2026-09-24, titogo); sixteen at once brings it to
+// roughly the slowest few round trips, and bounds what one read can ask of
+// the bucket and hold decoded in memory.
+const readConcurrency = 16
+
// 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
@@ -16,52 +25,166 @@ import (
// 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.
+//
+// Listings and GETs run readConcurrency at a time, but the answer is
+// assembled strictly in listing order, so it is exactly what a serial walk
+// would return. Chunks are fetched at most readConcurrency ahead of the one
+// being assembled, so memory stays bounded by collectCap plus that window,
+// as it was by collectCap plus one chunk. keep runs on the fetching
+// goroutines and must be safe for concurrent use (a Filter's Match is).
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) {
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
boxDirs, err := s.ListDirs(ctx, prefix)
if err != nil {
return nil, false, err
}
- trimmed := false
- perBox := make([][]T, 0, len(boxDirs))
- for _, boxDir := range boxDirs {
- hourDirs, err := s.ListDirs(ctx, boxDir)
+ hourDirs := make([][]string, len(boxDirs))
+ if err := forEachBounded(ctx, len(boxDirs), func(i int) error {
+ dirs, err := s.ListDirs(ctx, boxDirs[i])
if err != nil {
- return nil, false, err
+ return err
}
- var recs []T
- for _, hourDir := range hourDirs { // lexically sorted = chronological
- if !hourIntersects(HourNameOfDir(hourDir), since, until) {
- continue
+ for _, d := range dirs { // lexically sorted = chronological
+ if hourIntersects(HourNameOfDir(d), since, until) {
+ hourDirs[i] = append(hourDirs[i], d)
}
- keys, err := s.List(ctx, hourDir)
- if err != nil {
- return nil, false, err
+ }
+ return nil
+ }); err != nil {
+ return nil, false, err
+ }
+ type hour struct {
+ box int
+ dir string
+ }
+ var hours []hour
+ for b, dirs := range hourDirs {
+ for _, d := range dirs {
+ hours = append(hours, hour{b, d})
+ }
+ }
+ hourKeys := make([][]string, len(hours))
+ if err := forEachBounded(ctx, len(hours), func(i int) error {
+ keys, err := s.List(ctx, hours[i].dir)
+ hourKeys[i] = keys // lexically sorted = seq order
+ return err
+ }); err != nil {
+ return nil, false, err
+ }
+ type chunk struct {
+ box int
+ key string
+ }
+ var chunks []chunk
+ for i, keys := range hourKeys {
+ for _, k := range keys {
+ chunks = append(chunks, chunk{hours[i].box, k})
+ }
+ }
+
+ // Fetch ahead, assemble in order. A slot's channel is buffered, so a
+ // fetch whose result is never read (the read failed first) still ends;
+ // the feeder stops at cancel.
+ type fetched struct {
+ recs []T
+ err error
+ }
+ slots := make([]chan fetched, len(chunks))
+ for i := range slots {
+ slots[i] = make(chan fetched, 1)
+ }
+ tokens := make(chan struct{}, readConcurrency)
+ go func() {
+ for i, c := range chunks {
+ select {
+ case tokens <- struct{}{}:
+ case <-ctx.Done():
+ return
}
- for _, key := range keys { // lexically sorted = seq order
- body, _, err := s.Get(ctx, key)
+ go func() {
+ body, _, err := s.Get(ctx, c.key)
if err != nil {
- return nil, false, err
+ slots[i] <- fetched{err: err}
+ return
}
- chunk, err := DecodeJSONL[T](body)
+ all, err := DecodeJSONL[T](body)
if err != nil {
- return nil, false, err
+ slots[i] <- fetched{err: err}
+ return
}
- for _, r := range chunk {
+ var kept []T
+ for _, r := range all {
if keep(r) {
- recs = append(recs, r)
+ kept = append(kept, r)
}
}
- if over := len(recs) - collectCap; over > 0 {
- recs = recs[over:]
- trimmed = true
- }
- }
+ slots[i] <- fetched{recs: kept}
+ }()
+ }
+ }()
+ trimmed := false
+ perBox := make([][]T, len(boxDirs))
+ for i, c := range chunks {
+ var f fetched
+ select {
+ case f = <-slots[i]:
+ case <-ctx.Done():
+ return nil, false, ctx.Err()
+ }
+ <-tokens
+ if f.err != nil {
+ return nil, false, f.err
+ }
+ recs := append(perBox[c.box], f.recs...)
+ if over := len(recs) - collectCap; over > 0 {
+ recs = recs[over:]
+ trimmed = true
}
- perBox = append(perBox, recs)
+ perBox[c.box] = recs
}
return perBox, trimmed, nil
}
+// forEachBounded runs fn(0..n-1) readConcurrency at a time and returns the
+// first error, cancelling nothing itself: fn's own ctx is the caller's, and
+// ReadChunks cancels it on return.
+func forEachBounded(ctx context.Context, n int, fn func(i int) error) error {
+ var (
+ wg sync.WaitGroup
+ mu sync.Mutex
+ firstErr error
+ )
+ tokens := make(chan struct{}, readConcurrency)
+ for i := 0; i < n; i++ {
+ mu.Lock()
+ failed := firstErr != nil
+ mu.Unlock()
+ if failed {
+ break // the read has already failed; start nothing more
+ }
+ select {
+ case tokens <- struct{}{}:
+ case <-ctx.Done():
+ wg.Wait()
+ return ctx.Err()
+ }
+ wg.Add(1)
+ go func() {
+ defer func() { <-tokens; wg.Done() }()
+ if err := fn(i); err != nil {
+ mu.Lock()
+ if firstErr == nil {
+ firstErr = err
+ }
+ mu.Unlock()
+ }
+ }()
+ }
+ wg.Wait()
+ return firstErr
+}
+
// 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
diff --git a/internal/applog/read_test.go b/internal/applog/read_test.go
index 28f83b181..90ef3f1cc 100644
--- a/internal/applog/read_test.go
+++ b/internal/applog/read_test.go
@@ -2,6 +2,11 @@ package applog
import (
"context"
+ "errors"
+ "fmt"
+ "math/rand/v2"
+ "slices"
+ "sync"
"testing"
"time"
@@ -133,3 +138,88 @@ func TestReadChunksWalksAnyFamily(t *testing.T) {
t.Fatalf("out-of-window read = %+v, %v", perBox, err)
}
}
+
+// slowStore delays every Get by a random few milliseconds, so fetches
+// finish out of order, counts how many are in flight at once, and can fail
+// one key.
+type slowStore struct {
+ release.Store
+ mu sync.Mutex
+ inFlight int
+ peak int
+ fail string
+}
+
+func (s *slowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ s.mu.Lock()
+ s.inFlight++
+ if s.inFlight > s.peak {
+ s.peak = s.inFlight
+ }
+ s.mu.Unlock()
+ defer func() { s.mu.Lock(); s.inFlight--; s.mu.Unlock() }()
+ time.Sleep(time.Duration(rand.IntN(4)) * time.Millisecond)
+ if key == s.fail {
+ return nil, "", errors.New("bucket had a bad day")
+ }
+ return s.Store.Get(ctx, key)
+}
+
+// TestReadChunksConcurrentIsSerialOrder: fetched in parallel and finishing
+// in any order, the answer is still exactly the serial walk's — each box's
+// records in hour then seq order, the cap keeping the tail — with no more
+// than readConcurrency GETs in flight, and one failed chunk fails the read.
+func TestReadChunksConcurrentIsSerialOrder(t *testing.T) {
+ ctx := context.Background()
+ dir := release.NewDirStore(t.TempDir())
+ base := time.Date(2026, 9, 24, 0, 0, 0, 0, time.UTC)
+ want := map[string][]string{}
+ var someKey string
+ for _, box := range []string{"box1", "box2", "box3"} {
+ for h := 0; h < 5; h++ {
+ hour := base.Add(time.Duration(h) * time.Hour)
+ for c := 0; c < 8; c++ {
+ seq := uint64(1000 + h*100 + c)
+ msg := fmt.Sprintf("%s h%d c%d", box, h, c)
+ body, err := EncodeChunk([]Record{{TS: hour.Add(time.Duration(c) * time.Minute), Box: box, Seq: seq, Msg: msg}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ key := ChunkKey("bde", "ledgerpay", box, HourOf(hour), seq, len(body))
+ if _, err := dir.Put(ctx, key, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ want[box] = append(want[box], msg)
+ someKey = key
+ }
+ }
+ }
+ s := &slowStore{Store: dir}
+ all := func(Record) bool { return true }
+ perBox, trimmed, err := ReadChunks[Record](ctx, s, Prefix("bde", "ledgerpay"), time.Time{}, time.Time{}, all, 1000)
+ if err != nil || trimmed || len(perBox) != 3 {
+ t.Fatalf("read: %v trimmed=%v boxes=%d", err, trimmed, len(perBox))
+ }
+ for i, box := range []string{"box1", "box2", "box3"} {
+ var got []string
+ for _, r := range perBox[i] {
+ got = append(got, r.Msg)
+ }
+ if !slices.Equal(got, want[box]) {
+ t.Fatalf("%s order:\n got %v\nwant %v", box, got, want[box])
+ }
+ }
+ if s.peak < 2 || s.peak > readConcurrency {
+ t.Fatalf("peak in-flight GETs = %d, want 2..%d", s.peak, readConcurrency)
+ }
+
+ perBox, trimmed, err = ReadChunks[Record](ctx, s, Prefix("bde", "ledgerpay"), time.Time{}, time.Time{}, all, 5)
+ if err != nil || !trimmed || len(perBox[0]) != 5 || perBox[0][4].Msg != "box1 h4 c7" || perBox[0][0].Msg != "box1 h4 c3" {
+ t.Fatalf("capped read kept %+v (err %v, trimmed %v)", perBox[0], err, trimmed)
+ }
+
+ s.fail = someKey
+ if _, _, err := ReadChunks[Record](ctx, s, Prefix("bde", "ledgerpay"), time.Time{}, time.Time{}, all, 1000); err == nil {
+ t.Fatal("a failed chunk must fail the read, never return a partial answer")
+ }
+}