1package observability
2
3import (
4 "bufio"
5 "encoding/json"
6 "os"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/bnema/vev/internal/ports"
12)
13
14func TestTransportObservabilityJSONLOwnsTicksAndSchema(t *testing.T) {
15 path := t.TempDir() + "/daemon.jsonl"
16 observer, closer, err := NewJSONL(path, fixedRuntimeClock{now: time.Unix(0, 41)}, "daemon-process")
17 if err != nil {
18 t.Fatalf("NewJSONL() error = %v", err)
19 }
20
21 observer.ObserveRuntime(validRuntimeMark(ports.RuntimeComposeStart))
22 // A producer's clock domain must never enter a JSONL record. Rejection is
23 // observable as no second record, rather than a silently rewritten mark.
24 producerTicked := validRuntimeMark(ports.RuntimeComposeEnd)
25 producerTicked.Tick = 99
26 observer.ObserveRuntime(producerTicked)
27 if err := closer.Close(); err != nil {
28 t.Fatalf("Close() error = %v", err)
29 }
30
31 records := readJSONLRecords(t, path)
32 if len(records) != 1 {
33 t.Fatalf("JSONL records = %d, want 1 (caller Tick must be rejected)", len(records))
34 }
35 wantKeys := map[string]bool{
36 "schema": true, "process_id": true, "component": true, "scenario": true,
37 "run": true, "sequence": true, "request_id": true, "epoch": true,
38 "kind": true, "tick": true, "bytes": true, "fragments": true,
39 "retransmits": true, "pending": true, "ack_rtt_nanos": true, "valid": true,
40 }
41 if len(records[0]) != len(wantKeys) {
42 t.Fatalf("JSONL fields = %#v, want exactly canonical schema", records[0])
43 }
44 for key := range wantKeys {
45 if _, ok := records[0][key]; !ok {
46 t.Errorf("JSONL missing %q", key)
47 }
48 }
49 if got := records[0]["tick"]; got != float64(41) {
50 t.Errorf("JSONL tick = %v, want observer clock tick 41", got)
51 }
52 if got := records[0]["process_id"]; got != "daemon-process" {
53 t.Errorf("JSONL process_id = %v, want observer process", got)
54 }
55}
56
57func TestTransportObservabilityJSONLRejectsInvalidRequiredFields(t *testing.T) {
58 for _, mark := range []ports.RuntimeMark{
59 {},
60 {Schema: 1, ProcessID: "p", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, Kind: ports.RuntimeEmitStart},
61 {Schema: 2, ProcessID: "p", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, Kind: ports.RuntimeEmitStart, Valid: true},
62 } {
63 path := t.TempDir() + "/invalid.jsonl"
64 observer, closer, err := NewJSONL(path, fixedRuntimeClock{now: time.Unix(0, 5)}, "p")
65 if err != nil {
66 t.Fatalf("NewJSONL() error = %v", err)
67 }
68 observer.ObserveRuntime(mark)
69 if err := closer.Close(); err != nil {
70 t.Fatalf("Close() error = %v", err)
71 }
72 if records := readJSONLRecords(t, path); len(records) != 0 {
73 t.Errorf("invalid mark %#v wrote %#v", mark, records)
74 }
75 }
76}
77
78func TestTransportObservabilityJSONLWriteErrorSurfacesOnClose(t *testing.T) {
79 // /dev/full accepts Open but rejects every write, exercising the concrete
80 // observer lifecycle without changing the one-way RuntimeObserver port.
81 observer, closer, err := NewJSONL("/dev/full", fixedRuntimeClock{now: time.Unix(0, 1)}, "p")
82 if err != nil {
83 t.Skipf("/dev/full unavailable: %v", err)
84 }
85 observer.ObserveRuntime(validRuntimeMark(ports.RuntimeEmitStart))
86 if err := closer.Close(); err == nil {
87 t.Fatal("Close() error = nil, want surfaced JSONL write error")
88 }
89}
90
91func TestTransportObservabilitySameProcessSpanClockDomains(t *testing.T) {
92 for _, tc := range []struct {
93 name string
94 start ports.RuntimeMarkKind
95 end ports.RuntimeMarkKind
96 field string
97 }{
98 {"diff", ports.RuntimeDiffStart, ports.RuntimeDiffEnd, "diff_duration"},
99 {"queue", ports.RuntimeQueueEnqueued, ports.RuntimeQueueDequeued, "queue_wait"},
100 {"ack", ports.RuntimeACKBlockedStart, ports.RuntimeACKBlockedEnd, "ack_blocked_interval"},
101 {"send", ports.RuntimeAdapterSendStart, ports.RuntimeAdapterSendEnd, "adapter_send_duration"},
102 {"receive", ports.RuntimeAdapterReceiveStart, ports.RuntimeAdapterReceiveEnd, "adapter_receive_duration"},
103 } {
104 t.Run(tc.name, func(t *testing.T) {
105 start := validRuntimeMark(tc.start)
106 start.Tick = 11
107 end := start
108 end.Kind, end.Tick = tc.end, 19
109 if got, err := sameProcessDuration(start, end); err != nil || got != 8 {
110 t.Fatalf("%s = %d, %v; want 8, nil", tc.field, got, err)
111 }
112
113 end.ProcessID = "other-process"
114 if _, err := sameProcessDuration(start, end); err == nil {
115 t.Fatalf("%s accepted cross-process ticks", tc.field)
116 }
117 })
118 }
119}
120
121func TestTransportObservabilityJSONLSerializesConcurrentMarks(t *testing.T) {
122 path := t.TempDir() + "/concurrent.jsonl"
123 observer, closer, err := NewJSONL(path, fixedRuntimeClock{now: time.Unix(0, 73)}, "ipc-process")
124 if err != nil {
125 t.Fatalf("NewJSONL() error = %v", err)
126 }
127 var wg sync.WaitGroup
128 for i := uint64(0); i < 32; i++ {
129 wg.Add(1)
130 go func(sequence uint64) {
131 defer wg.Done()
132 mark := validRuntimeMark(ports.RuntimeAdapterSendEnd)
133 mark.Sequence = sequence + 1
134 observer.ObserveRuntime(mark)
135 }(i)
136 }
137 wg.Wait()
138 if err := closer.Close(); err != nil {
139 t.Fatalf("Close() error = %v", err)
140 }
141 if records := readJSONLRecords(t, path); len(records) != 32 {
142 t.Fatalf("concurrently serialized records = %d, want 32", len(records))
143 }
144}
145
146func sameProcessDuration(start, end ports.RuntimeMark) (int64, error) {
147 if start.ProcessID == "" || start.ProcessID != end.ProcessID {
148 return 0, os.ErrInvalid
149 }
150 if start.Scenario != end.Scenario || start.Run != end.Run || start.Sequence != end.Sequence || start.RequestID != end.RequestID || start.Epoch != end.Epoch || end.Tick < start.Tick {
151 return 0, os.ErrInvalid
152 }
153 return end.Tick - start.Tick, nil
154}
155
156func validRuntimeMark(kind ports.RuntimeMarkKind) ports.RuntimeMark {
157 return ports.RuntimeMark{Schema: 1, ProcessID: "daemon-process", Component: "daemon", Scenario: "resize", Run: 7, Sequence: 9, RequestID: 11, Epoch: 13, Kind: kind, Valid: true}
158}
159
160func readJSONLRecords(t *testing.T, path string) []map[string]any {
161 t.Helper()
162 f, err := os.Open(path)
163 if err != nil {
164 t.Fatalf("Open(%q): %v", path, err)
165 }
166 defer func() { _ = f.Close() }()
167 var records []map[string]any
168 scanner := bufio.NewScanner(f)
169 for scanner.Scan() {
170 var record map[string]any
171 if err := json.Unmarshal(scanner.Bytes(), &record); err != nil {
172 t.Fatalf("invalid JSONL record %q: %v", scanner.Text(), err)
173 }
174 records = append(records, record)
175 }
176 if err := scanner.Err(); err != nil {
177 t.Fatalf("reading JSONL: %v", err)
178 }
179 return records
180}
181
182type fixedRuntimeClock struct{ now time.Time }
183
184func (c fixedRuntimeClock) Now() time.Time { return c.now }
185func (c fixedRuntimeClock) NewTimer(d time.Duration) ports.Timer {
186 return fixedRuntimeTimer{timer: time.NewTimer(d)}
187}
188
189type fixedRuntimeTimer struct{ timer *time.Timer }
190
191func (t fixedRuntimeTimer) C() <-chan time.Time { return t.timer.C }
192func (t fixedRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
193func (t fixedRuntimeTimer) Stop() bool { return t.timer.Stop() }