| 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 |
amadan