1package dgram
2
3import (
4 "encoding/json"
5 "errors"
6 "net"
7 "os"
8 "reflect"
9 "strings"
10 "sync"
11 "testing"
12 "time"
13
14 "github.com/bnema/vev/internal/adapters/observability"
15 "github.com/bnema/vev/internal/ports"
16)
17
18func TestTransportObservabilityDgramFailedCloseEndsReceiveAndSend(t *testing.T) {
19 behaviorClock := newManualClock(time.Unix(0, 417))
20 link := newSimulatedLink(behaviorClock, packetPolicy{})
21 aPC, _ := newPairWithCapacity(link, 64)
22 closeErr := errors.New("packet connection close failed")
23 tracePath := t.TempDir() + "/dgram-close.jsonl"
24 observer, closer, err := observability.NewJSONL(tracePath, dgramObserverClock{now: time.Unix(0, 521)}, "dgram-process")
25 if err != nil {
26 t.Fatalf("NewJSONL() error = %v", err)
27 }
28 receiveStarted := make(chan struct{})
29 signalingObserver := &dgramStartObserver{RuntimeObserver: observer, receiveStarted: receiveStarted}
30 reporter := ports.NewSerializedRuntimeObserver(signalingObserver, 64)
31 defer reporter.Close()
32 transport, err := NewTransportWithOptions(&dgramCloseFailPC{fakePC: aPC, err: closeErr}, testAddr("b"), key(), 1, 2, Options{Clock: behaviorClock, ResendAfter: time.Hour, Heartbeat: time.Hour}, WithRuntimeObserver(reporter))
33 if err != nil {
34 t.Fatalf("NewTransportWithOptions() error = %v", err)
35 }
36 if err := transport.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("write failure")}); err == nil {
37 t.Fatal("Send() error = nil on packet write failure")
38 }
39 recvDone := make(chan error, 1)
40 go func() { _, err := transport.Recv(); recvDone <- err }()
41 <-receiveStarted
42 if err := transport.Close(); !errors.Is(err, closeErr) {
43 t.Fatalf("Close() error = %v, want %v", err, closeErr)
44 }
45 if err := <-recvDone; err == nil {
46 t.Fatal("Recv() error = nil after failed Close()")
47 }
48 reporter.Close()
49 if err := closer.Close(); err != nil {
50 t.Fatalf("trace Close() error = %v", err)
51 }
52 assertDgramFailedSpanPairs(t, tracePath)
53}
54
55type dgramStartObserver struct {
56 ports.RuntimeObserver
57 receiveStarted chan struct{}
58 once sync.Once
59}
60
61func (o *dgramStartObserver) ObserveRuntime(mark ports.RuntimeMark) {
62 if mark.Kind == ports.RuntimeAdapterReceiveStart {
63 o.once.Do(func() { close(o.receiveStarted) })
64 }
65 o.RuntimeObserver.ObserveRuntime(mark)
66}
67
68type dgramCloseFailPC struct {
69 *fakePC
70 err error
71}
72
73func (p *dgramCloseFailPC) WriteTo([]byte, net.Addr) (int, error) {
74 return 0, errors.New("packet write failed")
75}
76
77func (p *dgramCloseFailPC) Close() error {
78 _ = p.fakePC.Close()
79 return p.err
80}
81
82func assertDgramFailedSpanPairs(t *testing.T, path string) {
83 t.Helper()
84 data, err := os.ReadFile(path)
85 if err != nil {
86 t.Fatalf("ReadFile(trace): %v", err)
87 }
88 type mark struct {
89 ProcessID string `json:"process_id"`
90 Scenario string `json:"scenario"`
91 Run uint64 `json:"run"`
92 Sequence uint64 `json:"sequence"`
93 RequestID uint64 `json:"request_id"`
94 Epoch uint64 `json:"epoch"`
95 Kind string `json:"kind"`
96 Valid bool `json:"valid"`
97 }
98 type spanKey struct {
99 processID, scenario string
100 run, sequence, requestID, epoch uint64
101 kind string
102 }
103 starts := make(map[spanKey]mark)
104 pairs := map[string]int{}
105 for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
106 var m mark
107 if err := json.Unmarshal([]byte(line), &m); err != nil {
108 t.Fatalf("Unmarshal(trace): %v", err)
109 }
110 key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch, m.Kind}
111 switch m.Kind {
112 case "adapter_receive_start", "adapter_send_start":
113 if _, exists := starts[key]; exists {
114 t.Fatalf("duplicate adapter start: %#v", m)
115 }
116 starts[key] = m
117 case "adapter_receive_end", "adapter_send_end":
118 key.kind = strings.TrimSuffix(m.Kind, "_end") + "_start"
119 start, ok := starts[key]
120 if !ok || m.Valid {
121 t.Fatalf("unmatched or valid failed end: start=%#v end=%#v", start, m)
122 }
123 pairs[strings.TrimSuffix(m.Kind, "_end")]++
124 delete(starts, key)
125 }
126 }
127 if len(starts) != 0 {
128 t.Fatalf("unmatched adapter starts: %#v", starts)
129 }
130 if pairs["adapter_send"] != 1 || pairs["adapter_receive"] != 1 || len(pairs) != 2 {
131 t.Fatalf("failed adapter span pairs=%v, want exactly one send and one receive", pairs)
132 }
133}
134
135func TestTransportObservabilityDgramKeepsBehaviorClockSeparate(t *testing.T) {
136 behaviorClock := newManualClock(time.Unix(0, 401))
137 link := newSimulatedLink(behaviorClock, packetPolicy{})
138 aPC, bPC := newPairWithCapacity(link, 64)
139 tracePath := t.TempDir() + "/dgram.jsonl"
140 observer, closer, err := observability.NewJSONL(tracePath, dgramObserverClock{now: time.Unix(0, 509)}, "dgram-process")
141 if err != nil {
142 t.Fatalf("NewJSONL() error = %v", err)
143 }
144 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
145 defer reporter.Close()
146
147 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: behaviorClock, ResendAfter: time.Hour, Heartbeat: time.Hour}, WithRuntimeObserver(reporter))
148 if err != nil {
149 t.Fatalf("NewTransportWithOptions(a) error = %v", err)
150 }
151 defer func() { _ = a.Close() }()
152 b, err := NewTransportWithOptions(bPC, testAddr("a"), key(), 2, 1, Options{Clock: behaviorClock, ResendAfter: time.Hour, Heartbeat: time.Hour}, WithRuntimeObserver(reporter))
153 if err != nil {
154 t.Fatalf("NewTransportWithOptions(b) error = %v", err)
155 }
156 defer func() { _ = b.Close() }()
157
158 want := ports.Frame{Type: ports.MsgOutput, Payload: []byte("datagram carriage remains exact")}
159 if err := a.Send(want); err != nil {
160 t.Fatalf("Send() error = %v", err)
161 }
162 gotCh := make(chan ports.Frame, 1)
163 errCh := make(chan error, 1)
164 go func() {
165 frame, recvErr := b.Recv()
166 if recvErr != nil {
167 errCh <- recvErr
168 return
169 }
170 gotCh <- frame
171 }()
172 select {
173 case err := <-errCh:
174 t.Fatalf("Recv() error = %v", err)
175 case got := <-gotCh:
176 if !reflect.DeepEqual(got, want) {
177 t.Fatalf("carried frame = %#v, want %#v", got, want)
178 }
179 case <-time.After(time.Second):
180 t.Fatal("Recv() did not receive deterministic simulated datagram")
181 }
182 reporter.Close()
183 if err := closer.Close(); err != nil {
184 t.Fatalf("trace Close() error = %v", err)
185 }
186 trace, err := os.ReadFile(tracePath)
187 if err != nil {
188 t.Fatalf("ReadFile(trace): %v", err)
189 }
190 for _, kind := range []string{"adapter_send_start", "adapter_send_end", "adapter_receive_start", "adapter_receive_end"} {
191 if !strings.Contains(string(trace), `"kind":"`+kind+`"`) {
192 t.Errorf("trace missing %s", kind)
193 }
194 }
195}
196
197type dgramObserverClock struct{ now time.Time }
198
199func (c dgramObserverClock) Now() time.Time { return c.now }
200func (c dgramObserverClock) NewTimer(d time.Duration) ports.Timer {
201 return dgramObserverTimer{timer: time.NewTimer(d)}
202}
203
204type dgramObserverTimer struct{ timer *time.Timer }
205
206func (t dgramObserverTimer) C() <-chan time.Time { return t.timer.C }
207func (t dgramObserverTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
208func (t dgramObserverTimer) Stop() bool { return t.timer.Stop() }