1package ipc
2
3import (
4 "encoding/binary"
5 "errors"
6 "io"
7 "net"
8 "reflect"
9 "sync"
10 "testing"
11 "time"
12
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/protocol"
15 "github.com/bnema/vev/internal/protocol/wire"
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 := wire.Envelope{Payload: mustEncodeClient(protocol.Hello{Version: 1, Intent: protocol.IntentNew, Name: "w0"})}
27 serverToClient := wire.Envelope{Payload: mustEncodeServer(protocol.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 accepted: the canonical envelope ceiling admits a
99 // payload of exactly wire.AbsoluteEnvelopeLimit, so one byte below stays near cap.
100 payload := make([]byte, wire.AbsoluteEnvelopeLimit-1)
101 for i := range payload {
102 payload[i] = byte(i)
103 }
104 want := wire.Envelope{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 len(got.Payload) != len(want.Payload) {
119 t.Fatalf("server.Recv() = len %d, want len %d", len(got.Payload), 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([]wire.Envelope, count)
138 for i := range frames {
139 frames[i] = wire.Envelope{Payload: []byte("ping")}
140 }
141 // Vary a couple to prove ordering and content are both preserved.
142 frames[7] = wire.Envelope{Payload: mustEncodeClient(protocol.Input{Data: []byte("hop")})}
143 frames[150] = wire.Envelope{Payload: mustEncodeClient(protocol.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[:], wire.AbsoluteEnvelopeLimit+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, wire.AbsoluteEnvelopeLimit+1) // one byte past the canonical ceiling
231 err := client.Send(wire.Envelope{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.(wire.AsyncTransport)
255 if !ok {
256 t.Fatal("IPC transport does not implement AsyncTransport")
257 }
258
259 welcome := wire.Envelope{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(wire.Envelope{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(wire.Envelope{Payload: []byte("second")}); err != nil {
284 t.Fatalf("SendAsync second output: %v", err)
285 }
286 for _, want := range []wire.Envelope{
287 {Payload: []byte("first")},
288 {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 client := NewTransport(clientConn)
313 defer closeTransport(t, client)
314 async, ok := server.(wire.AsyncTransport)
315 if !ok {
316 t.Fatal("IPC transport does not implement AsyncTransport")
317 }
318
319 queued := fillAsyncEgress(t, async, timeout)
320 want := wire.Envelope{Payload: []byte("synchronous")}
321 sent := make(chan error, 1)
322 go func() { sent <- server.Send(want) }()
323 // Send must park behind the full egress queue; give the goroutine a
324 // moment to reach the queue, then prove it has not returned.
325 time.Sleep(20 * time.Millisecond)
326 select {
327 case err := <-sent:
328 t.Fatalf("Send returned before capacity was available: %v", err)
329 default:
330 }
331
332 for _, want := range append(queued, want) {
333 got, err := client.Recv()
334 if err != nil {
335 t.Fatalf("Recv: %v", err)
336 }
337 if !reflect.DeepEqual(got, want) {
338 t.Fatalf("Recv = %#v, want %#v", got, want)
339 }
340 }
341 select {
342 case err := <-sent:
343 if err != nil {
344 t.Fatalf("Send after draining: %v", err)
345 }
346 case <-time.After(timeout):
347 t.Fatal("Send did not return after draining")
348 }
349}
350
351func TestTransportCloseInterruptsSendWaitingForEgressCapacity(t *testing.T) {
352 const timeout = time.Second
353
354 serverConn, clientConn := net.Pipe()
355 defer func() { _ = clientConn.Close() }()
356
357 server := NewTransport(serverConn)
358 defer closeTransport(t, server)
359 async, ok := server.(wire.AsyncTransport)
360 if !ok {
361 t.Fatal("IPC transport does not implement AsyncTransport")
362 }
363 fillAsyncEgress(t, async, timeout)
364
365 sent := make(chan error, 1)
366 go func() { sent <- server.Send(wire.Envelope{Payload: []byte("synchronous")}) }()
367 time.Sleep(20 * time.Millisecond)
368
369 closeTransport(t, server)
370 select {
371 case err := <-sent:
372 if !errors.Is(err, errClosed) {
373 t.Fatalf("Send after Close = %v, want errClosed", err)
374 }
375 case <-time.After(timeout):
376 t.Fatal("Close did not interrupt Send")
377 }
378}
379
380func closeTransport(t *testing.T, transport wire.Transport) {
381 t.Helper()
382
383 closed := make(chan error, 1)
384 go func() { closed <- transport.Close() }()
385 select {
386 case err := <-closed:
387 if err != nil {
388 t.Fatalf("Close: %v", err)
389 }
390 case <-time.After(time.Second):
391 t.Fatal("Close did not return")
392 }
393}
394
395// fillAsyncEgress admits envelopes until the bounded egress queue reports
396// backpressure, returning every admitted envelope in transmission order.
397func fillAsyncEgress(t *testing.T, async wire.AsyncTransport, timeout time.Duration) []wire.Envelope {
398 t.Helper()
399
400 deadline := time.Now().Add(timeout)
401 queued := make([]wire.Envelope, 0, 16)
402 for i := 0; ; i++ {
403 envelope := wire.Envelope{Payload: []byte{byte(i)}}
404 err := async.SendAsync(envelope)
405 switch {
406 case err == nil:
407 queued = append(queued, envelope)
408 case errors.Is(err, ErrBackpressure):
409 if len(queued) == 0 {
410 t.Fatal("SendAsync reported backpressure before admitting any envelope")
411 }
412 return queued
413 default:
414 t.Fatalf("SendAsync queued envelope: %v", err)
415 }
416 if time.Now().After(deadline) {
417 t.Fatal("SendAsync never reported bounded backpressure")
418 }
419 }
420}
421
422func TestTransportAsyncEgressIsBoundedAndCloseInterruptsWorkers(t *testing.T) {
423 serverConn, clientConn := net.Pipe()
424 defer func() { _ = clientConn.Close() }()
425
426 tr := NewTransport(serverConn)
427 defer closeTransport(t, tr)
428 async, ok := tr.(wire.AsyncTransport)
429 if !ok {
430 t.Fatal("IPC transport does not implement AsyncTransport")
431 }
432
433 // The first write blocks in net.Pipe because the peer never drains it.
434 if err := async.SendAsync(wire.Envelope{Payload: []byte("blocked")}); err != nil {
435 t.Fatalf("SendAsync blocked frame: %v", err)
436 }
437 var backpressure error
438 for i := 0; i < 64; i++ {
439 err := async.SendAsync(wire.Envelope{Payload: []byte("queued")})
440 if errors.Is(err, ErrBackpressure) {
441 backpressure = err
442 break
443 }
444 if err != nil {
445 t.Fatalf("SendAsync unexpected error: %v", err)
446 }
447 }
448 if !errors.Is(backpressure, ErrBackpressure) {
449 t.Fatalf("SendAsync never reported bounded backpressure, got %v", backpressure)
450 }
451
452 recvDone := make(chan error, 1)
453 go func() {
454 _, err := tr.Recv()
455 recvDone <- err
456 }()
457 closeTransport(t, tr)
458 select {
459 case err := <-recvDone:
460 if err == nil {
461 t.Fatal("Recv succeeded after Close")
462 }
463 case <-time.After(time.Second):
464 t.Fatal("Close did not interrupt Recv")
465 }
466 // Close waits for the sole writer to stop, so admission is closed for good.
467 if err := async.SendAsync(wire.Envelope{Payload: []byte("after close")}); !errors.Is(err, errClosed) {
468 t.Fatalf("SendAsync after Close = %v, want errClosed", err)
469 }
470}
471
472// TestTransportRecvBoundedEnforcesCallerLimit proves the IPC transport exposes
473// the shared bounded receive: an over-limit length prefix is refused before the
474// body is read, a zero limit refuses a non-empty frame, and an in-limit frame
475// round-trips byte for byte.
476func TestTransportRecvBoundedEnforcesCallerLimit(t *testing.T) {
477 writePrefix := func(conn net.Conn, length uint32) {
478 go func() {
479 var header [4]byte
480 binary.BigEndian.PutUint32(header[:], length)
481 _, _ = conn.Write(header[:])
482 }()
483 }
484
485 t.Run("over-limit prefix refused before body", func(t *testing.T) {
486 c1, c2 := net.Pipe()
487 defer func() { _ = c1.Close() }()
488 defer func() { _ = c2.Close() }()
489 server := NewTransport(c2).(wire.BoundedTransport)
490 writePrefix(c1, 8)
491 _, err := server.RecvBounded(4)
492 if !errors.Is(err, ErrFrameTooLarge) {
493 t.Fatalf("RecvBounded() error = %v, want ErrFrameTooLarge", err)
494 }
495 })
496
497 t.Run("zero limit refuses non-empty frame", func(t *testing.T) {
498 c1, c2 := net.Pipe()
499 defer func() { _ = c1.Close() }()
500 defer func() { _ = c2.Close() }()
501 server := NewTransport(c2).(wire.BoundedTransport)
502 writePrefix(c1, 1)
503 _, err := server.RecvBounded(0)
504 if !errors.Is(err, ErrFrameTooLarge) {
505 t.Fatalf("RecvBounded() error = %v, want ErrFrameTooLarge", err)
506 }
507 })
508
509 t.Run("in-limit frame round trips", func(t *testing.T) {
510 c1, c2 := net.Pipe()
511 defer func() { _ = c1.Close() }()
512 defer func() { _ = c2.Close() }()
513 server := NewTransport(c2).(wire.BoundedTransport)
514 payload := []byte("bounded ipc envelope")
515 go func() {
516 var header [4]byte
517 binary.BigEndian.PutUint32(header[:], uint32(len(payload)))
518 _, _ = c1.Write(append(header[:], payload...))
519 }()
520 envelope, err := server.RecvBounded(uint64(len(payload)))
521 if err != nil {
522 t.Fatalf("RecvBounded() error = %v", err)
523 }
524 if !reflect.DeepEqual(envelope.Payload, payload) {
525 t.Fatalf("RecvBounded() payload = %q, want %q", envelope.Payload, payload)
526 }
527 })
528}