1package traffic
2
3import (
4 "context"
5 "errors"
6 "net"
7 "net/netip"
8 "strconv"
9 "sync"
10 "sync/atomic"
11 "time"
12
13 "github.com/bnema/gordon/internal/domain"
14)
15
16const (
17 udpBufferSize = 64 << 10
18 udpWorkerCount = 16
19 udpDatagramBacklog = 1024
20)
21
22type udpEntryPointRuntime struct {
23 manager *Manager
24 entryPoint domain.EntryPoint
25 packetConn net.PacketConn
26 counters trafficCounters
27 trusted []*net.IPNet
28
29 started atomic.Bool
30 closed atomic.Bool
31
32 ctx context.Context
33 cancel context.CancelFunc
34 // runWG tracks every goroutine owned by this runtime (read loop, datagram
35 // workers, expiry loop and per-session backend loops). Counts are only added
36 // while holding mu and before the owning goroutine starts; stop() flips
37 // accepting under the same lock before it ever waits, so no Add can race a
38 // Wait.
39 runWG sync.WaitGroup
40
41 datagrams chan udpDatagram
42
43 // dialContext dials the upstream backend. It is a private instance field so
44 // tests can drive in-flight dial interleavings deterministically; production
45 // code always uses the standard UDP dialer set at construction and never
46 // mutates it after the runtime starts.
47 dialContext func(ctx context.Context, network, address string) (net.Conn, error)
48
49 mu sync.Mutex
50 accepting bool
51 sessions map[netip.AddrPort]*udpSession
52}
53
54type udpDatagram struct {
55 clientAddr netip.AddrPort
56 packet []byte
57}
58
59type udpSession struct {
60 clientAddr netip.AddrPort
61 // remoteAddr is the immutable packet destination precomputed once at
62 // publication, so backendLoop never has to allocate or format an address
63 // per packet. It is shared read-only across the session's goroutines.
64 remoteAddr *net.UDPAddr
65 backend net.Conn
66 backendRef domain.TrafficBackend
67 lastSeen atomic.Int64
68 once sync.Once
69}
70
71func newUDPEntryPointRuntime(parentCtx context.Context, manager *Manager, entryPoint domain.EntryPoint, packetConn net.PacketConn, trusted []*net.IPNet) *udpEntryPointRuntime {
72 ctx, cancel := context.WithCancel(context.WithoutCancel(parentCtx))
73 return &udpEntryPointRuntime{
74 manager: manager,
75 entryPoint: entryPoint,
76 packetConn: packetConn,
77 trusted: trusted,
78 ctx: ctx,
79 cancel: cancel,
80 accepting: true,
81 datagrams: make(chan udpDatagram, udpDatagramBacklog),
82 sessions: map[netip.AddrPort]*udpSession{},
83 dialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
84 return (&net.Dialer{}).DialContext(ctx, network, address)
85 },
86 }
87}
88
89func (r *udpEntryPointRuntime) start() {
90 if !r.started.CompareAndSwap(false, true) {
91 return
92 }
93 r.mu.Lock()
94 if !r.accepting {
95 r.mu.Unlock()
96 return
97 }
98 r.runWG.Add(udpWorkerCount + 2) // read loop, expiry loop and datagram workers
99 r.mu.Unlock()
100 entryPoint := r.entryPointSnapshot()
101 trafficInfo(r.ctx).Str("entrypoint", entryPoint.Name).Str("address", entryPoint.Address).Str("protocol", string(entryPoint.Protocol)).Msg("started udp traffic entrypoint")
102 for range udpWorkerCount {
103 go r.datagramWorker()
104 }
105 go r.readLoop()
106 go r.expireLoop()
107}
108
109func (r *udpEntryPointRuntime) readLoop() {
110 defer r.runWG.Done()
111 buf := make([]byte, udpBufferSize)
112 for {
113 n, clientAddr, err := r.packetConn.ReadFrom(buf)
114 if err != nil {
115 if r.isClosed() || errors.Is(err, net.ErrClosed) {
116 return
117 }
118 r.counters.totalErrors.Add(1)
119 continue
120 }
121 key, ok := udpAddrPort(clientAddr)
122 if !ok {
123 r.counters.totalRefused.Add(1)
124 continue
125 }
126 if !trustedAddrPort(r.trustedSnapshot(), key) {
127 r.counters.totalRefused.Add(1)
128 continue
129 }
130 select {
131 case <-r.ctx.Done():
132 return
133 default:
134 }
135 if len(r.datagrams) >= cap(r.datagrams) {
136 r.counters.totalRefused.Add(1)
137 continue
138 }
139 packet := append([]byte(nil), buf[:n]...)
140 select {
141 case r.datagrams <- udpDatagram{clientAddr: key, packet: packet}:
142 case <-r.ctx.Done():
143 return
144 default:
145 r.counters.totalRefused.Add(1)
146 }
147 }
148}
149
150func (r *udpEntryPointRuntime) datagramWorker() {
151 defer r.runWG.Done()
152 for {
153 select {
154 case <-r.ctx.Done():
155 return
156 case datagram := <-r.datagrams:
157 select {
158 case <-r.ctx.Done():
159 return
160 default:
161 }
162 r.handleDatagram(datagram.clientAddr, datagram.packet)
163 }
164 }
165}
166
167func (r *udpEntryPointRuntime) handleDatagram(clientAddr netip.AddrPort, packet []byte) {
168 options := effectiveUDPOptions(snapshotUDPOptions(r.manager.snapshot.Load()))
169 session, ok := r.session(clientAddr, options)
170 if !ok {
171 return
172 }
173 n, err := session.backend.Write(packet)
174 r.counters.bytesIn.Add(int64(n))
175 if err != nil {
176 r.counters.totalErrors.Add(1)
177 r.removeSession(session)
178 }
179}
180
181func (r *udpEntryPointRuntime) session(clientAddr netip.AddrPort, options domain.UDPOptions) (*udpSession, bool) {
182 r.mu.Lock()
183 if existing := r.sessions[clientAddr]; existing != nil {
184 existing.touch()
185 r.mu.Unlock()
186 return existing, true
187 }
188 if !r.accepting {
189 r.mu.Unlock()
190 return nil, false
191 }
192 r.mu.Unlock()
193
194 backend, ok := r.resolveUDPBackend()
195 if !ok {
196 r.counters.totalRefused.Add(1)
197 return nil, false
198 }
199
200 r.mu.Lock()
201 if options.MaxSessions > 0 && len(r.sessions) >= options.MaxSessions {
202 r.mu.Unlock()
203 r.counters.totalRefused.Add(1)
204 return nil, false
205 }
206 r.mu.Unlock()
207 dialCtx, cancel := context.WithTimeout(r.ctx, udpDialTimeout(options))
208 backendConn, err := r.dialContext(dialCtx, "udp", net.JoinHostPort(backend.Host, strconv.Itoa(backend.Port)))
209 cancel()
210 if err != nil {
211 r.counters.totalErrors.Add(1)
212 return nil, false
213 }
214 session := &udpSession{
215 clientAddr: clientAddr,
216 remoteAddr: net.UDPAddrFromAddrPort(clientAddr),
217 backend: backendConn,
218 backendRef: backend,
219 }
220 session.touch()
221 return r.publishSession(session, options)
222}
223
224// publishSession runs the locked post-dial admission, re-dedup and publication
225// step. It must only be called after the backend dial completes. It returns the
226// session callers should use:
227// - the newly dialed session once it is published,
228// - an existing session that won a concurrent creation race, with the newly
229// dialed backend closed and the existing session touched,
230// - nil when admission has closed or the session limit was reached, with the
231// newly dialed backend closed.
232//
233// The runWG count for a published session is added while holding mu and before
234// its goroutine starts, so it can never race stop()'s Wait (see the runWG
235// field comment). The backendLoop goroutine starts after the lock is released.
236func (r *udpEntryPointRuntime) publishSession(session *udpSession, options domain.UDPOptions) (*udpSession, bool) {
237 r.mu.Lock()
238 if !r.accepting {
239 r.mu.Unlock()
240 _ = session.backend.Close()
241 return nil, false
242 }
243 if existing := r.sessions[session.clientAddr]; existing != nil {
244 r.mu.Unlock()
245 _ = session.backend.Close()
246 existing.touch()
247 return existing, true
248 }
249 if options.MaxSessions > 0 && len(r.sessions) >= options.MaxSessions {
250 r.mu.Unlock()
251 _ = session.backend.Close()
252 r.counters.totalRefused.Add(1)
253 return nil, false
254 }
255 r.sessions[session.clientAddr] = session
256 r.runWG.Add(1)
257 r.counters.activeUDPSessions.Add(1)
258 r.counters.totalAccepted.Add(1)
259 r.mu.Unlock()
260
261 go r.backendLoop(session)
262 return session, true
263}
264
265func trafficBackendEqual(left domain.TrafficBackend, right domain.TrafficBackend) bool {
266 return left.Name == right.Name && left.Host == right.Host && left.Port == right.Port && left.Protocol == right.Protocol
267}
268
269func udpDialTimeout(options domain.UDPOptions) time.Duration {
270 if options.IdleTimeout > 0 && options.IdleTimeout < 5*time.Second {
271 return options.IdleTimeout
272 }
273 return 5 * time.Second
274}
275
276func (r *udpEntryPointRuntime) backendLoop(session *udpSession) {
277 defer r.runWG.Done()
278 defer r.removeSession(session)
279 buf := make([]byte, udpBufferSize)
280 for {
281 n, err := session.backend.Read(buf)
282 if err != nil {
283 if !errors.Is(err, net.ErrClosed) && !r.isClosed() {
284 r.counters.totalErrors.Add(1)
285 }
286 return
287 }
288 written, err := r.packetConn.WriteTo(buf[:n], session.remoteAddr)
289 r.counters.bytesOut.Add(int64(written))
290 if err != nil {
291 if !r.isClosed() {
292 r.counters.totalErrors.Add(1)
293 }
294 return
295 }
296 }
297}
298
299func (r *udpEntryPointRuntime) expireLoop() {
300 defer r.runWG.Done()
301 ticker := time.NewTicker(25 * time.Millisecond)
302 defer ticker.Stop()
303 for {
304 select {
305 case <-r.ctx.Done():
306 return
307 case <-ticker.C:
308 r.expireIdleSessions(effectiveUDPOptions(snapshotUDPOptions(r.manager.snapshot.Load())).IdleTimeout)
309 }
310 }
311}
312
313func (r *udpEntryPointRuntime) expireIdleSessions(idleTimeout time.Duration) {
314 if idleTimeout <= 0 {
315 idleTimeout = defaultUDPOptions().IdleTimeout
316 }
317 cutoff := time.Now().Add(-idleTimeout).UnixNano()
318 expired := []*udpSession{}
319 r.mu.Lock()
320 for key, session := range r.sessions {
321 if session.lastSeen.Load() <= cutoff {
322 delete(r.sessions, key)
323 expired = append(expired, session)
324 }
325 }
326 r.mu.Unlock()
327 for _, session := range expired {
328 session.close()
329 r.counters.activeUDPSessions.Add(-1)
330 }
331}
332
333// removeSession drops session from the session map and closes its backend only
334// when it is still the session currently registered for its client address. A
335// stale remover (expiry, backend read/write failure, reload or CIDR cleanup) can
336// therefore never evict or close a replacement published under the same key.
337// The backend socket is closed outside the lock.
338func (r *udpEntryPointRuntime) removeSession(session *udpSession) {
339 if session == nil {
340 return
341 }
342 r.mu.Lock()
343 current := r.sessions[session.clientAddr]
344 if current == session {
345 delete(r.sessions, session.clientAddr)
346 }
347 r.mu.Unlock()
348 if current != session {
349 return
350 }
351 session.close()
352 r.counters.activeUDPSessions.Add(-1)
353}
354
355func (r *udpEntryPointRuntime) resolveUDPBackend() (domain.TrafficBackend, bool) {
356 graph := r.manager.snapshot.Load()
357 if graph == nil {
358 return domain.TrafficBackend{}, false
359 }
360 entryPoint := r.entryPointSnapshot()
361 var router domain.TrafficRouter
362 for _, candidate := range graph.Routers {
363 if candidate.EntryPoint == entryPoint.Name && candidate.Protocol == domain.RouterProtocolUDP {
364 if router.Name != "" {
365 return domain.TrafficBackend{}, false
366 }
367 router = candidate
368 }
369 }
370 if router.Name == "" {
371 return domain.TrafficBackend{}, false
372 }
373 for _, service := range graph.Services {
374 if service.Name == router.Service && len(service.Backends) == 1 && service.Backends[0].Protocol == domain.NetworkProtocolUDP {
375 return service.Backends[0], true
376 }
377 }
378 return domain.TrafficBackend{}, false
379}
380
381func (r *udpEntryPointRuntime) stop(ctx context.Context, drainTimeout time.Duration) {
382 if r.closed.CompareAndSwap(false, true) {
383 entryPoint := r.entryPointSnapshot()
384 trafficInfo(ctx).Str("entrypoint", entryPoint.Name).Str("address", entryPoint.Address).Msg("stopping udp traffic entrypoint")
385 // Closing admission under mu serializes with start() and session(),
386 // which are the only places that add to runWG. Once this critical
387 // section completes no further Add can occur, so waitRuntime may Wait
388 // safely.
389 r.mu.Lock()
390 r.accepting = false
391 r.mu.Unlock()
392 r.cancel()
393 _ = r.packetConn.Close()
394 }
395 if r.waitRuntime(ctx, drainTimeout) {
396 trafficInfo(ctx).Str("entrypoint", r.entryPointSnapshot().Name).Msg("stopped udp traffic entrypoint")
397 return
398 }
399 trafficDebug(ctx).Str("entrypoint", r.entryPointSnapshot().Name).Dur("drain_timeout", drainTimeout).Msg("forcing udp traffic entrypoint drain")
400 r.closeSessions()
401 if !r.waitRuntime(ctx, drainTimeout) {
402 r.runWG.Wait()
403 }
404 trafficInfo(ctx).Str("entrypoint", r.entryPointSnapshot().Name).Msg("stopped udp traffic entrypoint")
405}
406
407func (r *udpEntryPointRuntime) drainSessionsAfter(drainTimeout time.Duration) {
408 trafficDebug(r.ctx).Str("entrypoint", r.entryPointSnapshot().Name).Dur("drain_timeout", drainTimeout).Msg("scheduled udp session drain after router removal")
409 r.drainSessionsMatchingAfter(func(*udpSession) bool { return true }, drainTimeout)
410}
411
412func (r *udpEntryPointRuntime) drainSessionsNotMatchingAfter(backend domain.TrafficBackend, drainTimeout time.Duration) {
413 trafficDebug(r.ctx).Str("entrypoint", r.entryPointSnapshot().Name).Dur("drain_timeout", drainTimeout).Msg("scheduled stale udp session drain after backend update")
414 r.drainSessionsMatchingAfter(func(session *udpSession) bool { return !trafficBackendEqual(session.backendRef, backend) }, drainTimeout)
415}
416
417func (r *udpEntryPointRuntime) drainSessionsMatchingAfter(match func(*udpSession) bool, drainTimeout time.Duration) {
418 go func() {
419 if drainTimeout <= 0 {
420 drainTimeout = defaultUDPOptions().DrainTimeout
421 }
422 select {
423 case <-time.After(drainTimeout):
424 r.closeSessionsMatching(match)
425 case <-r.ctx.Done():
426 }
427 }()
428}
429
430func (r *udpEntryPointRuntime) waitRuntime(ctx context.Context, drainTimeout time.Duration) bool {
431 if drainTimeout <= 0 {
432 drainTimeout = defaultUDPOptions().DrainTimeout
433 }
434 drainCtx, cancel := context.WithTimeout(ctx, drainTimeout)
435 defer cancel()
436 done := make(chan struct{})
437 go func() {
438 r.runWG.Wait()
439 close(done)
440 }()
441 select {
442 case <-done:
443 return true
444 case <-drainCtx.Done():
445 return false
446 }
447}
448
449func (r *udpEntryPointRuntime) closeSessions() {
450 r.closeSessionsMatching(func(*udpSession) bool { return true })
451}
452
453func (r *udpEntryPointRuntime) closeSessionsMatching(match func(*udpSession) bool) {
454 r.mu.Lock()
455 sessions := make([]*udpSession, 0, len(r.sessions))
456 for _, session := range r.sessions {
457 if match(session) {
458 sessions = append(sessions, session)
459 }
460 }
461 r.mu.Unlock()
462 for _, session := range sessions {
463 r.removeSession(session)
464 }
465}
466
467func (r *udpEntryPointRuntime) matches(entryPoint domain.EntryPoint) bool {
468 current := r.entryPointSnapshot()
469 return current.Name == entryPoint.Name && current.Address == entryPoint.Address && current.Protocol == entryPoint.Protocol && trustedCIDRsEqual(current.TrustedCIDRs, entryPoint.TrustedCIDRs)
470}
471
472func (r *udpEntryPointRuntime) sameAddress(entryPoint domain.EntryPoint) bool {
473 return r.entryPointSnapshot().Address == entryPoint.Address
474}
475
476func (r *udpEntryPointRuntime) updateEntryPoint(entryPoint domain.EntryPoint, trusted []*net.IPNet) {
477 r.mu.Lock()
478 r.entryPoint = entryPoint
479 r.trusted = trusted
480 stale := make([]*udpSession, 0)
481 for _, session := range r.sessions {
482 if !trustedAddrPort(trusted, session.clientAddr) {
483 stale = append(stale, session)
484 }
485 }
486 r.mu.Unlock()
487 for _, session := range stale {
488 r.removeSession(session)
489 }
490}
491
492func (r *udpEntryPointRuntime) entryPointSnapshot() domain.EntryPoint {
493 r.mu.Lock()
494 defer r.mu.Unlock()
495 return r.entryPoint
496}
497
498func (r *udpEntryPointRuntime) trustedSnapshot() []*net.IPNet {
499 r.mu.Lock()
500 defer r.mu.Unlock()
501 return r.trusted
502}
503
504func (r *udpEntryPointRuntime) isClosed() bool { return r.closed.Load() }
505
506// udpAddrPort converts a packet source address into a typed, comparable session
507// key. IPv4-mapped addresses are normalized to their IPv4 form so that both
508// representations share a single session; IPv6 zones are preserved. Invalid
509// addresses are rejected before any trusted-list handling so they can never be
510// admitted as a zero-valued session key.
511func udpAddrPort(addr net.Addr) (netip.AddrPort, bool) {
512 if udpAddr, ok := addr.(*net.UDPAddr); ok {
513 key := normalizeUDPAddrPort(udpAddr.AddrPort())
514 return key, key.IsValid()
515 }
516 parsed, err := netip.ParseAddrPort(addr.String())
517 if err != nil {
518 return netip.AddrPort{}, false
519 }
520 key := normalizeUDPAddrPort(parsed)
521 return key, key.IsValid()
522}
523
524func normalizeUDPAddrPort(addrPort netip.AddrPort) netip.AddrPort {
525 if !addrPort.IsValid() {
526 return addrPort
527 }
528 return netip.AddrPortFrom(addrPort.Addr().Unmap(), addrPort.Port())
529}
530
531// trustedAddrPort reports whether addr is within the trusted CIDRs. An empty
532// trusted list admits every client.
533//
534// Matching is intentionally by IP only: the IPv6 zone identifies the local
535// interface a packet arrived on, not the peer, so a zoned link-local client
536// matches the same CIDR as its zoneless form. Zones remain part of the session
537// key (see udpAddrPort) and are dropped only for trust evaluation. The address
538// is unmapped before matching so a v4-mapped client matches an IPv4 CIDR.
539func trustedAddrPort(trusted []*net.IPNet, addr netip.AddrPort) bool {
540 if !addr.IsValid() {
541 return false
542 }
543 if len(trusted) == 0 {
544 return true
545 }
546 ip := addr.Addr().Unmap().AsSlice()
547 for _, network := range trusted {
548 if network.Contains(ip) {
549 return true
550 }
551 }
552 return false
553}
554
555func (s *udpSession) touch() { s.lastSeen.Store(time.Now().UnixNano()) }
556
557func (s *udpSession) close() {
558 s.once.Do(func() {
559 _ = s.backend.Close()
560 })
561}