1package ipc
2
3import (
4 "encoding/binary"
5 "errors"
6 "io"
7 "net"
8 "reflect"
9 "runtime"
10 "sync"
11 "testing"
12 "time"
13
14 "github.com/bnema/vev/internal/domain"
15 "github.com/bnema/vev/internal/ports"
16)
17
18func TestTransportSendRecvBothDirections(t *testing.T) {
19 c1, c2 := net.Pipe()
20 defer func() { _ = c1.Close() }()
21 defer func() { _ = c2.Close() }()
22
23 client := NewTransport(c1)
24 server := NewTransport(c2)
25
26 clientToServer := ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(ports.Hello{Version: 1, Intent: ports.IntentNew, Name: "w0"})}
27 serverToClient := ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(ports.Welcome{SessionID: "s1", SessionName: "main"})}
28
29 var wg sync.WaitGroup
30 wg.Go(func() {
31 if err := client.Send(clientToServer); err != nil {
32 t.Errorf("client.Send() error = %v", err)
33 }
34 })
35
36 got, err := server.Recv()
37 wg.Wait() // ensure the sender goroutine has fully returned before any Fatalf
38 if err != nil {
39 t.Fatalf("server.Recv() error = %v", err)
40 }
41 if !reflect.DeepEqual(got, clientToServer) {
42 t.Fatalf("server.Recv() = %#v, want %#v", got, clientToServer)
43 }
44
45 wg.Go(func() {
46 if err := server.Send(serverToClient); err != nil {
47 t.Errorf("server.Send() error = %v", err)
48 }
49 })
50
51 got, err = client.Recv()
52 wg.Wait()
53 if err != nil {
54 t.Fatalf("client.Recv() error = %v", err)
55 }
56 if !reflect.DeepEqual(got, serverToClient) {
57 t.Fatalf("client.Recv() = %#v, want %#v", got, serverToClient)
58 }
59}
60
61func TestTransportEOFOnClose(t *testing.T) {
62 c1, c2 := net.Pipe()
63 server := NewTransport(c2)
64
65 if err := c1.Close(); err != nil {
66 t.Fatalf("c1.Close() error = %v", err)
67 }
68
69 _, err := server.Recv()
70 if !errors.Is(err, io.EOF) {
71 t.Fatalf("server.Recv() error = %v, want io.EOF", err)
72 }
73}
74
75func TestTransportCloseClosesUnderlyingConn(t *testing.T) {
76 c1, c2 := net.Pipe()
77 client := NewTransport(c1)
78 defer func() { _ = c2.Close() }()
79
80 if err := client.Close(); err != nil {
81 t.Fatalf("Close() error = %v", err)
82 }
83
84 // Writing to the closed end should now fail.
85 if _, err := c1.Write([]byte("x")); err == nil {
86 t.Fatalf("expected write on closed conn to fail")
87 }
88}
89
90func TestTransportLargeFrameNearCap(t *testing.T) {
91 c1, c2 := net.Pipe()
92 defer func() { _ = c1.Close() }()
93 defer func() { _ = c2.Close() }()
94
95 client := NewTransport(c1)
96 server := NewTransport(c2)
97
98 // Largest payload whose frame length (1 + len(payload)) still fits
99 // exactly at ports.MaxFrameLen.
100 payload := make([]byte, ports.MaxFrameLen-1)
101 for i := range payload {
102 payload[i] = byte(i)
103 }
104 want := ports.Frame{Type: ports.MsgOutput, Payload: payload}
105
106 var wg sync.WaitGroup
107 wg.Go(func() {
108 if err := client.Send(want); err != nil {
109 t.Errorf("client.Send() error = %v", err)
110 }
111 })
112
113 got, err := server.Recv()
114 wg.Wait() // ensure the sender goroutine has fully returned before any Fatalf
115 if err != nil {
116 t.Fatalf("server.Recv() error = %v", err)
117 }
118 if got.Type != want.Type || len(got.Payload) != len(want.Payload) {
119 t.Fatalf("server.Recv() = type %v len %d, want type %v len %d", got.Type, len(got.Payload), want.Type, len(want.Payload))
120 }
121 for i := range got.Payload {
122 if got.Payload[i] != want.Payload[i] {
123 t.Fatalf("payload mismatch at byte %d: got %d want %d", i, got.Payload[i], want.Payload[i])
124 }
125 }
126}
127
128func TestTransportManySmallFramesBackToBack(t *testing.T) {
129 c1, c2 := net.Pipe()
130 defer func() { _ = c1.Close() }()
131 defer func() { _ = c2.Close() }()
132
133 client := NewTransport(c1)
134 server := NewTransport(c2)
135
136 const count = 200
137 frames := make([]ports.Frame, count)
138 for i := range frames {
139 frames[i] = ports.Frame{Type: ports.MsgPing, Payload: nil}
140 }
141 // Vary a couple to prove ordering and content are both preserved.
142 frames[7] = ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{Data: []byte("hop")})}
143 frames[150] = ports.Frame{Type: ports.MsgResize, Payload: mustMarshalResize(ports.Resize{Size: domain.Size{Cols: 5, Rows: 6}})}
144
145 var wg sync.WaitGroup
146 wg.Go(func() {
147 for _, f := range frames {
148 if err := client.Send(f); err != nil {
149 t.Errorf("client.Send() error = %v", err)
150 return
151 }
152 }
153 })
154
155 // Report mismatches with Errorf (not Fatalf) and keep draining: the
156 // sender goroutine's net.Pipe writes block until read, so bailing out
157 // early here would deadlock (or leak) the sender.
158 for i, want := range frames {
159 got, err := server.Recv()
160 if err != nil {
161 t.Errorf("server.Recv() frame %d error = %v", i, err)
162 continue
163 }
164 if !reflect.DeepEqual(got, want) {
165 t.Errorf("server.Recv() frame %d = %#v, want %#v", i, got, want)
166 }
167 }
168 wg.Wait()
169}
170
171func TestTransportRecvOversizeFrame(t *testing.T) {
172 c1, c2 := net.Pipe()
173 defer func() { _ = c1.Close() }()
174 defer func() { _ = c2.Close() }()
175
176 server := NewTransport(c2)
177
178 go func() {
179 var hdr [4]byte
180 binary.BigEndian.PutUint32(hdr[:], ports.MaxFrameLen+1)
181 _, _ = c1.Write(hdr[:])
182 }()
183
184 _, err := server.Recv()
185 if !errors.Is(err, ErrFrameTooLarge) {
186 t.Fatalf("server.Recv() error = %v, want ErrFrameTooLarge", err)
187 }
188}
189
190func TestTransportRecvZeroLengthFrame(t *testing.T) {
191 c1, c2 := net.Pipe()
192 defer func() { _ = c1.Close() }()
193 defer func() { _ = c2.Close() }()
194
195 server := NewTransport(c2)
196
197 go func() {
198 var hdr [4]byte
199 binary.BigEndian.PutUint32(hdr[:], 0)
200 _, _ = c1.Write(hdr[:])
201 }()
202
203 _, err := server.Recv()
204 if !errors.Is(err, ErrZeroLengthFrame) {
205 t.Fatalf("server.Recv() error = %v, want ErrZeroLengthFrame", err)
206 }
207}
208
209func TestTransportRecvTruncatedHeader(t *testing.T) {
210 c1, c2 := net.Pipe()
211 server := NewTransport(c2)
212
213 go func() {
214 _, _ = c1.Write([]byte{0x00, 0x01}) // only 2 of 4 header bytes
215 _ = c1.Close()
216 }()
217
218 _, err := server.Recv()
219 if err == nil {
220 t.Fatalf("server.Recv() expected error on truncated header, got nil")
221 }
222}
223
224func TestTransportSendOversizePayloadRejected(t *testing.T) {
225 c1, c2 := net.Pipe()
226 defer func() { _ = c1.Close() }()
227 defer func() { _ = c2.Close() }()
228
229 client := NewTransport(c1)
230 payload := make([]byte, ports.MaxFrameLen) // +1 byte type => exceeds max
231 err := client.Send(ports.Frame{Type: ports.MsgOutput, Payload: payload})
232 if !errors.Is(err, ErrFrameTooLarge) {
233 t.Fatalf("client.Send() error = %v, want ErrFrameTooLarge", err)
234 }
235}
236
237func TestTransportAsyncEgressPreservesWelcomeBeforeOutput(t *testing.T) {
238 const timeout = time.Second
239
240 serverConn, clientConn := net.Pipe()
241 defer func() { _ = serverConn.Close() }()
242 defer func() { _ = clientConn.Close() }()
243 if err := serverConn.SetDeadline(time.Now().Add(timeout)); err != nil {
244 t.Fatalf("SetDeadline server connection: %v", err)
245 }
246 if err := clientConn.SetDeadline(time.Now().Add(timeout)); err != nil {
247 t.Fatalf("SetDeadline client connection: %v", err)
248 }
249
250 server := NewTransport(serverConn)
251 defer closeTransport(t, server)
252 client := NewTransport(clientConn)
253 defer closeTransport(t, client)
254 async, ok := server.(ports.AsyncTransport)
255 if !ok {
256 t.Fatal("IPC transport does not implement AsyncTransport")
257 }
258
259 welcome := ports.Frame{Type: ports.MsgWelcome, Payload: []byte("welcome")}
260 sent := make(chan error, 1)
261 go func() { sent <- server.Send(welcome) }()
262 got, err := client.Recv()
263 if err != nil {
264 t.Fatalf("Recv welcome: %v", err)
265 }
266 if !reflect.DeepEqual(got, welcome) {
267 t.Fatalf("welcome = %#v, want %#v", got, welcome)
268 }
269 select {
270 case err := <-sent:
271 if err != nil {
272 t.Fatalf("Send welcome: %v", err)
273 }
274 case <-time.After(timeout):
275 t.Fatal("Send welcome did not return")
276 }
277
278 payload := []byte("first")
279 if err := async.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: payload}); err != nil {
280 t.Fatalf("SendAsync first output: %v", err)
281 }
282 payload[0] = 'X' // async ownership must not retain caller memory.
283 if err := async.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("second")}); err != nil {
284 t.Fatalf("SendAsync second output: %v", err)
285 }
286 for _, want := range []ports.Frame{
287 {Type: ports.MsgOutput, Payload: []byte("first")},
288 {Type: ports.MsgOutput, Payload: []byte("second")},
289 } {
290 got, err := client.Recv()
291 if err != nil {
292 t.Fatalf("Recv output: %v", err)
293 }
294 if !reflect.DeepEqual(got, want) {
295 t.Fatalf("output = %#v, want %#v", got, want)
296 }
297 }
298}
299
300func TestTransportSendWaitsForEgressCapacity(t *testing.T) {
301 const timeout = time.Second
302
303 serverConn, clientConn := net.Pipe()
304 defer func() { _ = serverConn.Close() }()
305 defer func() { _ = clientConn.Close() }()
306 if err := clientConn.SetDeadline(time.Now().Add(timeout)); err != nil {
307 t.Fatalf("SetDeadline client connection: %v", err)
308 }
309
310 server := NewTransport(serverConn)
311 defer closeTransport(t, server)
312 transport, ok := server.(*unixTransport)
313 if !ok {
314 t.Fatal("IPC transport is not a unixTransport")
315 }
316 client := NewTransport(clientConn)
317 defer closeTransport(t, client)
318 async, ok := server.(ports.AsyncTransport)
319 if !ok {
320 t.Fatal("IPC transport does not implement AsyncTransport")
321 }
322
323 queued := fillAsyncEgress(t, transport, async, timeout)
324 want := ports.Frame{Type: ports.MsgOutput, Payload: []byte("synchronous")}
325 sent := make(chan error, 1)
326 go func() { sent <- server.Send(want) }()
327 waitForEgressSender(t, transport, timeout)
328
329 select {
330 case err := <-sent:
331 t.Fatalf("Send returned before capacity was available: %v", err)
332 default:
333 }
334
335 for _, want := range append(queued, want) {
336 got, err := client.Recv()
337 if err != nil {
338 t.Fatalf("Recv: %v", err)
339 }
340 if !reflect.DeepEqual(got, want) {
341 t.Fatalf("Recv = %#v, want %#v", got, want)
342 }
343 }
344 select {
345 case err := <-sent:
346 if err != nil {
347 t.Fatalf("Send after draining: %v", err)
348 }
349 case <-time.After(timeout):
350 t.Fatal("Send did not return after draining")
351 }
352}
353
354func TestTransportCloseInterruptsSendWaitingForEgressCapacity(t *testing.T) {
355 const timeout = time.Second
356
357 serverConn, clientConn := net.Pipe()
358 defer func() { _ = clientConn.Close() }()
359
360 server := NewTransport(serverConn)
361 defer closeTransport(t, server)
362 transport, ok := server.(*unixTransport)
363 if !ok {
364 t.Fatal("IPC transport is not a unixTransport")
365 }
366 async, ok := server.(ports.AsyncTransport)
367 if !ok {
368 t.Fatal("IPC transport does not implement AsyncTransport")
369 }
370 fillAsyncEgress(t, transport, async, timeout)
371
372 sent := make(chan error, 1)
373 go func() { sent <- server.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte("synchronous")}) }()
374 waitForEgressSender(t, transport, timeout)
375
376 closeTransport(t, server)
377 select {
378 case err := <-sent:
379 if !errors.Is(err, errClosed) {
380 t.Fatalf("Send after Close = %v, want errClosed", err)
381 }
382 case <-time.After(timeout):
383 t.Fatal("Close did not interrupt Send")
384 }
385}
386
387func waitForEgressSender(t *testing.T, transport *unixTransport, timeout time.Duration) {
388 t.Helper()
389
390 timer := time.NewTimer(timeout)
391 defer timer.Stop()
392 for {
393 transport.egressMu.Lock()
394 senders := transport.egressSenders
395 transport.egressMu.Unlock()
396 if senders > 0 {
397 return
398 }
399 select {
400 case <-timer.C:
401 t.Fatal("Send did not register as an egress capacity waiter")
402 default:
403 runtime.Gosched()
404 }
405 }
406}
407
408func closeTransport(t *testing.T, transport ports.Transport) {
409 t.Helper()
410
411 closed := make(chan error, 1)
412 go func() { closed <- transport.Close() }()
413 select {
414 case err := <-closed:
415 if err != nil {
416 t.Fatalf("Close: %v", err)
417 }
418 case <-time.After(time.Second):
419 t.Fatal("Close did not return")
420 }
421}
422
423func fillAsyncEgress(t *testing.T, transport *unixTransport, async ports.AsyncTransport, timeout time.Duration) []ports.Frame {
424 t.Helper()
425
426 first := ports.Frame{Type: ports.MsgOutput, Payload: []byte{0}}
427 if err := async.SendAsync(first); err != nil {
428 t.Fatalf("SendAsync blocked frame: %v", err)
429 }
430 waitForEgressDequeue(t, transport, timeout)
431
432 queued := []ports.Frame{first}
433 for i := 1; ; i++ {
434 frame := ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
435 err := async.SendAsync(frame)
436 switch {
437 case err == nil:
438 queued = append(queued, frame)
439 case errors.Is(err, ErrBackpressure):
440 return queued
441 default:
442 t.Fatalf("SendAsync queued frame: %v", err)
443 }
444 }
445}
446
447func waitForEgressDequeue(t *testing.T, transport *unixTransport, timeout time.Duration) {
448 t.Helper()
449
450 timer := time.NewTimer(timeout)
451 defer timer.Stop()
452 for {
453 if len(transport.egress) == 0 {
454 return
455 }
456 select {
457 case <-timer.C:
458 t.Fatal("IPC writer did not dequeue the blocked frame")
459 default:
460 runtime.Gosched()
461 }
462 }
463}
464
465func TestTransportAsyncEgressIsBoundedAndCloseInterruptsWorkers(t *testing.T) {
466 serverConn, clientConn := net.Pipe()
467 defer func() { _ = clientConn.Close() }()
468
469 tr := NewTransport(serverConn)
470 defer closeTransport(t, tr)
471 transport, ok := tr.(*unixTransport)
472 if !ok {
473 t.Fatal("IPC transport is not a unixTransport")
474 }
475 async, ok := tr.(ports.AsyncTransport)
476 if !ok {
477 t.Fatal("IPC transport does not implement AsyncTransport")
478 }
479
480 // The first write blocks in net.Pipe because the peer never drains it.
481 if err := async.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("blocked")}); err != nil {
482 t.Fatalf("SendAsync blocked frame: %v", err)
483 }
484 var backpressure error
485 for range sendQueueCapacity + 2 {
486 err := async.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")})
487 if errors.Is(err, ErrBackpressure) {
488 backpressure = err
489 break
490 }
491 if err != nil {
492 t.Fatalf("SendAsync unexpected error: %v", err)
493 }
494 }
495 if !errors.Is(backpressure, ErrBackpressure) {
496 t.Fatalf("SendAsync never reported bounded backpressure, got %v", backpressure)
497 }
498
499 recvDone := make(chan error, 1)
500 go func() {
501 _, err := tr.Recv()
502 recvDone <- err
503 }()
504 closeTransport(t, tr)
505 select {
506 case err := <-recvDone:
507 if err == nil {
508 t.Fatal("Recv succeeded after Close")
509 }
510 case <-time.After(time.Second):
511 t.Fatal("Close did not interrupt Recv")
512 }
513 select {
514 case <-transport.writerDone:
515 default:
516 t.Fatal("Close returned before the IPC writer terminated")
517 }
518}