| 1 | package applog |
| 2 | |
| 3 | import ( |
| 4 | "strings" |
| 5 | "testing" |
| 6 | "time" |
| 7 | ) |
| 8 | |
| 9 | func testClock(start time.Time) (func() time.Time, func(d time.Duration)) { |
| 10 | now := start |
| 11 | return func() time.Time { return now }, func(d time.Duration) { now = now.Add(d) } |
| 12 | } |
| 13 | |
| 14 | func TestSinkStampsAndSequences(t *testing.T) { |
| 15 | now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)) |
| 16 | sk := NewSink("box1", now) |
| 17 | sk.App("bde", "ledgerpay", "ledgerpay.bde.oncarlos.com", "hello") |
| 18 | sk.Event("bde", "ledgerpay", "ledgerpay.bde.oncarlos.com", LevelError, "wake failed: seed timed out") |
| 19 | |
| 20 | got := sk.Drain() |
| 21 | if len(got) != 1 { |
| 22 | t.Fatalf("apps drained: %d, want 1", len(got)) |
| 23 | } |
| 24 | recs := got[AppKey{"bde", "ledgerpay"}] |
| 25 | if len(recs) != 2 { |
| 26 | t.Fatalf("records: %d, want 2", len(recs)) |
| 27 | } |
| 28 | if recs[0].Box != "box1" || recs[0].Stream != StreamApp || recs[0].TS.IsZero() || recs[0].Host != "ledgerpay.bde.oncarlos.com" { |
| 29 | t.Fatalf("app record not stamped: %+v", recs[0]) |
| 30 | } |
| 31 | if recs[1].Stream != StreamPlatform || recs[1].Level != LevelError { |
| 32 | t.Fatalf("event record wrong: %+v", recs[1]) |
| 33 | } |
| 34 | if recs[1].Seq != recs[0].Seq+1 { |
| 35 | t.Fatalf("seq not monotonic: %d then %d", recs[0].Seq, recs[1].Seq) |
| 36 | } |
| 37 | // Drain resets; seq does not. |
| 38 | sk.App("bde", "ledgerpay", "h", "again") |
| 39 | recs2 := sk.Drain()[AppKey{"bde", "ledgerpay"}] |
| 40 | if len(recs2) != 1 || recs2[0].Seq != recs[1].Seq+1 { |
| 41 | t.Fatalf("after drain: %+v", recs2) |
| 42 | } |
| 43 | } |
| 44 | |
| 45 | // TestSinkAppLinesCarryParsedJSON: the sink is where ParseAppLine is |
| 46 | // applied, so a structured app line arrives in the bucket with fields and |
| 47 | // an honored level rather than as opaque text. |
| 48 | func TestSinkAppLinesCarryParsedJSON(t *testing.T) { |
| 49 | now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)) |
| 50 | sk := NewSink("box1", now) |
| 51 | sk.App("bde", "ledgerpay", "h", `{"level":"warn","msg":"slow","request_id":"r1"}`) |
| 52 | rec := sk.Drain()[AppKey{"bde", "ledgerpay"}][0] |
| 53 | if rec.Msg != "slow" || rec.Level != LevelWarn || rec.Fields["request_id"] != "r1" { |
| 54 | t.Fatalf("structured line not parsed at the sink: %+v", rec) |
| 55 | } |
| 56 | } |
| 57 | |
| 58 | // TestSinkRingCapDropsOldestAndSaysSo: over the per-app byte cap the oldest |
| 59 | // lines go first, and the drain carries one synthetic platform record |
| 60 | // naming the count — never silent thinning (spec §4). |
| 61 | func TestSinkRingCapDropsOldestAndSaysSo(t *testing.T) { |
| 62 | now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)) |
| 63 | sk := NewSink("box1", now) |
| 64 | sk.MaxBytesPerApp = 600 // tiny for the test |
| 65 | sk.Burst = 1 << 20 // rate cap out of the way |
| 66 | for i := 0; i < 50; i++ { |
| 67 | sk.App("bde", "ledgerpay", "h", strings.Repeat("x", 40)) |
| 68 | } |
| 69 | recs := sk.Drain()[AppKey{"bde", "ledgerpay"}] |
| 70 | if len(recs) >= 50 { |
| 71 | t.Fatalf("nothing evicted: %d records", len(recs)) |
| 72 | } |
| 73 | last := recs[len(recs)-1] |
| 74 | if last.Stream != StreamPlatform || last.Level != LevelWarn || |
| 75 | !strings.Contains(last.Msg, "dropped") { |
| 76 | t.Fatalf("no honest drop record; last = %+v", last) |
| 77 | } |
| 78 | // The synthetic record is accounted once per drain, not once per drop. |
| 79 | for _, r := range recs[:len(recs)-1] { |
| 80 | if strings.Contains(r.Msg, "dropped") { |
| 81 | t.Fatalf("drop record duplicated mid-stream: %+v", r) |
| 82 | } |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | // TestSinkRateCapRefills: the token bucket drops lines while exhausted and |
| 87 | // admits again after time passes; drops surface in the same honest record. |
| 88 | func TestSinkRateCapRefills(t *testing.T) { |
| 89 | now, advance := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)) |
| 90 | sk := NewSink("box1", now) |
| 91 | sk.RatePerMin = 300 |
| 92 | sk.Burst = 300 |
| 93 | sk.App("bde", "ledgerpay", "h", strings.Repeat("a", 200)) // fits burst |
| 94 | sk.App("bde", "ledgerpay", "h", strings.Repeat("b", 200)) // bucket empty → dropped |
| 95 | recs := sk.Drain()[AppKey{"bde", "ledgerpay"}] |
| 96 | if len(recs) != 2 { // the admitted line + the drop notice |
| 97 | t.Fatalf("got %d records: %+v", len(recs), recs) |
| 98 | } |
| 99 | if !strings.Contains(recs[1].Msg, "dropped 1") { |
| 100 | t.Fatalf("drop notice wrong: %+v", recs[1]) |
| 101 | } |
| 102 | advance(time.Minute) |
| 103 | sk.App("bde", "ledgerpay", "h", strings.Repeat("c", 200)) |
| 104 | if got := sk.Drain()[AppKey{"bde", "ledgerpay"}]; len(got) != 1 || !strings.HasPrefix(got[0].Msg, "ccc") { |
| 105 | t.Fatalf("bucket did not refill: %+v", got) |
| 106 | } |
| 107 | } |
| 108 | |
| 109 | // TestSinkEventsBypassTheRateCap: platform events are the diagnosis — a |
| 110 | // chatty app must not be able to crowd out its own wake-failure line. |
| 111 | func TestSinkEventsBypassTheRateCap(t *testing.T) { |
| 112 | now, _ := testClock(time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)) |
| 113 | sk := NewSink("box1", now) |
| 114 | sk.RatePerMin = 10 |
| 115 | sk.Burst = 10 |
| 116 | sk.App("bde", "ledgerpay", "h", strings.Repeat("a", 400)) // exhausts the bucket |
| 117 | sk.Event("bde", "ledgerpay", "h", LevelError, "wake failed") |
| 118 | recs := sk.Drain()[AppKey{"bde", "ledgerpay"}] |
| 119 | var sawEvent bool |
| 120 | for _, r := range recs { |
| 121 | if r.Stream == StreamPlatform && r.Msg == "wake failed" { |
| 122 | sawEvent = true |
| 123 | } |
| 124 | } |
| 125 | if !sawEvent { |
| 126 | t.Fatalf("event lost to the app's own rate cap: %+v", recs) |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | func TestSinkDrainEmptyIsEmpty(t *testing.T) { |
| 131 | now, _ := testClock(time.Now()) |
| 132 | sk := NewSink("box1", now) |
| 133 | if got := sk.Drain(); len(got) != 0 { |
| 134 | t.Fatalf("empty sink drained %v", got) |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | // TestIsDropNotice: the rate-cap line Drain appends is recognised, and only |
| 139 | // it — Availability leaves it out by this test, and a platform event that |
| 140 | // merely starts with "dropped" must not vanish with it. |
| 141 | func TestIsDropNotice(t *testing.T) { |
| 142 | now, _ := testClock(time.Date(2026, 9, 24, 12, 0, 0, 0, time.UTC)) |
| 143 | sk := NewSink("box1", now) |
| 144 | sk.RatePerMin, sk.Burst = 300, 300 |
| 145 | sk.App("bab", "web", "h", strings.Repeat("a", 200)) // fits burst |
| 146 | sk.App("bab", "web", "h", strings.Repeat("b", 200)) // bucket empty → dropped |
| 147 | var notice *Record |
| 148 | for _, r := range sk.Drain()[AppKey{"bab", "web"}] { |
| 149 | if r.Stream == StreamPlatform { |
| 150 | notice = &r |
| 151 | } |
| 152 | } |
| 153 | if notice == nil || !IsDropNotice(*notice) { |
| 154 | t.Fatalf("drain's rate-cap line not recognised: %+v", notice) |
| 155 | } |
| 156 | for _, r := range []Record{ |
| 157 | {Stream: StreamPlatform, Level: LevelWarn, Msg: "dropped connection to the database"}, |
| 158 | {Stream: StreamApp, Msg: notice.Msg}, |
| 159 | } { |
| 160 | if IsDropNotice(r) { |
| 161 | t.Errorf("IsDropNotice(%+v) = true", r) |
| 162 | } |
| 163 | } |
| 164 | } |
| 165 | |
amadan