1package dgram
2
3import (
4 "net"
5 "sync"
6 "testing"
7 "time"
8
9 "github.com/bnema/vev/internal/ports"
10)
11
12// gatedControlPC stops the first control write after sendControl has selected
13// its socket and installed its deadline. Close is separately gated so the test
14// can observe whether a hop attempts to retire that socket too early.
15type gatedControlPC struct {
16 *fakePC
17 entered chan struct{}
18 releaseWrite chan struct{}
19 closeStarted chan struct{}
20 releaseClose chan struct{}
21 gateWrite sync.Once
22 gateClose sync.Once
23}
24
25func (p *gatedControlPC) SetWriteDeadline(time.Time) error {
26 p.gateWrite.Do(func() {
27 close(p.entered)
28 <-p.releaseWrite
29 })
30 return nil
31}
32
33func (p *gatedControlPC) Close() error {
34 p.gateClose.Do(func() {
35 close(p.closeStarted)
36 <-p.releaseClose
37 })
38 return p.fakePC.Close()
39}
40
41func TestACKWriteCompletesBeforeRebindRetiresPacketConn(t *testing.T) {
42 old, bPC := newPair()
43 gated := &gatedControlPC{
44 fakePC: old,
45 entered: make(chan struct{}),
46 releaseWrite: make(chan struct{}),
47 closeStarted: make(chan struct{}),
48 releaseClose: make(chan struct{}),
49 }
50 var releaseWrite, releaseClose sync.Once
51 release := func() {
52 releaseWrite.Do(func() { close(gated.releaseWrite) })
53 releaseClose.Do(func() { close(gated.releaseClose) })
54 }
55
56 var fresh *fakePC
57 a, err := NewTransportWithOptions(gated, bPC.addr, key(), 1, 2, Options{
58 ResendAfter: time.Hour,
59 Heartbeat: time.Hour,
60 RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
61 fresh = &fakePC{
62 addr: testAddr("a2"),
63 in: make(chan packet, 10),
64 read: make(chan struct{}, 10),
65 peers: map[string]*fakePC{"b": bPC},
66 }
67 bPC.peers["a2"] = fresh
68 return fresh, nil
69 },
70 })
71 if err != nil {
72 t.Fatal(err)
73 }
74 b, err := NewTransportWithOptions(bPC, old.addr, key(), 2, 1, Options{
75 ResendAfter: time.Hour,
76 Heartbeat: time.Hour,
77 })
78 if err != nil {
79 release()
80 _ = a.Close()
81 t.Fatal(err)
82 }
83 defer func() {
84 release()
85 _ = a.Close()
86 _ = b.Close()
87 }()
88
89 // A decoded ACK clears this pending record, proving delivery to the peer,
90 // rather than merely observing a write on the old PacketConn.
91 b.mu.Lock()
92 b.pending[7] = &pending{}
93 processed := make(chan struct{}, 1)
94 b.afterPacketProcessed = func() { processed <- struct{}{} }
95 b.mu.Unlock()
96
97 a.mu.Lock()
98 a.linkState = ports.LinkStateProbing
99 a.health.generation = 1
100 a.mu.Unlock()
101 controlDone := make(chan error, 1)
102 go func() { controlDone <- a.sendControl(recAck, 7) }()
103 awaitSignal(t, gated.entered, "ACK control write deadline")
104
105 hopped := make(chan struct{})
106 go func() {
107 a.hopPacketConnOnce(1)
108 close(hopped)
109 }()
110
111 // A hop must wait for the control write, even though this ACK does not wait
112 // for data's writeMu. Before the write is released it must not retire old.
113 select {
114 case <-gated.closeStarted:
115 t.Fatal("rebind began closing packet conn during control write")
116 case <-hopped:
117 t.Fatal("rebind completed during control write")
118 case <-time.After(25 * time.Millisecond):
119 }
120
121 releaseWrite.Do(func() { close(gated.releaseWrite) })
122 if err := awaitResult(t, controlDone, "ACK control write"); err != nil {
123 t.Fatalf("ACK control write: %v", err)
124 }
125 awaitSignal(t, processed, "decoded ACK delivery")
126 b.mu.Lock()
127 _, pending := b.pending[7]
128 b.mu.Unlock()
129 if pending {
130 t.Fatal("decoded ACK did not clear peer pending record")
131 }
132
133 awaitSignal(t, gated.closeStarted, "rebind retires old packet conn")
134 releaseClose.Do(func() { close(gated.releaseClose) })
135 select {
136 case <-hopped:
137 case <-time.After(time.Second):
138 t.Fatal("rebind did not complete after control write")
139 }
140 if fresh == nil {
141 t.Fatal("rebind did not install fresh packet conn")
142 }
143}