1package ports
2
3import (
4 "reflect"
5 "sync"
6 "testing"
7 "time"
8)
9
10func TestRuntimeObserverContract(t *testing.T) {
11 wantFields := []struct {
12 name string
13 typ reflect.Type
14 }{
15 {"Schema", reflect.TypeFor[uint16]()},
16 {"ProcessID", reflect.TypeFor[string]()},
17 {"Component", reflect.TypeFor[string]()},
18 {"Scenario", reflect.TypeFor[string]()},
19 {"Run", reflect.TypeFor[uint64]()},
20 {"Sequence", reflect.TypeFor[uint64]()},
21 {"RequestID", reflect.TypeFor[uint64]()},
22 {"Epoch", reflect.TypeFor[uint64]()},
23 {"Kind", reflect.TypeFor[RuntimeMarkKind]()},
24 {"Tick", reflect.TypeFor[int64]()},
25 {"Bytes", reflect.TypeFor[uint64]()},
26 {"Fragments", reflect.TypeFor[uint64]()},
27 {"Retransmits", reflect.TypeFor[uint64]()},
28 {"Pending", reflect.TypeFor[uint64]()},
29 {"AckRTTNanos", reflect.TypeFor[int64]()},
30 {"Valid", reflect.TypeFor[bool]()},
31 }
32
33 typ := reflect.TypeFor[RuntimeMark]()
34 if typ.NumField() != len(wantFields) {
35 t.Fatalf("RuntimeMark has %d fields, want exactly %d", typ.NumField(), len(wantFields))
36 }
37 for i, want := range wantFields {
38 field := typ.Field(i)
39 if field.Name != want.name || field.Type != want.typ {
40 t.Fatalf("RuntimeMark field %d = %s %v, want %s %v", i, field.Name, field.Type, want.name, want.typ)
41 }
42 }
43
44 var _ RuntimeObserver = runtimeObserverFunc(func(RuntimeMark) {})
45}
46
47func TestSerializedRuntimeObserverFlushesInOrderAndCloses(t *testing.T) {
48 var mu sync.Mutex
49 var got []RuntimeMark
50 observer := NewSerializedRuntimeObserver(runtimeObserverFunc(func(mark RuntimeMark) {
51 mu.Lock()
52 got = append(got, mark)
53 mu.Unlock()
54 }), 2)
55 observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitStart, 1, true))
56 observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitEnd, 1, true))
57 observer.Flush()
58 observer.Close()
59 observer.Close()
60
61 mu.Lock()
62 defer mu.Unlock()
63 if len(got) != 2 || got[0].Kind != RuntimeEmitStart || got[1].Kind != RuntimeEmitEnd {
64 t.Fatalf("serialized marks = %#v", got)
65 }
66}
67
68func TestEnsureSerializedRuntimeObserverReusesExistingOwner(t *testing.T) {
69 existing := NewSerializedRuntimeObserver(runtimeObserverFunc(func(RuntimeMark) {}), 1)
70 reporter, owned := EnsureSerializedRuntimeObserver(existing, 1)
71 if owned || reporter != existing {
72 t.Fatalf("EnsureSerializedRuntimeObserver() = (%T, %t), want existing unowned reporter", reporter, owned)
73 }
74 existing.Close()
75}
76
77func TestSerializedRuntimeObserverReportsBoundedQueueLoss(t *testing.T) {
78 entered := make(chan struct{})
79 release := make(chan struct{})
80 var mu sync.Mutex
81 var got []RuntimeMark
82 observer := NewSerializedRuntimeObserver(runtimeObserverFunc(func(mark RuntimeMark) {
83 if mark.Kind == RuntimeEmitStart {
84 close(entered)
85 <-release
86 }
87 mu.Lock()
88 got = append(got, mark)
89 mu.Unlock()
90 }), 2)
91 observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitStart, 1, true))
92 select {
93 case <-entered:
94 case <-time.After(time.Second):
95 t.Fatal("observer did not start")
96 }
97 observer.ObserveRuntime(NewRuntimeMark("client", RuntimeComposeStart, 1, true))
98 observer.ObserveRuntime(NewRuntimeMark("client", RuntimeComposeEnd, 1, true))
99 observer.ObserveRuntime(NewRuntimeMark("client", RuntimeEmitEnd, 1, true)) // bounded loss
100 close(release)
101 observer.Flush()
102 observer.Close()
103
104 mu.Lock()
105 defer mu.Unlock()
106 if len(got) != 4 || got[0].Kind != RuntimeEmitStart || got[1].Kind != RuntimeComposeStart || got[2].Kind != RuntimeComposeEnd || got[3].Kind != RuntimeTransportDiagnostic || got[3].Bytes != 1 || got[3].Valid {
107 t.Fatalf("bounded-loss marks = %#v", got)
108 }
109}
110
111func TestP4RuntimeScreenMarkKindsValidateSchema1(t *testing.T) {
112 if RuntimeMarkSchema != 1 {
113 t.Fatalf("RuntimeMarkSchema = %d, want 1", RuntimeMarkSchema)
114 }
115 for _, tc := range []struct {
116 kind RuntimeMarkKind
117 want string
118 }{
119 {RuntimeScreenSnapshot, "screen_snapshot"},
120 {RuntimeScreenDelta, "screen_delta"},
121 {RuntimeScreenResetRequested, "screen_reset_requested"},
122 } {
123 if string(tc.kind) != tc.want {
124 t.Fatalf("screen mark kind = %q, want %q", tc.kind, tc.want)
125 }
126 mark := NewRuntimeMark("daemon", tc.kind, 17, true)
127 if !RuntimeMarkValid(mark) {
128 t.Fatalf("screen mark kind %q did not validate: %+v", tc.kind, mark)
129 }
130 }
131}
132
133func TestRuntimeObserverRequiredSpanKinds(t *testing.T) {
134 pairs := []struct {
135 start RuntimeMarkKind
136 end RuntimeMarkKind
137 }{
138 {RuntimeDiffStart, RuntimeDiffEnd},
139 {RuntimeQueueEnqueued, RuntimeQueueDequeued},
140 {RuntimeACKBlockedStart, RuntimeACKBlockedEnd},
141 {RuntimeAdapterSendStart, RuntimeAdapterSendEnd},
142 {RuntimeAdapterReceiveStart, RuntimeAdapterReceiveEnd},
143 }
144 for _, pair := range pairs {
145 if pair.start == "" || pair.end == "" || pair.start == pair.end {
146 t.Fatalf("invalid required span mapping %#v", pair)
147 }
148 }
149}
150
151func TestRuntimeCorrelationObserverUsesHarnessProcessInputs(t *testing.T) {
152 var got []RuntimeMark
153 observer, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(mark RuntimeMark) {
154 got = append(got, mark)
155 }), RuntimeCorrelationInputs{Scenario: "1x4-idle-local", Run: 7})
156 if err != nil {
157 t.Fatal(err)
158 }
159
160 observer.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Kind: RuntimeEmitEnd, Valid: true})
161 observer.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Sequence: 9, RequestID: 10, Epoch: 11, Kind: RuntimeEmitStart, Valid: true})
162
163 if len(got) != 2 {
164 t.Fatalf("marks=%+v", got)
165 }
166 for _, mark := range got {
167 if mark.Scenario != "1x4-idle-local" || mark.Run != 7 {
168 t.Fatalf("harness correlation was not applied: %+v", mark)
169 }
170 if mark.Sequence == 0 || mark.RequestID == 0 || mark.Epoch == 0 {
171 t.Fatalf("missing deterministic operation identity: %+v", mark)
172 }
173 }
174 if got[1].Sequence != 9 || got[1].RequestID != 10 || got[1].Epoch != 11 {
175 t.Fatalf("explicit operation identity changed: %+v", got[1])
176 }
177}
178
179func TestRuntimeCorrelationObserverAllocatesMissingIDsProcessWide(t *testing.T) {
180 var got []RuntimeMark
181 collect := runtimeObserverFunc(func(mark RuntimeMark) { got = append(got, mark) })
182 first, err := NewRuntimeCorrelationObserver(collect, RuntimeCorrelationInputs{Scenario: "scenario", Run: 1})
183 if err != nil {
184 t.Fatal(err)
185 }
186 second, err := NewRuntimeCorrelationObserver(collect, RuntimeCorrelationInputs{Scenario: "scenario", Run: 1})
187 if err != nil {
188 t.Fatal(err)
189 }
190
191 first.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "first", Kind: RuntimeEmitStart, Valid: true})
192 second.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "second", Kind: RuntimeEmitEnd, Valid: true})
193
194 if len(got) != 2 {
195 t.Fatalf("marks=%+v", got)
196 }
197 if got[0].Sequence == got[1].Sequence {
198 t.Fatalf("wrapper-local sequence collision: %+v", got)
199 }
200 for _, mark := range got {
201 if mark.RequestID != mark.Sequence || mark.Epoch != mark.RequestID {
202 t.Fatalf("missing-ID fallback changed: %+v", mark)
203 }
204 }
205}
206
207func TestRuntimeCorrelationObserverRejectsInvalidHarnessInputs(t *testing.T) {
208 for _, in := range []RuntimeCorrelationInputs{{}, {Scenario: "scenario"}} {
209 if _, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(RuntimeMark) {}), in); err == nil {
210 t.Fatalf("invalid inputs accepted: %+v", in)
211 }
212 }
213}
214
215func TestRuntimeMarkHasNoRenderingPayloadOrPolicyFields(t *testing.T) {
216 typ := reflect.TypeFor[RuntimeMark]()
217 for _, forbidden := range []string{"Cell", "Cells", "Payload", "Frame", "Renderer", "Policy", "Transport"} {
218 if _, ok := typ.FieldByName(forbidden); ok {
219 t.Fatalf("RuntimeMark must not carry %s", forbidden)
220 }
221 }
222}
223
224type runtimeObserverFunc func(RuntimeMark)
225
226func (f runtimeObserverFunc) ObserveRuntime(mark RuntimeMark) { f(mark) }