1package dgram
2
3import (
4 "encoding/binary"
5 "errors"
6 "net"
7 "testing"
8 "time"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 pdgram "github.com/bnema/vev/pkg/dgram"
13 "github.com/stretchr/testify/require"
14)
15
16func TestPendingWireBytesMatchSealedDatagrams(t *testing.T) {
17 const mtu = 128
18 aPC, bPC := newPair()
19 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
20 MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
21 })
22 if err != nil {
23 t.Fatal(err)
24 }
25 defer func() { _ = a.Close() }()
26
27 maxFragmentData := mtu - pdgram.HeaderSize - a.codec.Overhead() - 16
28 frame := ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, maxFragmentData-dataRecordHeaderSize+1)}
29 if err := a.Send(frame); err != nil {
30 t.Fatal(err)
31 }
32
33 want := 0
34 for len(bPC.in) > 0 {
35 want += len((<-bPC.in).b)
36 }
37 a.mu.Lock()
38 got := a.pending[1].wireBytes
39 initialCwnd := a.congestion.cwndBytes
40 a.mu.Unlock()
41 if got != want {
42 t.Fatalf("pending wire bytes=%d, want sealed datagram bytes=%d", got, want)
43 }
44
45 ack := make([]byte, 9)
46 ack[0] = recAck
47 binary.BigEndian.PutUint64(ack[1:], 1)
48 a.handleRecord(ack)
49 a.mu.Lock()
50 gotCwnd := a.congestion.cwndBytes
51 a.mu.Unlock()
52 wantCwnd := initialCwnd + max(1, mtu*want/initialCwnd)
53 if gotCwnd != wantCwnd {
54 t.Fatalf("cwnd after ACK=%d, want %d from %d acknowledged sealed bytes", gotCwnd, wantCwnd, want)
55 }
56}
57
58func TestSendAsyncTransfersFragmentedRecordToBytePacedSender(t *testing.T) {
59 const mtu = 128
60 aPC, bPC := newPair()
61 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
62 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
63 Clock: clk, MTU: mtu, ResendAfter: time.Hour, Heartbeat: time.Hour,
64 })
65 if err != nil {
66 t.Fatal(err)
67 }
68 defer func() { _ = a.Close() }()
69
70 maxFragmentData := mtu - pdgram.HeaderSize - a.codec.Overhead()
71 frame := ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 2*maxFragmentData)}
72 if err := a.SendAsync(frame); err != nil {
73 t.Fatal(err)
74 }
75 eventuallyPacketCount(t, bPC, dataBurstPackets)
76 if got := len(bPC.in); got != dataBurstPackets {
77 t.Fatalf("immediate datagrams=%d, want %d MTU burst", got, dataBurstPackets)
78 }
79 awaitSignal(t, clk.timerCreated, "shared data sender pacing timer")
80 clk.advance(initialPacingRTT)
81 eventuallyPacketCount(t, bPC, 3)
82}
83
84func TestSmallOutputsShareInitialByteBurst(t *testing.T) {
85 aPC, bPC := newPair()
86 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
87 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
88 if err != nil {
89 t.Fatal(err)
90 }
91 defer func() { _ = a.Close() }()
92
93 for i := range 3 {
94 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
95 t.Fatal(err)
96 }
97 }
98 eventuallyPacketCount(t, bPC, 3)
99 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}); err != nil {
100 t.Fatal(err)
101 }
102 eventuallyPacketCount(t, bPC, 4)
103}
104
105func TestOutputPacingSendsAtMostOneOversizedFramePerTick(t *testing.T) {
106 aPC, _ := newPair()
107 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
108 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
109 if err != nil {
110 t.Fatal(err)
111 }
112 defer func() { _ = a.Close() }()
113
114 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("prime")}); err != nil {
115 t.Fatal(err)
116 }
117 large := mustMarshalOutput(ports.Output{Epoch: 1, Base: 1, New: 2, Size: domain.Size{Cols: 1, Rows: 1}, Data: make([]byte, 24*a.mtu)})
118 for range 2 {
119 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: large}); err != nil {
120 t.Fatal(err)
121 }
122 }
123 waitForManualTimers(t, clk, 4)
124 previousPackets := len(aPC.in)
125 for tick := range 128 {
126 clk.advance(initialPacingRTT)
127 time.Sleep(time.Millisecond)
128 packets := len(aPC.in)
129 if delta := packets - previousPackets; delta > initialCongestionPackets {
130 t.Fatalf("tick %d emitted %d packets, cwnd packets %d", tick, delta, initialCongestionPackets)
131 }
132 previousPackets = packets
133 a.mu.Lock()
134 last := a.pending[3]
135 done := len(a.outputQueue) == 0 && last != nil && !last.initialInFlight
136 a.mu.Unlock()
137 if done {
138 return
139 }
140 }
141 t.Fatal("oversized queued frames did not complete within pacing ticks")
142}
143
144func TestPacedInitialSendTimestampsFinalFragmentCompletion(t *testing.T) {
145 aPC, _ := newPair()
146 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
147 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
148 Clock: clk, ResendAfter: 500 * time.Millisecond, MaxResendAfter: time.Second, Heartbeat: time.Hour,
149 })
150 if err != nil {
151 t.Fatal(err)
152 }
153 defer func() { _ = a.Close() }()
154 large := ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 24*a.mtu)}
155 done := make(chan error, 1)
156 go func() { done <- a.Send(large) }()
157 start := clk.Now()
158 for range 128 {
159 a.mu.Lock()
160 p := a.pending[1]
161 if p != nil && p.initialInFlight && !p.last.IsZero() {
162 a.mu.Unlock()
163 t.Fatal("pending last timestamp set before final fragment")
164 }
165 a.mu.Unlock()
166 clk.advance(initialPacingRTT)
167 time.Sleep(time.Millisecond)
168 select {
169 case err := <-done:
170 if err != nil {
171 t.Fatal(err)
172 }
173 a.mu.Lock()
174 completed := a.pending[1].last
175 a.mu.Unlock()
176 if elapsed := completed.Sub(start); elapsed <= initialPacingRTT {
177 t.Fatalf("large send completed too early after %v", elapsed)
178 }
179 before := len(aPC.in)
180 clk.advance(499 * time.Millisecond)
181 a.resendPending()
182 if got := len(aPC.in); got != before {
183 t.Fatalf("resend emitted before one RTO after final fragment: %d -> %d", before, got)
184 }
185 a.handleRecord(ackRecord(1))
186 a.mu.Lock()
187 srtt := a.srtt
188 a.mu.Unlock()
189 require.Equal(t, 499*time.Millisecond, srtt, "RTT sample must start at final-fragment completion")
190 return
191 default:
192 }
193 }
194 t.Fatal("large initial send did not complete")
195}
196
197func TestOwnedSynchronousMaxSideEffectCompletesThroughConcurrentPacedWork(t *testing.T) {
198 aPC, bPC := newPair()
199 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
200 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
201 Clock: clk, ResendAfter: 250 * time.Millisecond, MaxResendAfter: time.Second, Heartbeat: time.Hour,
202 })
203 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
204 defer func() { _ = a.Close() }()
205 defer func() { _ = b.Close() }()
206 a.mu.Lock()
207 a.srtt = 500 * time.Millisecond
208 a.mu.Unlock()
209 for i := range 8 {
210 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
211 t.Fatal(err)
212 }
213 }
214 now := clk.Now()
215 a.mu.Lock()
216 for _, seq := range []uint64{100, 101} {
217 a.pending[seq] = &pending{
218 frame: ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 20*a.mtu)},
219 enqueued: now.Add(-time.Second), first: now.Add(-time.Second), last: now.Add(-time.Second),
220 }
221 }
222 a.mu.Unlock()
223 go a.resendPending()
224 sideEffect := ports.Frame{Type: ports.MsgOutput, Payload: mustMarshalOutput(ports.Output{Epoch: 1, Size: domain.Size{Cols: 1, Rows: 1}, Data: make([]byte, 100*1024)})}
225 done := make(chan error, 1)
226 go func() { done <- a.SendSynchronous(sideEffect) }()
227
228 for range 512 {
229 clk.advance(initialPacingRTT)
230 time.Sleep(time.Millisecond)
231 select {
232 case err := <-done:
233 if err != nil {
234 t.Fatal(err)
235 }
236 if a.isClosed() {
237 t.Fatal("healthy owned synchronous send force-closed transport")
238 }
239 return
240 default:
241 }
242 }
243 t.Fatal("maximum side effect did not complete through queued/retransmit pacing")
244}
245
246func TestInputWaitsForDequeuedOutputBatch(t *testing.T) {
247 aPC, bPC := newPair()
248 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
249 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
250 if err != nil {
251 t.Fatal(err)
252 }
253 defer func() { _ = a.Close() }()
254
255 a.writeMu.Lock()
256 for _, payload := range []string{"first", "queued"} {
257 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte(payload)}); err != nil {
258 t.Fatal(err)
259 }
260 }
261 done := make(chan error, 1)
262 go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
263 select {
264 case err := <-done:
265 a.writeMu.Unlock()
266 t.Fatalf("input completed before dequeued output batch: %v", err)
267 case <-time.After(20 * time.Millisecond):
268 }
269 a.writeMu.Unlock()
270 select {
271 case err := <-done:
272 if err != nil {
273 t.Fatal(err)
274 }
275 case <-time.After(time.Second):
276 t.Fatal("input did not complete after output batch was released")
277 }
278 if got := len(bPC.in); got != 3 {
279 t.Fatalf("datagrams when input completed=%d, want output batch before input", got)
280 }
281}
282
283func TestPacedControlQueueFailureClosesTransport(t *testing.T) {
284 aPC, bPC := newPair()
285 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
286 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
287 if err != nil {
288 t.Fatal(err)
289 }
290 defer func() { _ = a.Close() }()
291
292 if err := bPC.Close(); err != nil {
293 t.Fatal(err)
294 }
295 inputDone := make(chan error, 1)
296 go func() { inputDone <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
297 if err := <-inputDone; err == nil {
298 t.Fatal("data send succeeded after paced wire failure")
299 }
300 select {
301 case <-a.done:
302 case <-time.After(time.Second):
303 t.Fatal("transport remained open after paced control queue write failed")
304 }
305}
306
307func TestRemoveQueuedPendingResolvesSynchronousWaiters(t *testing.T) {
308 tr, _ := newResendTestTransport(t, 10)
309 wantErr := errors.New("paced send failed")
310 done := make(chan error, 1)
311
312 tr.mu.Lock()
313 tr.pending[7] = &pending{}
314 tr.mu.Unlock()
315 tr.removeQueuedPending([]queuedSend{{seq: 7, reliable: true, done: done}}, wantErr)
316
317 require.ErrorIs(t, <-done, wantErr)
318 _, ok := <-done
319 require.False(t, ok, "completion channel must be closed")
320 tr.mu.Lock()
321 _, pending := tr.pending[7]
322 tr.mu.Unlock()
323 require.False(t, pending)
324}
325
326func TestCloseUnblocksQueuedSynchronousFrames(t *testing.T) {
327 frames := []struct {
328 name string
329 frame ports.Frame
330 }{
331 {name: "control", frame: ports.Frame{Type: ports.MsgPing}},
332 {name: "terminal side effect", frame: ports.Frame{Type: ports.MsgOutput, Payload: mustMarshalOutput(ports.Output{Epoch: 1, Size: domain.Size{Cols: 1, Rows: 1}, Data: []byte("osc")})}},
333 }
334 for _, tt := range frames {
335 t.Run(tt.name, func(t *testing.T) {
336 aPC, _ := newPair()
337 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
338 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
339 if err != nil {
340 t.Fatal(err)
341 }
342 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: make([]byte, 3*a.mtu)}); err != nil {
343 t.Fatal(err)
344 }
345 done := make(chan error, 1)
346 go func() { done <- a.Send(tt.frame) }()
347 eventually(t, time.Second, func() bool {
348 a.mu.Lock()
349 defer a.mu.Unlock()
350 return len(a.pending) == 2
351 })
352 _ = a.Close()
353 if err := <-done; err == nil {
354 t.Fatal("queued synchronous Send returned nil after close")
355 }
356 })
357 }
358}
359
360func TestOutputPaceFailureClosesTransportWithWriteError(t *testing.T) {
361 aPC, bPC := newPair()
362 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
363 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
364 if err != nil {
365 t.Fatal(err)
366 }
367 defer func() { _ = a.Close() }()
368
369 if err := bPC.Close(); err != nil {
370 t.Fatal(err)
371 }
372 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("fails")}); err != nil {
373 t.Fatal(err)
374 }
375
376 select {
377 case <-a.done:
378 case <-time.After(time.Second):
379 t.Fatal("transport remained open after paced output write failed")
380 }
381 if err := a.Send(ports.Frame{Type: ports.MsgInput}); err == nil || err.Error() != "peer closed" {
382 t.Fatalf("send after paced write failure err=%v, want peer closed", err)
383 }
384}
385
386func TestInputAndControlRemainOrderedInBoundedPacedQueue(t *testing.T) {
387 aPC, bPC := newPair()
388 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
389 a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour})
390 if err != nil {
391 t.Fatal(err)
392 }
393 defer func() { _ = a.Close() }()
394
395 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("output")}); err != nil {
396 t.Fatal(err)
397 }
398 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued")}); err != nil {
399 t.Fatal(err)
400 }
401 eventuallyPacketCount(t, bPC, 2)
402 inputDone := make(chan error, 1)
403 go func() { inputDone <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("typed")}) }()
404 eventuallyPacketCount(t, bPC, 3)
405 if err := <-inputDone; err != nil {
406 t.Fatal(err)
407 }
408 if err := a.SendAsync(ports.Frame{Type: ports.MsgOutput, Payload: []byte("queued-before-control")}); err != nil {
409 t.Fatal(err)
410 }
411 controlDone := make(chan error, 1)
412 go func() { controlDone <- a.Send(ports.Frame{Type: ports.MsgPing}) }()
413 eventuallyPacketCount(t, bPC, 5)
414 if err := <-controlDone; err != nil {
415 t.Fatal(err)
416 }
417}
418
419func TestSendWriteTimeoutBoundsBlockedPacketConn(t *testing.T) {
420 pc := newBlockingWritePC()
421 tr, err := NewTransportWithOptions(pc, testAddr("peer"), key(), 1, 2, Options{
422 WriteTimeout: 10 * time.Millisecond,
423 ResendAfter: time.Hour,
424 Heartbeat: time.Hour,
425 })
426 if err != nil {
427 t.Fatal(err)
428 }
429 defer func() { _ = tr.Close() }()
430
431 done := make(chan error, 1)
432 go func() { done <- tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("blocked")}) }()
433 select {
434 case err := <-done:
435 if err == nil {
436 t.Fatal("Send returned nil, want write timeout")
437 }
438 var timeout interface{ Timeout() bool }
439 if !errors.As(err, &timeout) || !timeout.Timeout() {
440 t.Fatalf("Send err=%v, want timeout error", err)
441 }
442 tr.mu.Lock()
443 pending := len(tr.pending)
444 tr.mu.Unlock()
445 if pending != 0 {
446 t.Fatalf("pending after failed initial send=%d, want 0", pending)
447 }
448 case <-time.After(250 * time.Millisecond):
449 t.Fatal("Send blocked without honoring WriteTimeout")
450 }
451}
452
453func TestPendingReliableQueueBackpressuresUntilAck(t *testing.T) {
454 aPC, bPC := newPair()
455 aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
456 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1})
457 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
458 defer func() { _ = a.Close() }()
459 defer func() { _ = b.Close() }()
460
461 if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
462 t.Fatal(err)
463 }
464 done := make(chan error, 1)
465 go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
466
467 select {
468 case err := <-done:
469 t.Fatalf("send returned before an ack released pending capacity: %v", err)
470 case <-time.After(25 * time.Millisecond):
471 }
472
473 a.handleRecord(ackRecord(1))
474 select {
475 case err := <-done:
476 if err != nil {
477 t.Fatalf("send after ack: %v", err)
478 }
479 case <-time.After(time.Second):
480 t.Fatal("send did not resume after ack released pending capacity")
481 }
482 a.mu.Lock()
483 pending := len(a.pending)
484 a.mu.Unlock()
485 if pending > 1 {
486 t.Fatalf("pending=%d, want <= 1", pending)
487 }
488}
489
490func TestPendingReliableQueueReturnsErrPendingFullWhileConnected(t *testing.T) {
491 aPC, bPC := newPair()
492 aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
493 clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
494 a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{MaxPending: 1, MaxPendingWait: 50 * time.Millisecond, Clock: clk})
495 b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
496 defer func() { _ = a.Close() }()
497 defer func() { _ = b.Close() }()
498
499 if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("first")}); err != nil {
500 t.Fatal(err)
501 }
502 done := make(chan error, 1)
503 go func() { done <- a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("second")}) }()
504 waitForManualTimers(t, clk, 4)
505
506 clk.advance(50 * time.Millisecond)
507 select {
508 case err := <-done:
509 if !errors.Is(err, ErrPendingFull) {
510 t.Fatalf("send err=%v, want ErrPendingFull", err)
511 }
512 case <-time.After(time.Second):
513 t.Fatal("send did not return after clock advanced past MaxPendingWait")
514 }
515}
516
517func TestWritePacketDeadlineUsesTransportClock(t *testing.T) {
518 wantNow := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
519 pc := &deadlineCapturePC{addr: testAddr("a"), done: make(chan struct{})}
520 tr, err := NewTransportWithOptions(pc, testAddr("b"), key(), 1, 2, Options{
521 Clock: fixedClock{now: wantNow},
522 WriteTimeout: 250 * time.Millisecond,
523 ResendAfter: time.Hour,
524 Heartbeat: time.Hour,
525 })
526 if err != nil {
527 t.Fatal(err)
528 }
529 defer func() { _ = tr.Close() }()
530
531 if err := tr.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte("deadline")}); err != nil {
532 t.Fatal(err)
533 }
534 got, _ := pc.deadline.Load().(time.Time)
535 if want := wantNow.Add(250 * time.Millisecond); !got.Equal(want) {
536 t.Fatalf("write deadline=%v, want transport clock deadline %v", got, want)
537 }
538}