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