1package ipc
2
3import (
4 "encoding/json"
5 "net"
6 "os"
7 "reflect"
8 "strings"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/bnema/vev/internal/adapters/observability"
14 "github.com/bnema/vev/internal/ports"
15)
16
17func TestTransportObservabilityIPCEOFAndCloseEndFailedSpans(t *testing.T) {
18 trace := t.TempDir() + "/ipc-shutdown.jsonl"
19 observer, closer, err := observability.NewJSONL(trace, ipcRuntimeClock{now: time.Unix(0, 223)}, "ipc-process")
20 if err != nil {
21 t.Fatalf("NewJSONL() error = %v", err)
22 }
23 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
24 defer reporter.Close()
25
26 // EOF from the peer and a locally closed blocked receive are distinct
27 // shutdown paths; both must record a failed, correlated end.
28 for _, closeLocal := range []bool{false, true} {
29 left, right := net.Pipe()
30 var started <-chan struct{}
31 conn := left
32 if closeLocal {
33 observedConn := &ipcReadStartedConn{Conn: left, started: make(chan struct{})}
34 conn, started = observedConn, observedConn.started
35 }
36 transport := NewTransport(conn, WithRuntimeObserver(reporter))
37 recvDone := make(chan error, 1)
38 go func() { _, err := transport.Recv(); recvDone <- err }()
39 if closeLocal {
40 <-started
41 if err := transport.Close(); err != nil {
42 t.Fatalf("Close() error = %v", err)
43 }
44 } else if err := right.Close(); err != nil {
45 t.Fatalf("peer Close() error = %v", err)
46 }
47 if err := <-recvDone; err == nil {
48 t.Fatal("Recv() error = nil after transport shutdown")
49 }
50 _ = left.Close()
51 _ = right.Close()
52 }
53
54 // A peer shutdown also makes Send fail after its start mark.
55 left, right := net.Pipe()
56 transport := NewTransport(left, WithRuntimeObserver(reporter))
57 _ = right.Close()
58 if err := transport.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("failed")}); err == nil {
59 t.Fatal("Send() error = nil after peer close")
60 }
61 _ = transport.Close()
62 reporter.Close()
63 if err := closer.Close(); err != nil {
64 t.Fatalf("trace Close() error = %v", err)
65 }
66 assertIPCFailedSpanPairs(t, trace)
67}
68
69type ipcReadStartedConn struct {
70 net.Conn
71 started chan struct{}
72 once sync.Once
73}
74
75func (c *ipcReadStartedConn) Read(p []byte) (int, error) {
76 c.once.Do(func() { close(c.started) })
77 return c.Conn.Read(p)
78}
79
80func assertIPCFailedSpanPairs(t *testing.T, path string) {
81 t.Helper()
82 data, err := os.ReadFile(path)
83 if err != nil {
84 t.Fatalf("ReadFile(trace): %v", err)
85 }
86 type mark struct {
87 ProcessID string `json:"process_id"`
88 Component string `json:"component"`
89 Scenario string `json:"scenario"`
90 Run uint64 `json:"run"`
91 Sequence uint64 `json:"sequence"`
92 RequestID uint64 `json:"request_id"`
93 Epoch uint64 `json:"epoch"`
94 Kind string `json:"kind"`
95 Valid bool `json:"valid"`
96 }
97 type spanKey struct {
98 processID, scenario string
99 run, sequence, requestID, epoch uint64
100 }
101 starts := make(map[spanKey]mark)
102 startCounts := map[string]int{}
103 for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
104 var m mark
105 if err := json.Unmarshal([]byte(line), &m); err != nil {
106 t.Fatalf("Unmarshal(trace): %v", err)
107 }
108 key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch}
109 switch m.Kind {
110 case "adapter_receive_start", "adapter_send_start":
111 if m.ProcessID != "ipc-process" || m.Component != "ipc" || m.Scenario != "runtime" || m.Run != 1 || m.Sequence == 0 || m.RequestID != m.Sequence || m.Epoch != m.Sequence || !m.Valid {
112 t.Fatalf("adapter start has invalid identity: %#v", m)
113 }
114 if _, exists := starts[key]; exists {
115 t.Fatalf("duplicate adapter start for %#v", key)
116 }
117 starts[key] = m
118 startCounts[m.Kind]++
119 case "adapter_receive_end", "adapter_send_end":
120 start, ok := starts[key]
121 if !ok || strings.TrimSuffix(start.Kind, "_start") != strings.TrimSuffix(m.Kind, "_end") {
122 t.Fatalf("unmatched adapter end %#v", m)
123 }
124 if m.Valid {
125 t.Fatalf("failed adapter end is valid: %#v", m)
126 }
127 delete(starts, key)
128 }
129 }
130 if len(starts) != 0 {
131 t.Fatalf("unmatched adapter starts: %#v", starts)
132 }
133 if got := startCounts["adapter_receive_start"]; got != 2 {
134 t.Errorf("adapter receive starts = %d, want 2", got)
135 }
136 if got := startCounts["adapter_send_start"]; got != 1 {
137 t.Errorf("adapter send starts = %d, want 1", got)
138 }
139}
140
141func TestTransportObservabilityIPCMarksCarriageWithoutChangingBytes(t *testing.T) {
142 trace := t.TempDir() + "/ipc.jsonl"
143 observer, closer, err := observability.NewJSONL(trace, ipcRuntimeClock{now: time.Unix(0, 211)}, "ipc-process")
144 if err != nil {
145 t.Fatalf("NewJSONL() error = %v", err)
146 }
147 reporter := ports.NewSerializedRuntimeObserver(observer, 64)
148 defer reporter.Close()
149
150 left, right := net.Pipe()
151 defer func() { _ = left.Close() }()
152 defer func() { _ = right.Close() }()
153 client := NewTransport(left, WithRuntimeObserver(reporter))
154 server := NewTransport(right, WithRuntimeObserver(reporter))
155 want := ports.Frame{Type: ports.MsgOutput, Payload: []byte("wire bytes stay exact")}
156
157 var wg sync.WaitGroup
158 wg.Go(func() {
159 if err := client.Send(want); err != nil {
160 t.Errorf("Send() error = %v", err)
161 }
162 })
163 got, err := server.Recv()
164 wg.Wait()
165 if err != nil {
166 t.Fatalf("Recv() error = %v", err)
167 }
168 if !reflect.DeepEqual(got, want) {
169 t.Fatalf("carried frame = %#v, want %#v", got, want)
170 }
171 reporter.Close()
172 if err := closer.Close(); err != nil {
173 t.Fatalf("trace Close() error = %v", err)
174 }
175 traceBytes, err := os.ReadFile(trace)
176 if err != nil {
177 t.Fatalf("ReadFile(trace): %v", err)
178 }
179 for _, kind := range []string{"adapter_send_start", "adapter_send_end", "adapter_receive_start", "adapter_receive_end"} {
180 if !strings.Contains(string(traceBytes), `"kind":"`+kind+`"`) {
181 t.Errorf("trace missing %s", kind)
182 }
183 }
184}
185
186type ipcRuntimeClock struct{ now time.Time }
187
188func (c ipcRuntimeClock) Now() time.Time { return c.now }
189func (c ipcRuntimeClock) NewTimer(d time.Duration) ports.Timer {
190 return ipcRuntimeTimer{timer: time.NewTimer(d)}
191}
192
193type ipcRuntimeTimer struct{ timer *time.Timer }
194
195func (t ipcRuntimeTimer) C() <-chan time.Time { return t.timer.C }
196func (t ipcRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
197func (t ipcRuntimeTimer) Stop() bool { return t.timer.Stop() }