| 1 | package applog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "math/rand/v2" |
| 8 | "slices" |
| 9 | "sync" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/release" |
| 14 | ) |
| 15 | |
| 16 | // seedWindow writes chunks for two boxes across three hours. |
| 17 | func seedWindow(t *testing.T, s release.Store) { |
| 18 | t.Helper() |
| 19 | ctx := context.Background() |
| 20 | write := func(box, hour string, seq uint64, recs []Record) { |
| 21 | t.Helper() |
| 22 | body, err := EncodeChunk(recs) |
| 23 | if err != nil { |
| 24 | t.Fatal(err) |
| 25 | } |
| 26 | if _, err := s.Put(ctx, must(t)(ChunkKey("bde", "ledgerpay", box, hour, seq, len(body))), body, ""); err != nil { |
| 27 | t.Fatal(err) |
| 28 | } |
| 29 | } |
| 30 | ts := func(h, m int) time.Time { return time.Date(2026, 8, 10, h, m, 0, 0, time.UTC) } |
| 31 | write("box1", "2026-08-10-10", 100, []Record{ |
| 32 | {TS: ts(10, 5), Box: "box1", Seq: 100, Stream: StreamApp, Host: "h1", Msg: "old line"}, |
| 33 | }) |
| 34 | write("box1", "2026-08-10-11", 200, []Record{ |
| 35 | {TS: ts(11, 10), Box: "box1", Seq: 200, Stream: StreamApp, Host: "h1", Msg: "b1 app"}, |
| 36 | {TS: ts(11, 30), Box: "box1", Seq: 201, Stream: StreamPlatform, Level: LevelError, Host: "h1", Msg: "wake failed: x"}, |
| 37 | }) |
| 38 | write("box2", "2026-08-10-11", 500, []Record{ |
| 39 | {TS: ts(11, 20), Box: "box2", Seq: 500, Stream: StreamApp, Host: "h2", Msg: "b2 app"}, |
| 40 | }) |
| 41 | } |
| 42 | |
| 43 | func TestReadWindowMergesAndClipsHours(t *testing.T) { |
| 44 | s := release.NewDirStore(t.TempDir()) |
| 45 | seedWindow(t, s) |
| 46 | ctx := context.Background() |
| 47 | |
| 48 | // Window covering hour 11 only: the hour-10 chunk is never even listed. |
| 49 | f := Filter{Since: time.Date(2026, 8, 10, 11, 0, 0, 0, time.UTC)} |
| 50 | recs, trimmed, err := ReadWindow(ctx, s, "bde", "ledgerpay", f, 1000) |
| 51 | if err != nil || trimmed { |
| 52 | t.Fatalf("read: %v trimmed=%v", err, trimmed) |
| 53 | } |
| 54 | want := []string{"b1 app", "b2 app", "wake failed: x"} // (ts, box, seq) order |
| 55 | if len(recs) != len(want) { |
| 56 | t.Fatalf("got %d records: %+v", len(recs), recs) |
| 57 | } |
| 58 | for i, m := range want { |
| 59 | if recs[i].Msg != m { |
| 60 | t.Fatalf("order[%d] = %q, want %q", i, recs[i].Msg, m) |
| 61 | } |
| 62 | } |
| 63 | |
| 64 | // Stream filter composes. |
| 65 | recs, _, err = ReadWindow(ctx, s, "bde", "ledgerpay", Filter{Stream: StreamPlatform}, 1000) |
| 66 | if err != nil || len(recs) != 1 || recs[0].Level != LevelError { |
| 67 | t.Fatalf("platform filter: %v %+v", err, recs) |
| 68 | } |
| 69 | |
| 70 | // Unbounded window sees all three hours. |
| 71 | recs, _, err = ReadWindow(ctx, s, "bde", "ledgerpay", Filter{}, 1000) |
| 72 | if err != nil || len(recs) != 4 { |
| 73 | t.Fatalf("unbounded: %v, %d records", err, len(recs)) |
| 74 | } |
| 75 | |
| 76 | // collectCap keeps the TAIL and says it trimmed. |
| 77 | recs, trimmed, err = ReadWindow(ctx, s, "bde", "ledgerpay", Filter{}, 2) |
| 78 | if err != nil || !trimmed || len(recs) != 2 { |
| 79 | t.Fatalf("capped: %v trimmed=%v n=%d", err, trimmed, len(recs)) |
| 80 | } |
| 81 | if recs[len(recs)-1].Msg != "wake failed: x" { |
| 82 | t.Fatalf("cap dropped the tail: %+v", recs) |
| 83 | } |
| 84 | |
| 85 | // An app with no logs at all is an empty answer, not an error. |
| 86 | recs, _, err = ReadWindow(ctx, s, "bde", "ghost", Filter{}, 1000) |
| 87 | if err != nil || len(recs) != 0 { |
| 88 | t.Fatalf("empty app: %v %+v", err, recs) |
| 89 | } |
| 90 | } |
| 91 | |
| 92 | func TestHourIntersects(t *testing.T) { |
| 93 | since := time.Date(2026, 8, 10, 11, 0, 0, 0, time.UTC) |
| 94 | for hour, want := range map[string]bool{ |
| 95 | "2026-08-10-10": false, // ends exactly at since — nothing at/after since lives there |
| 96 | "2026-08-10-11": true, |
| 97 | "2026-08-10-12": true, |
| 98 | "not-an-hour": false, |
| 99 | } { |
| 100 | if got := hourIntersects(hour, since, time.Time{}); got != want { |
| 101 | t.Errorf("hourIntersects(%q, since) = %v, want %v", hour, got, want) |
| 102 | } |
| 103 | } |
| 104 | until := time.Date(2026, 8, 10, 11, 30, 0, 0, time.UTC) |
| 105 | if hourIntersects("2026-08-10-12", time.Time{}, until) { |
| 106 | t.Error("an hour starting after until intersects") |
| 107 | } |
| 108 | if !hourIntersects("2026-08-10-11", since, until) { |
| 109 | t.Error("the window's own hour should intersect") |
| 110 | } |
| 111 | } |
| 112 | |
| 113 | func TestReadChunksWalksAnyFamily(t *testing.T) { |
| 114 | ctx := context.Background() |
| 115 | s := release.NewDirStore(t.TempDir()) |
| 116 | type ev struct { |
| 117 | TS time.Time `json:"ts"` |
| 118 | Name string `json:"name"` |
| 119 | } |
| 120 | now := time.Date(2026, 9, 7, 10, 30, 0, 0, time.UTC) |
| 121 | body, _ := EncodeJSONL([]ev{{TS: now, Name: "boom"}, {TS: now.Add(time.Minute), Name: "skip"}}) |
| 122 | key := must(t)(FamilyChunkKey(FamilyErrors, "bab", "web", "box1", HourOf(now), 1, len(body))) |
| 123 | if _, err := s.Put(ctx, key, body, ""); err != nil { |
| 124 | t.Fatal(err) |
| 125 | } |
| 126 | perBox, trimmed, err := ReadChunks[ev](ctx, s, must(t)(FamilyPrefix(FamilyErrors, "bab", "web")), |
| 127 | now.Add(-time.Hour), now.Add(time.Hour), func(e ev) bool { return e.Name != "skip" }, 100) |
| 128 | if err != nil || trimmed { |
| 129 | t.Fatalf("ReadChunks: %v trimmed=%v", err, trimmed) |
| 130 | } |
| 131 | if len(perBox) != 1 || len(perBox[0]) != 1 || perBox[0][0].Name != "boom" { |
| 132 | t.Fatalf("perBox = %+v", perBox) |
| 133 | } |
| 134 | // An hour outside the window is never listed. |
| 135 | perBox, _, err = ReadChunks[ev](ctx, s, must(t)(FamilyPrefix(FamilyErrors, "bab", "web")), |
| 136 | now.Add(2*time.Hour), time.Time{}, func(ev) bool { return true }, 100) |
| 137 | if err != nil || len(perBox) != 1 || len(perBox[0]) != 0 { |
| 138 | t.Fatalf("out-of-window read = %+v, %v", perBox, err) |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | // slowStore delays every Get by a random few milliseconds, so fetches |
| 143 | // finish out of order, counts how many are in flight at once, and can fail |
| 144 | // one key. |
| 145 | type slowStore struct { |
| 146 | release.Store |
| 147 | mu sync.Mutex |
| 148 | inFlight int |
| 149 | peak int |
| 150 | fail string |
| 151 | } |
| 152 | |
| 153 | func (s *slowStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 154 | s.mu.Lock() |
| 155 | s.inFlight++ |
| 156 | if s.inFlight > s.peak { |
| 157 | s.peak = s.inFlight |
| 158 | } |
| 159 | s.mu.Unlock() |
| 160 | defer func() { s.mu.Lock(); s.inFlight--; s.mu.Unlock() }() |
| 161 | time.Sleep(time.Duration(rand.IntN(4)) * time.Millisecond) |
| 162 | if key == s.fail { |
| 163 | return nil, "", errors.New("bucket had a bad day") |
| 164 | } |
| 165 | return s.Store.Get(ctx, key) |
| 166 | } |
| 167 | |
| 168 | // TestReadChunksConcurrentIsSerialOrder: fetched in parallel and finishing |
| 169 | // in any order, the answer is still exactly the serial walk's — each box's |
| 170 | // records in hour then seq order, the cap keeping the tail — with no more |
| 171 | // than readConcurrency GETs in flight, and one failed chunk fails the read. |
| 172 | func TestReadChunksConcurrentIsSerialOrder(t *testing.T) { |
| 173 | ctx := context.Background() |
| 174 | dir := release.NewDirStore(t.TempDir()) |
| 175 | base := time.Date(2026, 9, 24, 0, 0, 0, 0, time.UTC) |
| 176 | want := map[string][]string{} |
| 177 | var someKey string |
| 178 | for _, box := range []string{"box1", "box2", "box3"} { |
| 179 | for h := 0; h < 5; h++ { |
| 180 | hour := base.Add(time.Duration(h) * time.Hour) |
| 181 | for c := 0; c < 8; c++ { |
| 182 | seq := uint64(1000 + h*100 + c) |
| 183 | msg := fmt.Sprintf("%s h%d c%d", box, h, c) |
| 184 | body, err := EncodeChunk([]Record{{TS: hour.Add(time.Duration(c) * time.Minute), Box: box, Seq: seq, Msg: msg}}) |
| 185 | if err != nil { |
| 186 | t.Fatal(err) |
| 187 | } |
| 188 | key := must(t)(ChunkKey("bde", "ledgerpay", box, HourOf(hour), seq, len(body))) |
| 189 | if _, err := dir.Put(ctx, key, body, ""); err != nil { |
| 190 | t.Fatal(err) |
| 191 | } |
| 192 | want[box] = append(want[box], msg) |
| 193 | someKey = key |
| 194 | } |
| 195 | } |
| 196 | } |
| 197 | s := &slowStore{Store: dir} |
| 198 | all := func(Record) bool { return true } |
| 199 | perBox, trimmed, err := ReadChunks[Record](ctx, s, must(t)(Prefix("bde", "ledgerpay")), time.Time{}, time.Time{}, all, 1000) |
| 200 | if err != nil || trimmed || len(perBox) != 3 { |
| 201 | t.Fatalf("read: %v trimmed=%v boxes=%d", err, trimmed, len(perBox)) |
| 202 | } |
| 203 | for i, box := range []string{"box1", "box2", "box3"} { |
| 204 | var got []string |
| 205 | for _, r := range perBox[i] { |
| 206 | got = append(got, r.Msg) |
| 207 | } |
| 208 | if !slices.Equal(got, want[box]) { |
| 209 | t.Fatalf("%s order:\n got %v\nwant %v", box, got, want[box]) |
| 210 | } |
| 211 | } |
| 212 | if s.peak < 2 || s.peak > readConcurrency { |
| 213 | t.Fatalf("peak in-flight GETs = %d, want 2..%d", s.peak, readConcurrency) |
| 214 | } |
| 215 | |
| 216 | perBox, trimmed, err = ReadChunks[Record](ctx, s, must(t)(Prefix("bde", "ledgerpay")), time.Time{}, time.Time{}, all, 5) |
| 217 | if err != nil || !trimmed || len(perBox[0]) != 5 || perBox[0][4].Msg != "box1 h4 c7" || perBox[0][0].Msg != "box1 h4 c3" { |
| 218 | t.Fatalf("capped read kept %+v (err %v, trimmed %v)", perBox[0], err, trimmed) |
| 219 | } |
| 220 | |
| 221 | s.fail = someKey |
| 222 | if _, _, err := ReadChunks[Record](ctx, s, must(t)(Prefix("bde", "ledgerpay")), time.Time{}, time.Time{}, all, 1000); err == nil { |
| 223 | t.Fatal("a failed chunk must fail the read, never return a partial answer") |
| 224 | } |
| 225 | } |
| 226 | |
amadan