detached f2100d6

test(dgram): classify UDP flood failure modes

commit
f2100d695ccf56f59c35f2e3f16136e775c52f2c
author
brice · · 2026-07-09 23:48 UTC
parent
7e5b90a
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +327 -73
Unified
UnifiedSplit
A internal/adapters/dgram/network_sim_test.go +327 -0
@@ -0,0 +1,327 @@
1+package dgram
2+
3+import (
4+ "bytes"
5+ "errors"
6+ "net"
7+ "sync"
8+ "testing"
9+ "time"
10+
11+ pdgram "github.com/bnema/vev/pkg/dgram"
12+)
13+
14+type testAddr string
15+
16+func (a testAddr) Network() string { return "mem" }
17+func (a testAddr) String() string { return string(a) }
18+
19+type packet struct {
20+ b []byte
21+ addr net.Addr
22+}
23+
24+// packetMeta is the deterministic simulator's view of a datagram before it is
25+// accepted by the link. It deliberately contains only link-level information.
26+type packetMeta struct {
27+ From, To net.Addr
28+ Bytes int
29+ At time.Time
30+ QueueBytes int
31+ Retransmit bool
32+}
33+
34+// packetPolicy makes loss and constrained links explicit in transport tests.
35+// A zero value delivers immediately, preserving newPair's historical behavior.
36+type packetPolicy struct {
37+ Drop func(packetMeta) bool
38+ Delay time.Duration
39+ MaxQueueBytes int
40+ BytesPerSecond int
41+}
42+
43+type scheduledPacket struct {
44+ packet
45+ due time.Time
46+}
47+
48+// simulatedLink has a bounded byte queue. Tests advance its clock explicitly;
49+// it never starts a goroutine or relies on wall-clock time.
50+type simulatedLink struct {
51+ mu sync.Mutex
52+ clock interface{ Now() time.Time }
53+ policy packetPolicy
54+ queue []scheduledPacket
55+ queueBytes int
56+ next time.Time
57+}
58+
59+func newSimulatedLink(clock interface{ Now() time.Time }, policy packetPolicy) *simulatedLink {
60+ return &simulatedLink{clock: clock, policy: policy}
61+}
62+
63+func (l *simulatedLink) enqueue(p *fakePC, peer *fakePC, b []byte) bool {
64+ l.mu.Lock()
65+ defer l.mu.Unlock()
66+ meta := packetMeta{From: p.addr, To: peer.addr, Bytes: len(b), At: l.clock.Now(), QueueBytes: l.queueBytes}
67+ if l.policy.Drop != nil && l.policy.Drop(meta) {
68+ return false
69+ }
70+ if l.policy.Delay == 0 && l.policy.BytesPerSecond == 0 {
71+ return l.deliverLocked(peer, packet{b: append([]byte(nil), b...), addr: p.addr})
72+ }
73+ if l.policy.MaxQueueBytes <= 0 || l.queueBytes+len(b) > l.policy.MaxQueueBytes {
74+ return false
75+ }
76+ due := meta.At.Add(l.policy.Delay)
77+ if l.policy.BytesPerSecond > 0 {
78+ if due.Before(l.next) {
79+ due = l.next
80+ }
81+ ns := int64(time.Second) * int64(len(b)) / int64(l.policy.BytesPerSecond)
82+ if ns == 0 {
83+ ns = 1
84+ }
85+ l.next = due.Add(time.Duration(ns))
86+ }
87+ l.queue = append(l.queue, scheduledPacket{packet: packet{b: append([]byte(nil), b...), addr: p.addr}, due: due})
88+ l.queueBytes += len(b)
89+ return true
90+}
91+
92+func (l *simulatedLink) flush(peer *fakePC) {
93+ l.mu.Lock()
94+ defer l.mu.Unlock()
95+ now := l.clock.Now()
96+ for len(l.queue) > 0 && !l.queue[0].due.After(now) {
97+ q := l.queue[0]
98+ l.queue = l.queue[1:]
99+ l.queueBytes -= len(q.b)
100+ l.deliverLocked(peer, q.packet)
101+ }
102+}
103+
104+func (l *simulatedLink) queuedBytes() int {
105+ l.mu.Lock()
106+ defer l.mu.Unlock()
107+ return l.queueBytes
108+}
109+
110+func (l *simulatedLink) deliverLocked(peer *fakePC, q packet) bool {
111+ peer.mu.Lock()
112+ defer peer.mu.Unlock()
113+ if peer.closed {
114+ return false
115+ }
116+ select {
117+ case peer.in <- q:
118+ return true
119+ default:
120+ // The endpoint queue is intentionally bounded just like the link queue.
121+ return false
122+ }
123+}
124+
125+type fakePC struct {
126+ addr net.Addr
127+ mu sync.Mutex
128+ closed bool
129+ in chan packet
130+ peers map[string]*fakePC
131+ drop func([]byte, net.Addr) bool
132+ afterWrite func()
133+ link *simulatedLink
134+}
135+
136+func newPair() (*fakePC, *fakePC) { return newPairWithLink(nil) }
137+
138+func newPairWithLink(link *simulatedLink) (*fakePC, *fakePC) {
139+ a := &fakePC{addr: testAddr("a"), in: make(chan packet, 100), peers: map[string]*fakePC{}, link: link}
140+ b := &fakePC{addr: testAddr("b"), in: make(chan packet, 100), peers: map[string]*fakePC{}, link: link}
141+ a.peers["b"] = b
142+ b.peers["a"] = a
143+ return a, b
144+}
145+func (p *fakePC) ReadFrom(b []byte) (int, net.Addr, error) {
146+ q, ok := <-p.in
147+ if !ok {
148+ return 0, nil, errors.New("closed")
149+ }
150+ copy(b, q.b)
151+ return len(q.b), q.addr, nil
152+}
153+func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
154+ p.mu.Lock()
155+ closed := p.closed
156+ p.mu.Unlock()
157+ if closed {
158+ return 0, errors.New("closed")
159+ }
160+ if p.drop != nil && p.drop(b, addr) {
161+ return len(b), nil
162+ }
163+ peer := p.peers[addr.String()]
164+ if peer == nil {
165+ return 0, errors.New("unknown peer")
166+ }
167+ if p.link != nil {
168+ p.link.enqueue(p, peer, b)
169+ } else {
170+ peer.mu.Lock()
171+ if peer.closed {
172+ peer.mu.Unlock()
173+ return 0, errors.New("peer closed")
174+ }
175+ peer.in <- packet{append([]byte(nil), b...), p.addr}
176+ peer.mu.Unlock()
177+ }
178+ if p.afterWrite != nil {
179+ p.afterWrite()
180+ }
181+ return len(b), nil
182+}
183+func (p *fakePC) Close() error {
184+ p.mu.Lock()
185+ defer p.mu.Unlock()
186+ if !p.closed {
187+ p.closed = true
188+ close(p.in)
189+ }
190+ return nil
191+}
192+func (p *fakePC) LocalAddr() net.Addr { return p.addr }
193+func (p *fakePC) SetDeadline(time.Time) error { return nil }
194+func (p *fakePC) SetReadDeadline(time.Time) error { return nil }
195+func (p *fakePC) SetWriteDeadline(time.Time) error { return nil }
196+
197+func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
198+
199+type floodMetrics struct {
200+ packetContact int
201+ completedRecord int
202+ ackProgress int
203+ queueBytes int
204+ retransmits int
205+}
206+
207+// TestTransportFloodClassification documents the three distinct symptoms that
208+// used to be conflated as a healthy UDP link: authenticated datagrams arriving,
209+// complete records arriving, and ACKs retiring sender work.
210+func TestTransportFloodClassification(t *testing.T) {
211+ const records = 32
212+ payload := make([]byte, 24*pdgram.DefaultMTU)
213+ fragments, err := pdgram.FragmentPayload(1, payload, pdgram.DefaultMTU)
214+ if err != nil {
215+ t.Fatal(err)
216+ }
217+ if len(fragments) < 24 {
218+ t.Fatalf("fragments=%d, want at least 24", len(fragments))
219+ }
220+ packets := make([][]byte, len(fragments))
221+ for i, fragment := range fragments {
222+ packets[i], err = pdgram.MarshalFragment(fragment)
223+ if err != nil {
224+ t.Fatal(err)
225+ }
226+ }
227+
228+ tests := []struct {
229+ name string
230+ run func(*manualClock) floodMetrics
231+ check func(*testing.T, floodMetrics)
232+ }{
233+ {
234+ name: "one fragment per record loss has packet contact without record or ACK progress",
235+ run: func(clk *manualClock) floodMetrics {
236+ attempt := 0
237+ link := newSimulatedLink(clk, packetPolicy{Drop: func(packetMeta) bool {
238+ drop := attempt%len(fragments) == len(fragments)-1
239+ attempt++
240+ return drop
241+ }})
242+ a, b := newPairWithLink(link)
243+ m := floodMetrics{}
244+ for range records {
245+ delivered := 0
246+ for _, fragment := range packets {
247+ _, _ = a.WriteTo(fragment, b.addr)
248+ select {
249+ case <-b.in:
250+ delivered++
251+ m.packetContact++
252+ default:
253+ }
254+ }
255+ if delivered == len(fragments) {
256+ m.completedRecord++
257+ m.ackProgress++
258+ }
259+ }
260+ return m
261+ },
262+ check: func(t *testing.T, m floodMetrics) {
263+ t.Helper()
264+ if m.packetContact == 0 || m.completedRecord != 0 || m.ackProgress != 0 {
265+ t.Fatalf("metrics=%+v, want packet contact only", m)
266+ }
267+ },
268+ },
269+ {
270+ name: "bounded bandwidth queue has no false progress and stays bounded",
271+ run: func(clk *manualClock) floodMetrics {
272+ limit := 2 * pdgram.DefaultMTU
273+ link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: pdgram.DefaultMTU})
274+ a, b := newPairWithLink(link)
275+ for range records {
276+ for _, fragment := range packets {
277+ _, _ = a.WriteTo(fragment, b.addr)
278+ }
279+ }
280+ return floodMetrics{queueBytes: link.queuedBytes()}
281+ },
282+ check: func(t *testing.T, m floodMetrics) {
283+ t.Helper()
284+ if m.packetContact != 0 || m.completedRecord != 0 || m.ackProgress != 0 || m.queueBytes <= 0 || m.queueBytes > 2*pdgram.DefaultMTU {
285+ t.Fatalf("metrics=%+v, want bounded queued bytes without progress", m)
286+ }
287+ },
288+ },
289+ {
290+ name: "retransmit storm has record progress but no ACK progress",
291+ run: func(clk *manualClock) floodMetrics {
292+ link := newSimulatedLink(clk, packetPolicy{})
293+ a, b := newPairWithLink(link)
294+ m := floodMetrics{}
295+ for range records {
296+ for attempt := 0; attempt < 3; attempt++ {
297+ for _, fragment := range packets {
298+ _, _ = a.WriteTo(fragment, b.addr)
299+ <-b.in
300+ m.packetContact++
301+ }
302+ if attempt == 0 {
303+ m.completedRecord++
304+ } else {
305+ m.retransmits++
306+ }
307+ }
308+ }
309+ // ACK packets are intentionally lost in this scenario.
310+ return m
311+ },
312+ check: func(t *testing.T, m floodMetrics) {
313+ t.Helper()
314+ if m.packetContact == 0 || m.completedRecord != records || m.ackProgress != 0 || m.retransmits == 0 {
315+ t.Fatalf("metrics=%+v, want record contact with retransmit-only pressure", m)
316+ }
317+ },
318+ },
319+ }
320+ for _, tt := range tests {
321+ t.Run(tt.name, func(t *testing.T) {
322+ clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
323+ metrics := tt.run(clk)
324+ tt.check(t, metrics)
325+ })
326+ }
327+}
@@ -0,0 +1,327 @@
1+package dgram
2+
3+import (
4+ "bytes"
5+ "errors"
6+ "net"
7+ "sync"
8+ "testing"
9+ "time"
10+
11+ pdgram "github.com/bnema/vev/pkg/dgram"
12+)
13+
14+type testAddr string
15+
16+func (a testAddr) Network() string { return "mem" }
17+func (a testAddr) String() string { return string(a) }
18+
19+type packet struct {
20+ b []byte
21+ addr net.Addr
22+}
23+
24+// packetMeta is the deterministic simulator's view of a datagram before it is
25+// accepted by the link. It deliberately contains only link-level information.
26+type packetMeta struct {
27+ From, To net.Addr
28+ Bytes int
29+ At time.Time
30+ QueueBytes int
31+ Retransmit bool
32+}
33+
34+// packetPolicy makes loss and constrained links explicit in transport tests.
35+// A zero value delivers immediately, preserving newPair's historical behavior.
36+type packetPolicy struct {
37+ Drop func(packetMeta) bool
38+ Delay time.Duration
39+ MaxQueueBytes int
40+ BytesPerSecond int
41+}
42+
43+type scheduledPacket struct {
44+ packet
45+ due time.Time
46+}
47+
48+// simulatedLink has a bounded byte queue. Tests advance its clock explicitly;
49+// it never starts a goroutine or relies on wall-clock time.
50+type simulatedLink struct {
51+ mu sync.Mutex
52+ clock interface{ Now() time.Time }
53+ policy packetPolicy
54+ queue []scheduledPacket
55+ queueBytes int
56+ next time.Time
57+}
58+
59+func newSimulatedLink(clock interface{ Now() time.Time }, policy packetPolicy) *simulatedLink {
60+ return &simulatedLink{clock: clock, policy: policy}
61+}
62+
63+func (l *simulatedLink) enqueue(p *fakePC, peer *fakePC, b []byte) bool {
64+ l.mu.Lock()
65+ defer l.mu.Unlock()
66+ meta := packetMeta{From: p.addr, To: peer.addr, Bytes: len(b), At: l.clock.Now(), QueueBytes: l.queueBytes}
67+ if l.policy.Drop != nil && l.policy.Drop(meta) {
68+ return false
69+ }
70+ if l.policy.Delay == 0 && l.policy.BytesPerSecond == 0 {
71+ return l.deliverLocked(peer, packet{b: append([]byte(nil), b...), addr: p.addr})
72+ }
73+ if l.policy.MaxQueueBytes <= 0 || l.queueBytes+len(b) > l.policy.MaxQueueBytes {
74+ return false
75+ }
76+ due := meta.At.Add(l.policy.Delay)
77+ if l.policy.BytesPerSecond > 0 {
78+ if due.Before(l.next) {
79+ due = l.next
80+ }
81+ ns := int64(time.Second) * int64(len(b)) / int64(l.policy.BytesPerSecond)
82+ if ns == 0 {
83+ ns = 1
84+ }
85+ l.next = due.Add(time.Duration(ns))
86+ }
87+ l.queue = append(l.queue, scheduledPacket{packet: packet{b: append([]byte(nil), b...), addr: p.addr}, due: due})
88+ l.queueBytes += len(b)
89+ return true
90+}
91+
92+func (l *simulatedLink) flush(peer *fakePC) {
93+ l.mu.Lock()
94+ defer l.mu.Unlock()
95+ now := l.clock.Now()
96+ for len(l.queue) > 0 && !l.queue[0].due.After(now) {
97+ q := l.queue[0]
98+ l.queue = l.queue[1:]
99+ l.queueBytes -= len(q.b)
100+ l.deliverLocked(peer, q.packet)
101+ }
102+}
103+
104+func (l *simulatedLink) queuedBytes() int {
105+ l.mu.Lock()
106+ defer l.mu.Unlock()
107+ return l.queueBytes
108+}
109+
110+func (l *simulatedLink) deliverLocked(peer *fakePC, q packet) bool {
111+ peer.mu.Lock()
112+ defer peer.mu.Unlock()
113+ if peer.closed {
114+ return false
115+ }
116+ select {
117+ case peer.in <- q:
118+ return true
119+ default:
120+ // The endpoint queue is intentionally bounded just like the link queue.
121+ return false
122+ }
123+}
124+
125+type fakePC struct {
126+ addr net.Addr
127+ mu sync.Mutex
128+ closed bool
129+ in chan packet
130+ peers map[string]*fakePC
131+ drop func([]byte, net.Addr) bool
132+ afterWrite func()
133+ link *simulatedLink
134+}
135+
136+func newPair() (*fakePC, *fakePC) { return newPairWithLink(nil) }
137+
138+func newPairWithLink(link *simulatedLink) (*fakePC, *fakePC) {
139+ a := &fakePC{addr: testAddr("a"), in: make(chan packet, 100), peers: map[string]*fakePC{}, link: link}
140+ b := &fakePC{addr: testAddr("b"), in: make(chan packet, 100), peers: map[string]*fakePC{}, link: link}
141+ a.peers["b"] = b
142+ b.peers["a"] = a
143+ return a, b
144+}
145+func (p *fakePC) ReadFrom(b []byte) (int, net.Addr, error) {
146+ q, ok := <-p.in
147+ if !ok {
148+ return 0, nil, errors.New("closed")
149+ }
150+ copy(b, q.b)
151+ return len(q.b), q.addr, nil
152+}
153+func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
154+ p.mu.Lock()
155+ closed := p.closed
156+ p.mu.Unlock()
157+ if closed {
158+ return 0, errors.New("closed")
159+ }
160+ if p.drop != nil && p.drop(b, addr) {
161+ return len(b), nil
162+ }
163+ peer := p.peers[addr.String()]
164+ if peer == nil {
165+ return 0, errors.New("unknown peer")
166+ }
167+ if p.link != nil {
168+ p.link.enqueue(p, peer, b)
169+ } else {
170+ peer.mu.Lock()
171+ if peer.closed {
172+ peer.mu.Unlock()
173+ return 0, errors.New("peer closed")
174+ }
175+ peer.in <- packet{append([]byte(nil), b...), p.addr}
176+ peer.mu.Unlock()
177+ }
178+ if p.afterWrite != nil {
179+ p.afterWrite()
180+ }
181+ return len(b), nil
182+}
183+func (p *fakePC) Close() error {
184+ p.mu.Lock()
185+ defer p.mu.Unlock()
186+ if !p.closed {
187+ p.closed = true
188+ close(p.in)
189+ }
190+ return nil
191+}
192+func (p *fakePC) LocalAddr() net.Addr { return p.addr }
193+func (p *fakePC) SetDeadline(time.Time) error { return nil }
194+func (p *fakePC) SetReadDeadline(time.Time) error { return nil }
195+func (p *fakePC) SetWriteDeadline(time.Time) error { return nil }
196+
197+func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
198+
199+type floodMetrics struct {
200+ packetContact int
201+ completedRecord int
202+ ackProgress int
203+ queueBytes int
204+ retransmits int
205+}
206+
207+// TestTransportFloodClassification documents the three distinct symptoms that
208+// used to be conflated as a healthy UDP link: authenticated datagrams arriving,
209+// complete records arriving, and ACKs retiring sender work.
210+func TestTransportFloodClassification(t *testing.T) {
211+ const records = 32
212+ payload := make([]byte, 24*pdgram.DefaultMTU)
213+ fragments, err := pdgram.FragmentPayload(1, payload, pdgram.DefaultMTU)
214+ if err != nil {
215+ t.Fatal(err)
216+ }
217+ if len(fragments) < 24 {
218+ t.Fatalf("fragments=%d, want at least 24", len(fragments))
219+ }
220+ packets := make([][]byte, len(fragments))
221+ for i, fragment := range fragments {
222+ packets[i], err = pdgram.MarshalFragment(fragment)
223+ if err != nil {
224+ t.Fatal(err)
225+ }
226+ }
227+
228+ tests := []struct {
229+ name string
230+ run func(*manualClock) floodMetrics
231+ check func(*testing.T, floodMetrics)
232+ }{
233+ {
234+ name: "one fragment per record loss has packet contact without record or ACK progress",
235+ run: func(clk *manualClock) floodMetrics {
236+ attempt := 0
237+ link := newSimulatedLink(clk, packetPolicy{Drop: func(packetMeta) bool {
238+ drop := attempt%len(fragments) == len(fragments)-1
239+ attempt++
240+ return drop
241+ }})
242+ a, b := newPairWithLink(link)
243+ m := floodMetrics{}
244+ for range records {
245+ delivered := 0
246+ for _, fragment := range packets {
247+ _, _ = a.WriteTo(fragment, b.addr)
248+ select {
249+ case <-b.in:
250+ delivered++
251+ m.packetContact++
252+ default:
253+ }
254+ }
255+ if delivered == len(fragments) {
256+ m.completedRecord++
257+ m.ackProgress++
258+ }
259+ }
260+ return m
261+ },
262+ check: func(t *testing.T, m floodMetrics) {
263+ t.Helper()
264+ if m.packetContact == 0 || m.completedRecord != 0 || m.ackProgress != 0 {
265+ t.Fatalf("metrics=%+v, want packet contact only", m)
266+ }
267+ },
268+ },
269+ {
270+ name: "bounded bandwidth queue has no false progress and stays bounded",
271+ run: func(clk *manualClock) floodMetrics {
272+ limit := 2 * pdgram.DefaultMTU
273+ link := newSimulatedLink(clk, packetPolicy{Delay: time.Second, MaxQueueBytes: limit, BytesPerSecond: pdgram.DefaultMTU})
274+ a, b := newPairWithLink(link)
275+ for range records {
276+ for _, fragment := range packets {
277+ _, _ = a.WriteTo(fragment, b.addr)
278+ }
279+ }
280+ return floodMetrics{queueBytes: link.queuedBytes()}
281+ },
282+ check: func(t *testing.T, m floodMetrics) {
283+ t.Helper()
284+ if m.packetContact != 0 || m.completedRecord != 0 || m.ackProgress != 0 || m.queueBytes <= 0 || m.queueBytes > 2*pdgram.DefaultMTU {
285+ t.Fatalf("metrics=%+v, want bounded queued bytes without progress", m)
286+ }
287+ },
288+ },
289+ {
290+ name: "retransmit storm has record progress but no ACK progress",
291+ run: func(clk *manualClock) floodMetrics {
292+ link := newSimulatedLink(clk, packetPolicy{})
293+ a, b := newPairWithLink(link)
294+ m := floodMetrics{}
295+ for range records {
296+ for attempt := 0; attempt < 3; attempt++ {
297+ for _, fragment := range packets {
298+ _, _ = a.WriteTo(fragment, b.addr)
299+ <-b.in
300+ m.packetContact++
301+ }
302+ if attempt == 0 {
303+ m.completedRecord++
304+ } else {
305+ m.retransmits++
306+ }
307+ }
308+ }
309+ // ACK packets are intentionally lost in this scenario.
310+ return m
311+ },
312+ check: func(t *testing.T, m floodMetrics) {
313+ t.Helper()
314+ if m.packetContact == 0 || m.completedRecord != records || m.ackProgress != 0 || m.retransmits == 0 {
315+ t.Fatalf("metrics=%+v, want record contact with retransmit-only pressure", m)
316+ }
317+ },
318+ },
319+ }
320+ for _, tt := range tests {
321+ t.Run(tt.name, func(t *testing.T) {
322+ clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
323+ metrics := tt.run(clk)
324+ tt.check(t, metrics)
325+ })
326+ }
327+}
M internal/adapters/dgram/transport_test.go +0 -73
@@ -16,79 +16,6 @@ import (
1616 "github.com/stretchr/testify/require"
1717 )
1818
19-type testAddr string
20-
21-func (a testAddr) Network() string { return "mem" }
22-func (a testAddr) String() string { return string(a) }
23-
24-type packet struct {
25- b []byte
26- addr net.Addr
27-}
28-type fakePC struct {
29- addr net.Addr
30- mu sync.Mutex
31- closed bool
32- in chan packet
33- peers map[string]*fakePC
34- drop func([]byte, net.Addr) bool
35- afterWrite func()
36-}
37-
38-func newPair() (*fakePC, *fakePC) {
39- a := &fakePC{addr: testAddr("a"), in: make(chan packet, 100), peers: map[string]*fakePC{}}
40- b := &fakePC{addr: testAddr("b"), in: make(chan packet, 100), peers: map[string]*fakePC{}}
41- a.peers["b"] = b
42- b.peers["a"] = a
43- return a, b
44-}
45-func (p *fakePC) ReadFrom(b []byte) (int, net.Addr, error) {
46- q, ok := <-p.in
47- if !ok {
48- return 0, nil, errors.New("closed")
49- }
50- copy(b, q.b)
51- return len(q.b), q.addr, nil
52-}
53-func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
54- p.mu.Lock()
55- closed := p.closed
56- p.mu.Unlock()
57- if closed {
58- return 0, errors.New("closed")
59- }
60- if p.drop != nil && p.drop(b, addr) {
61- return len(b), nil
62- }
63- peer := p.peers[addr.String()]
64- peer.mu.Lock()
65- if peer.closed {
66- peer.mu.Unlock()
67- return 0, errors.New("peer closed")
68- }
69- peer.in <- packet{append([]byte(nil), b...), p.addr}
70- peer.mu.Unlock()
71- if p.afterWrite != nil {
72- p.afterWrite()
73- }
74- return len(b), nil
75-}
76-func (p *fakePC) Close() error {
77- p.mu.Lock()
78- defer p.mu.Unlock()
79- if !p.closed {
80- p.closed = true
81- close(p.in)
82- }
83- return nil
84-}
85-func (p *fakePC) LocalAddr() net.Addr { return p.addr }
86-func (p *fakePC) SetDeadline(time.Time) error { return nil }
87-func (p *fakePC) SetReadDeadline(time.Time) error { return nil }
88-func (p *fakePC) SetWriteDeadline(time.Time) error { return nil }
89-
90-func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
91-
9219 type timeoutErr struct{}
9320
9421 func (timeoutErr) Error() string { return "i/o timeout" }
@@ -16,79 +16,6 @@ import (
16 "github.com/stretchr/testify/require"16 "github.com/stretchr/testify/require"
17 )17 )
18 18
19-type testAddr string
20-
21-func (a testAddr) Network() string { return "mem" }
22-func (a testAddr) String() string { return string(a) }
23-
24-type packet struct {
25- b []byte
26- addr net.Addr
27-}
28-type fakePC struct {
29- addr net.Addr
30- mu sync.Mutex
31- closed bool
32- in chan packet
33- peers map[string]*fakePC
34- drop func([]byte, net.Addr) bool
35- afterWrite func()
36-}
37-
38-func newPair() (*fakePC, *fakePC) {
39- a := &fakePC{addr: testAddr("a"), in: make(chan packet, 100), peers: map[string]*fakePC{}}
40- b := &fakePC{addr: testAddr("b"), in: make(chan packet, 100), peers: map[string]*fakePC{}}
41- a.peers["b"] = b
42- b.peers["a"] = a
43- return a, b
44-}
45-func (p *fakePC) ReadFrom(b []byte) (int, net.Addr, error) {
46- q, ok := <-p.in
47- if !ok {
48- return 0, nil, errors.New("closed")
49- }
50- copy(b, q.b)
51- return len(q.b), q.addr, nil
52-}
53-func (p *fakePC) WriteTo(b []byte, addr net.Addr) (int, error) {
54- p.mu.Lock()
55- closed := p.closed
56- p.mu.Unlock()
57- if closed {
58- return 0, errors.New("closed")
59- }
60- if p.drop != nil && p.drop(b, addr) {
61- return len(b), nil
62- }
63- peer := p.peers[addr.String()]
64- peer.mu.Lock()
65- if peer.closed {
66- peer.mu.Unlock()
67- return 0, errors.New("peer closed")
68- }
69- peer.in <- packet{append([]byte(nil), b...), p.addr}
70- peer.mu.Unlock()
71- if p.afterWrite != nil {
72- p.afterWrite()
73- }
74- return len(b), nil
75-}
76-func (p *fakePC) Close() error {
77- p.mu.Lock()
78- defer p.mu.Unlock()
79- if !p.closed {
80- p.closed = true
81- close(p.in)
82- }
83- return nil
84-}
85-func (p *fakePC) LocalAddr() net.Addr { return p.addr }
86-func (p *fakePC) SetDeadline(time.Time) error { return nil }
87-func (p *fakePC) SetReadDeadline(time.Time) error { return nil }
88-func (p *fakePC) SetWriteDeadline(time.Time) error { return nil }
89-
90-func key() []byte { return bytes.Repeat([]byte{1}, pdgram.KeySize) }
91-
92 type timeoutErr struct{}19 type timeoutErr struct{}
93 20
94 func (timeoutErr) Error() string { return "i/o timeout" }21 func (timeoutErr) Error() string { return "i/o timeout" }

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close