| 1 | package applog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "strings" |
| 6 | "sync" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/release" |
| 10 | ) |
| 11 | |
| 12 | // readConcurrency bounds one read's in-flight bucket calls: listings in |
| 13 | // the first half, chunk GETs in the second. A day of an app's logs is |
| 14 | // hundreds of chunks, and fetched one at a time that was the whole of the |
| 15 | // Errors tab's ~5 s (2026-09-24, titogo); sixteen at once brings it to |
| 16 | // roughly the slowest few round trips, and bounds what one read can ask of |
| 17 | // the bucket and hold decoded in memory. |
| 18 | const readConcurrency = 16 |
| 19 | |
| 20 | // ReadChunks walks every box's hour directories under prefix (a |
| 21 | // FamilyPrefix) whose hours intersect [since, until], decodes each chunk as |
| 22 | // T, keeps what passes keep, and returns one slice per box in listing |
| 23 | // order. Hour pruning happens at the PREFIX level — an hour outside the |
| 24 | // window is never listed, which is what makes a 1h read over 7d of history |
| 25 | // cheap (spec §3). collectCap bounds memory per box: past it the OLDEST |
| 26 | // records go and trimmed says so. Store errors fail the read — a partial |
| 27 | // answer that looks whole is the one thing this must never produce. |
| 28 | // |
| 29 | // Listings and GETs run readConcurrency at a time, but the answer is |
| 30 | // assembled strictly in listing order, so it is exactly what a serial walk |
| 31 | // would return. Chunks are fetched at most readConcurrency ahead of the one |
| 32 | // being assembled, so memory stays bounded by collectCap plus that window, |
| 33 | // as it was by collectCap plus one chunk. keep runs on the fetching |
| 34 | // goroutines and must be safe for concurrent use (a Filter's Match is). |
| 35 | 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) { |
| 36 | ctx, cancel := context.WithCancel(ctx) |
| 37 | defer cancel() |
| 38 | boxDirs, err := s.ListDirs(ctx, prefix) |
| 39 | if err != nil { |
| 40 | return nil, false, err |
| 41 | } |
| 42 | hourDirs := make([][]string, len(boxDirs)) |
| 43 | if err := forEachBounded(ctx, len(boxDirs), func(i int) error { |
| 44 | dirs, err := s.ListDirs(ctx, boxDirs[i]) |
| 45 | if err != nil { |
| 46 | return err |
| 47 | } |
| 48 | for _, d := range dirs { // lexically sorted = chronological |
| 49 | if hourIntersects(HourNameOfDir(d), since, until) { |
| 50 | hourDirs[i] = append(hourDirs[i], d) |
| 51 | } |
| 52 | } |
| 53 | return nil |
| 54 | }); err != nil { |
| 55 | return nil, false, err |
| 56 | } |
| 57 | type hour struct { |
| 58 | box int |
| 59 | dir string |
| 60 | } |
| 61 | var hours []hour |
| 62 | for b, dirs := range hourDirs { |
| 63 | for _, d := range dirs { |
| 64 | hours = append(hours, hour{b, d}) |
| 65 | } |
| 66 | } |
| 67 | hourKeys := make([][]string, len(hours)) |
| 68 | if err := forEachBounded(ctx, len(hours), func(i int) error { |
| 69 | keys, err := s.List(ctx, hours[i].dir) |
| 70 | hourKeys[i] = keys // lexically sorted = seq order |
| 71 | return err |
| 72 | }); err != nil { |
| 73 | return nil, false, err |
| 74 | } |
| 75 | type chunk struct { |
| 76 | box int |
| 77 | key string |
| 78 | } |
| 79 | var chunks []chunk |
| 80 | for i, keys := range hourKeys { |
| 81 | for _, k := range keys { |
| 82 | chunks = append(chunks, chunk{hours[i].box, k}) |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | // Fetch ahead, assemble in order. A slot's channel is buffered, so a |
| 87 | // fetch whose result is never read (the read failed first) still ends; |
| 88 | // the feeder stops at cancel. |
| 89 | type fetched struct { |
| 90 | recs []T |
| 91 | err error |
| 92 | } |
| 93 | slots := make([]chan fetched, len(chunks)) |
| 94 | for i := range slots { |
| 95 | slots[i] = make(chan fetched, 1) |
| 96 | } |
| 97 | tokens := make(chan struct{}, readConcurrency) |
| 98 | go func() { |
| 99 | for i, c := range chunks { |
| 100 | select { |
| 101 | case tokens <- struct{}{}: |
| 102 | case <-ctx.Done(): |
| 103 | return |
| 104 | } |
| 105 | go func() { |
| 106 | body, _, err := s.Get(ctx, c.key) |
| 107 | if err != nil { |
| 108 | slots[i] <- fetched{err: err} |
| 109 | return |
| 110 | } |
| 111 | all, err := DecodeJSONL[T](body) |
| 112 | if err != nil { |
| 113 | slots[i] <- fetched{err: err} |
| 114 | return |
| 115 | } |
| 116 | var kept []T |
| 117 | for _, r := range all { |
| 118 | if keep(r) { |
| 119 | kept = append(kept, r) |
| 120 | } |
| 121 | } |
| 122 | slots[i] <- fetched{recs: kept} |
| 123 | }() |
| 124 | } |
| 125 | }() |
| 126 | trimmed := false |
| 127 | perBox := make([][]T, len(boxDirs)) |
| 128 | for i, c := range chunks { |
| 129 | var f fetched |
| 130 | select { |
| 131 | case f = <-slots[i]: |
| 132 | case <-ctx.Done(): |
| 133 | return nil, false, ctx.Err() |
| 134 | } |
| 135 | <-tokens |
| 136 | if f.err != nil { |
| 137 | return nil, false, f.err |
| 138 | } |
| 139 | recs := append(perBox[c.box], f.recs...) |
| 140 | if over := len(recs) - collectCap; over > 0 { |
| 141 | recs = recs[over:] |
| 142 | trimmed = true |
| 143 | } |
| 144 | perBox[c.box] = recs |
| 145 | } |
| 146 | return perBox, trimmed, nil |
| 147 | } |
| 148 | |
| 149 | // forEachBounded runs fn(0..n-1) readConcurrency at a time and returns the |
| 150 | // first error, cancelling nothing itself: fn's own ctx is the caller's, and |
| 151 | // ReadChunks cancels it on return. |
| 152 | func forEachBounded(ctx context.Context, n int, fn func(i int) error) error { |
| 153 | var ( |
| 154 | wg sync.WaitGroup |
| 155 | mu sync.Mutex |
| 156 | firstErr error |
| 157 | ) |
| 158 | tokens := make(chan struct{}, readConcurrency) |
| 159 | for i := 0; i < n; i++ { |
| 160 | mu.Lock() |
| 161 | failed := firstErr != nil |
| 162 | mu.Unlock() |
| 163 | if failed { |
| 164 | break // the read has already failed; start nothing more |
| 165 | } |
| 166 | select { |
| 167 | case tokens <- struct{}{}: |
| 168 | case <-ctx.Done(): |
| 169 | wg.Wait() |
| 170 | return ctx.Err() |
| 171 | } |
| 172 | wg.Add(1) |
| 173 | go func() { |
| 174 | defer func() { <-tokens; wg.Done() }() |
| 175 | if err := fn(i); err != nil { |
| 176 | mu.Lock() |
| 177 | if firstErr == nil { |
| 178 | firstErr = err |
| 179 | } |
| 180 | mu.Unlock() |
| 181 | } |
| 182 | }() |
| 183 | } |
| 184 | wg.Wait() |
| 185 | return firstErr |
| 186 | } |
| 187 | |
| 188 | // ReadWindow assembles one app's merged timeline from the bucket: every |
| 189 | // box's chunks whose hour directories intersect [f.Since, f.Until], decoded, |
| 190 | // filtered through f, merged in (ts, box, seq) order. Hour pruning happens |
| 191 | // at the PREFIX level — an hour outside the window is never listed, which |
| 192 | // is what makes a 1h read over 7d of history cheap (spec §3). |
| 193 | // |
| 194 | // collectCap bounds memory against a pathological window: past it the |
| 195 | // OLDEST records go (the tail is what a log reader came for), and the |
| 196 | // second return says so. Store errors fail the read — a partial answer |
| 197 | // that looks whole is the one thing this must never produce. |
| 198 | func ReadWindow(ctx context.Context, s release.Store, acct, app string, f Filter, collectCap int) ([]Record, bool, error) { |
| 199 | prefix, err := Prefix(acct, app) |
| 200 | if err != nil { |
| 201 | return nil, false, err |
| 202 | } |
| 203 | perBox, trimmed, err := ReadChunks[Record](ctx, s, prefix, f.Since, f.Until, f.Match, collectCap) |
| 204 | if err != nil { |
| 205 | return nil, false, err |
| 206 | } |
| 207 | merged := MergeRecords(perBox...) |
| 208 | if over := len(merged) - collectCap; over > 0 { |
| 209 | merged = merged[over:] |
| 210 | trimmed = true |
| 211 | } |
| 212 | return merged, trimmed, nil |
| 213 | } |
| 214 | |
| 215 | // hourIntersects reports whether the hour directory named hour overlaps |
| 216 | // [since, until] (zero = unbounded). Unparsable names never match — an |
| 217 | // alien directory is not silently read as data. |
| 218 | func hourIntersects(hour string, since, until time.Time) bool { |
| 219 | start, err := time.ParseInLocation(hourFormat, hour, time.UTC) |
| 220 | if err != nil { |
| 221 | return false |
| 222 | } |
| 223 | end := start.Add(time.Hour) |
| 224 | if !since.IsZero() && !end.After(since) { |
| 225 | return false |
| 226 | } |
| 227 | if !until.IsZero() && start.After(until) { |
| 228 | return false |
| 229 | } |
| 230 | return true |
| 231 | } |
| 232 | |
| 233 | // HourNameOfDir extracts the final path segment of a "/"-terminated prefix. |
| 234 | func HourNameOfDir(dir string) string { |
| 235 | trimmed := strings.TrimSuffix(dir, "/") |
| 236 | if i := strings.LastIndexByte(trimmed, '/'); i >= 0 { |
| 237 | return trimmed[i+1:] |
| 238 | } |
| 239 | return trimmed |
| 240 | } |
| 241 | |
amadan