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