1// Package dgram adapts authenticated UDP-style packet connections to ports.Transport.
2package dgram
3
4import (
5 "context"
6 "errors"
7 "net"
8 "sync"
9 "time"
10
11 "github.com/bnema/vev/internal/ports"
12 pdgram "github.com/bnema/vev/pkg/dgram"
13)
14
15const (
16 recData byte = 1
17 recAck byte = 2
18 recProbe byte = 3
19 recPong byte = 4
20
21 defaultMTU = pdgram.DefaultMTU
22 defaultResend = 250 * time.Millisecond
23 defaultMaxResendAfter = 2 * time.Second
24 defaultMaxResendPerTick = 64
25 defaultWriteTimeout = 250 * time.Millisecond
26 defaultHeartbeat = 3 * time.Second
27 defaultDegraded = 10 * time.Second
28 defaultProbe = 20 * time.Second
29 defaultOffline = 30 * time.Second
30 defaultDead = 60 * time.Second
31 defaultMaxPending = 1024
32 defaultMaxPendingWait = 50 * time.Millisecond
33 defaultMaxRecvBuf = 1024
34 dataRecordHeaderSize = 12
35 maxRecvBuffer = defaultMaxRecvBuf
36 linkEventBufferSize = 16
37 probeReplyBufferSize = controlQueueSize // legacy test burst bound
38)
39
40var (
41 ErrPendingFull = errors.New("dgram: pending reliable queue full")
42 ErrLinkDead = errors.New("dgram: link dead")
43 errControlFull = errors.New("dgram: control queue full")
44)
45
46type RuntimeOption func(*Transport)
47
48// WithRuntimeObserver enables process-local marks without changing dgram's
49// behavior clock or carriage policy.
50func WithRuntimeObserver(observer ports.SerializedRuntimeObserver) RuntimeOption {
51 return func(t *Transport) { t.runtimeObserver = observer }
52}
53
54type Options struct {
55 MTU int
56 ResendAfter time.Duration
57 MaxResendAfter time.Duration
58 MaxResendPerTick int
59 WriteTimeout time.Duration
60 Heartbeat time.Duration
61 DegradedAfter time.Duration
62 ProbeAfter time.Duration
63 OfflineAfter time.Duration
64 DeadAfter time.Duration
65 MaxPending int
66 MaxPendingWait time.Duration
67 MaxRecvBuffer int
68 Clock ports.Clock
69 // Observe receives bounded, asynchronous, privacy-safe link health snapshots.
70 Observe DiagnosticObserver
71 // RebindPacketConn is an optional client-side hook used to hop the local UDP
72 // socket when the peer is offline. Server/proxy transports should leave it nil.
73 RebindPacketConn func(net.PacketConn) (net.PacketConn, error)
74}
75
76type realClock struct{}
77
78func (realClock) Now() time.Time { return time.Now() }
79func (realClock) NewTimer(d time.Duration) ports.Timer {
80 return realTimer{t: time.NewTimer(d)}
81}
82
83type realTimer struct{ t *time.Timer }
84
85func (r realTimer) C() <-chan time.Time { return r.t.C }
86func (r realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) }
87func (r realTimer) Stop() bool { return r.t.Stop() }
88
89func normalizeOptions(opts Options) Options {
90 if opts.MTU <= 0 {
91 opts.MTU = defaultMTU
92 }
93 if opts.ResendAfter <= 0 {
94 opts.ResendAfter = defaultResend
95 }
96 if opts.MaxResendAfter <= 0 {
97 opts.MaxResendAfter = defaultMaxResendAfter
98 }
99 if opts.MaxResendAfter < opts.ResendAfter {
100 opts.MaxResendAfter = opts.ResendAfter
101 }
102 if opts.MaxResendPerTick <= 0 {
103 opts.MaxResendPerTick = defaultMaxResendPerTick
104 }
105 if opts.WriteTimeout <= 0 {
106 opts.WriteTimeout = defaultWriteTimeout
107 }
108 if opts.Heartbeat <= 0 {
109 opts.Heartbeat = defaultHeartbeat
110 }
111 if opts.DegradedAfter <= 0 {
112 opts.DegradedAfter = defaultDegraded
113 }
114 if opts.ProbeAfter <= 0 {
115 opts.ProbeAfter = defaultProbe
116 }
117 if opts.OfflineAfter <= 0 {
118 opts.OfflineAfter = defaultOffline
119 }
120 if opts.DeadAfter <= 0 {
121 opts.DeadAfter = defaultDead
122 }
123 if opts.MaxPending <= 0 {
124 opts.MaxPending = defaultMaxPending
125 }
126 if opts.MaxPendingWait <= 0 {
127 opts.MaxPendingWait = defaultMaxPendingWait
128 }
129 if opts.MaxRecvBuffer <= 0 {
130 opts.MaxRecvBuffer = defaultMaxRecvBuf
131 }
132 if opts.Clock == nil {
133 opts.Clock = realClock{}
134 }
135 return opts
136}
137
138var _ ports.DatagramTransport = (*Transport)(nil)
139
140type Transport struct {
141 pc net.PacketConn
142 codec *pdgram.Codec
143 sendDir, recvDir uint32
144 mtu int
145
146 mu sync.Mutex
147 peer net.Addr
148 ctr uint64
149 seq uint64
150 probeSeq uint64
151 pending map[uint64]*pending
152 closed bool
153 closeErr error
154 health healthTracker
155 peerCounter uint64
156 peerCounterSet bool
157 // These mirrors preserve the established diagnostic snapshot fields while
158 // health owns link-state decisions.
159 lastAuthenticatedPacket time.Time
160 lastCompleteRecord time.Time
161 lastACKProgress time.Time
162 retransmits uint64
163 reassemblyInflight int
164 // afterAuthenticatedPacket is a test synchronization hook. It runs after
165 // authenticated-contact state has been committed.
166 afterAuthenticatedPacket func()
167 // afterMalformedFragment is a test synchronization hook. It runs after an
168 // authenticated datagram has been rejected as an invalid fragment.
169 afterMalformedFragment func()
170 // afterHealthDecision is a test synchronization hook. It runs after a health
171 // snapshot is taken and before that decision is conditionally committed.
172 afterHealthDecision func()
173 // afterLinkStateCommit is a test synchronization hook. It runs after state
174 // mutation and before ordered event publication.
175 afterLinkStateCommit func(ports.LinkState)
176 observe DiagnosticObserver
177 runtimeObserver ports.SerializedRuntimeObserver
178 operationMu sync.Mutex
179 operationCount int
180 closing bool
181 operationsDone chan struct{}
182 diagnosticCh chan Diagnostic
183 heartbeat time.Duration
184 resendAfter time.Duration
185 maxResendAfter time.Duration
186 maxResendPerTick int
187 srtt time.Duration
188 rttvar time.Duration
189 rto time.Duration
190 writeTimeout time.Duration
191 degradedAfter time.Duration
192 probeAfter time.Duration
193 offlineAfter time.Duration
194 deadAfter time.Duration
195 maxPending int
196 maxPendingWait time.Duration
197 maxRecvBuffer int
198 clock ports.Clock
199 congestion congestionController
200 linkState ports.LinkState
201 linkStateGeneration uint64
202 linkEvents chan ports.LinkEvent
203 linkEventMu sync.Mutex
204 probeWait map[uint64]chan struct{}
205 sendWake chan struct{}
206 control chan controlRecord
207 ackWake chan struct{}
208 ackSend chan uint64
209 controlMu sync.Mutex
210 ackSeq uint64
211 ackQueued bool
212 rebind func(net.PacketConn) (net.PacketConn, error)
213 hoppedOffline bool
214 hopGeneration uint64
215 outputQueue []queuedSend
216 outputWake chan struct{}
217 dataSend chan dataSendJob
218 retransmitWork chan []retransmitRecord
219 writeDeadlines *writeDeadlineState
220
221 recvMu sync.Mutex
222 replay *pdgram.ReplayWindow
223 reasm *pdgram.Reassembler
224 in chan ports.Frame
225 done chan struct{}
226
227 // Connection replacement takes writeMu, controlConnMu, then mu. Data writes
228 // take only writeMu; control writes take controlConnMu's read lock, then mu.
229 // This keeps control independent of data writes while preventing a rebind
230 // from retiring its PacketConn during a control write.
231 writeMu sync.Mutex
232 controlConnMu sync.RWMutex
233 // beforeDataPace is a test synchronization hook. It is read under mu.
234 beforeDataPace func()
235 // afterDataPaceWait observes a data pacer's next manual-clock deadline.
236 // Test observers must be non-blocking.
237 afterDataPaceWait func(time.Time)
238 // afterDataJob observes completion of a data or retransmit send job. Test
239 // observers must be non-blocking.
240 afterDataJob func()
241 // Timer hooks are test-only completion notifications. Observers must be
242 // non-blocking so timer loops never depend on test scheduling.
243 afterRetransmitTimer func()
244 afterHealthTimer func()
245 afterHeartbeatTimer func()
246 // afterPacketProcessed is a test synchronization hook called after an
247 // authenticated packet has been fully processed. It is read under mu.
248 afterPacketProcessed func()
249 // ACK hooks are test-only and observers must be non-blocking.
250 afterACKWakeAccepted func()
251 afterACKScheduled func()
252 afterACKDispatched func()
253 // outboundMu orders first transmissions and paced batches. Retransmits use
254 // writeMu directly because they do not establish new sequence order.
255 outboundMu sync.Mutex
256
257 deliverMu sync.Mutex
258 deliverCond *sync.Cond
259 nextRecvSeq uint64
260 recvBuf map[uint64]ports.Frame
261}
262
263type pending struct {
264 frame ports.Frame
265 enqueued time.Time
266 first time.Time
267 last time.Time
268 initialInFlight bool
269 attempts int
270 retransmitted bool
271 wireBytes int
272}
273
274type retransmitRecord struct {
275 seq uint64
276 f ports.Frame
277}
278
279type writeDeadlineState struct {
280 mu sync.Mutex
281 pc net.PacketConn
282 nextID uint64
283 deadlines map[uint64]time.Time
284}
285
286func newWriteDeadlineState(pc net.PacketConn) *writeDeadlineState {
287 return &writeDeadlineState{pc: pc, deadlines: make(map[uint64]time.Time)}
288}
289
290func (s *writeDeadlineState) begin(deadline time.Time) (func(), error) {
291 s.mu.Lock()
292 s.nextID++
293 id := s.nextID
294 s.deadlines[id] = deadline
295 if err := s.applyLocked(); err != nil {
296 delete(s.deadlines, id)
297 s.mu.Unlock()
298 return nil, err
299 }
300 s.mu.Unlock()
301 return func() {
302 s.mu.Lock()
303 delete(s.deadlines, id)
304 _ = s.applyLocked()
305 s.mu.Unlock()
306 }, nil
307}
308
309func (s *writeDeadlineState) applyLocked() error {
310 var earliest time.Time
311 for _, deadline := range s.deadlines {
312 if earliest.IsZero() || deadline.Before(earliest) {
313 earliest = deadline
314 }
315 }
316 return s.pc.SetWriteDeadline(earliest)
317}
318
319func (s *writeDeadlineState) expire(now time.Time) {
320 s.mu.Lock()
321 for id, deadline := range s.deadlines {
322 if !deadline.After(now) {
323 delete(s.deadlines, id)
324 }
325 }
326 _ = s.applyLocked()
327 s.mu.Unlock()
328}
329
330// Probe sends an authenticated datagram and waits for the peer to authenticate a
331// response or for ctx to expire. It is intended for UDP bootstrap reachability checks.
332func (t *Transport) Probe(ctx context.Context) error {
333 id, ch, err := t.registerProbe()
334 if err != nil {
335 return err
336 }
337 defer t.unregisterProbe(id)
338 sendResult := make(chan error, 1)
339 if !t.queueControlRecord(controlRecord{kind: recProbe, id: id, result: sendResult}) {
340 return errControlFull
341 }
342 for {
343 select {
344 case err := <-sendResult:
345 if err != nil {
346 return err
347 }
348 sendResult = nil
349 case <-ch:
350 return nil
351 case <-ctx.Done():
352 return ctx.Err()
353 case <-t.done:
354 t.mu.Lock()
355 err := t.closeErr
356 t.mu.Unlock()
357 if err != nil {
358 return err
359 }
360 return errors.New("dgram: closed")
361 }
362 }
363}
364
365func (t *Transport) registerProbe() (uint64, chan struct{}, error) {
366 t.mu.Lock()
367 defer t.mu.Unlock()
368 if t.closed {
369 if t.closeErr != nil {
370 return 0, nil, t.closeErr
371 }
372 return 0, nil, errors.New("dgram: closed")
373 }
374 t.probeSeq++
375 id := t.probeSeq
376 ch := make(chan struct{})
377 t.probeWait[id] = ch
378 return id, ch, nil
379}
380
381func (t *Transport) unregisterProbe(id uint64) {
382 t.mu.Lock()
383 delete(t.probeWait, id)
384 t.mu.Unlock()
385}
386
387func (t *Transport) pendingExists(seq uint64) bool {
388 t.mu.Lock()
389 defer t.mu.Unlock()
390 _, ok := t.pending[seq]
391 return ok
392}
393
394// DatagramTransport marks Transport as a UDP-style datagram transport.
395func (*Transport) DatagramTransport() {}