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 TestRuntimeObserverRequiredSpanKinds(t *testing.T) {
112 pairs := []struct {
113 start RuntimeMarkKind
114 end RuntimeMarkKind
115 }{
116 {RuntimeDiffStart, RuntimeDiffEnd},
117 {RuntimeQueueEnqueued, RuntimeQueueDequeued},
118 {RuntimeACKBlockedStart, RuntimeACKBlockedEnd},
119 {RuntimeAdapterSendStart, RuntimeAdapterSendEnd},
120 {RuntimeAdapterReceiveStart, RuntimeAdapterReceiveEnd},
121 }
122 for _, pair := range pairs {
123 if pair.start == "" || pair.end == "" || pair.start == pair.end {
124 t.Fatalf("invalid required span mapping %#v", pair)
125 }
126 }
127}
128
129func TestRuntimeCorrelationObserverUsesHarnessProcessInputs(t *testing.T) {
130 var got []RuntimeMark
131 observer, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(mark RuntimeMark) {
132 got = append(got, mark)
133 }), RuntimeCorrelationInputs{Scenario: "1x4-idle-local", Run: 7})
134 if err != nil {
135 t.Fatal(err)
136 }
137
138 observer.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Kind: RuntimeEmitEnd, Valid: true})
139 observer.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "daemon", Scenario: "runtime", Run: 1, Sequence: 9, RequestID: 10, Epoch: 11, Kind: RuntimeEmitStart, Valid: true})
140
141 if len(got) != 2 {
142 t.Fatalf("marks=%+v", got)
143 }
144 for _, mark := range got {
145 if mark.Scenario != "1x4-idle-local" || mark.Run != 7 {
146 t.Fatalf("harness correlation was not applied: %+v", mark)
147 }
148 if mark.Sequence == 0 || mark.RequestID == 0 || mark.Epoch == 0 {
149 t.Fatalf("missing deterministic operation identity: %+v", mark)
150 }
151 }
152 if got[1].Sequence != 9 || got[1].RequestID != 10 || got[1].Epoch != 11 {
153 t.Fatalf("explicit operation identity changed: %+v", got[1])
154 }
155}
156
157func TestRuntimeCorrelationObserverAllocatesMissingIDsProcessWide(t *testing.T) {
158 var got []RuntimeMark
159 collect := runtimeObserverFunc(func(mark RuntimeMark) { got = append(got, mark) })
160 first, err := NewRuntimeCorrelationObserver(collect, RuntimeCorrelationInputs{Scenario: "scenario", Run: 1})
161 if err != nil {
162 t.Fatal(err)
163 }
164 second, err := NewRuntimeCorrelationObserver(collect, RuntimeCorrelationInputs{Scenario: "scenario", Run: 1})
165 if err != nil {
166 t.Fatal(err)
167 }
168
169 first.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "first", Kind: RuntimeEmitStart, Valid: true})
170 second.ObserveRuntime(RuntimeMark{Schema: RuntimeMarkSchema, Component: "second", Kind: RuntimeEmitEnd, Valid: true})
171
172 if len(got) != 2 {
173 t.Fatalf("marks=%+v", got)
174 }
175 if got[0].Sequence == got[1].Sequence {
176 t.Fatalf("wrapper-local sequence collision: %+v", got)
177 }
178 for _, mark := range got {
179 if mark.RequestID != mark.Sequence || mark.Epoch != mark.RequestID {
180 t.Fatalf("missing-ID fallback changed: %+v", mark)
181 }
182 }
183}
184
185func TestRuntimeCorrelationObserverRejectsInvalidHarnessInputs(t *testing.T) {
186 for _, in := range []RuntimeCorrelationInputs{{}, {Scenario: "scenario"}} {
187 if _, err := NewRuntimeCorrelationObserver(runtimeObserverFunc(func(RuntimeMark) {}), in); err == nil {
188 t.Fatalf("invalid inputs accepted: %+v", in)
189 }
190 }
191}
192
193func TestRuntimeMarkHasNoRenderingPayloadOrPolicyFields(t *testing.T) {
194 typ := reflect.TypeFor[RuntimeMark]()
195 for _, forbidden := range []string{"Cell", "Cells", "Payload", "Frame", "Renderer", "Policy", "Transport"} {
196 if _, ok := typ.FieldByName(forbidden); ok {
197 t.Fatalf("RuntimeMark must not carry %s", forbidden)
198 }
199 }
200}
201
202type runtimeObserverFunc func(RuntimeMark)
203
204func (f runtimeObserverFunc) ObserveRuntime(mark RuntimeMark) { f(mark) }