diff --git a/backend/cmd/config.toml b/backend/cmd/config.toml index a71056e8f..747bc6bbb 100644 --- a/backend/cmd/config.toml +++ b/backend/cmd/config.toml @@ -36,6 +36,8 @@ keep_alive_ms = 1000 # Keep-alive interval in milliseconds [udp] ring_buffer_size = 64 # Size of the ring buffer for incoming packets (number of packets, not bytes) packet_chan_size = 16 # Size of the channel buffer +keep_alive_check_interval_ms = 5 # How often the keep-alive checks last-received times (ms) +keep_alive_timeout_ms = 100 # Max time without packets from an IP before it is considered disconnected (ms) # Logger Configuration diff --git a/backend/cmd/dev-config.toml b/backend/cmd/dev-config.toml index 1a6d3a437..47fcacad5 100644 --- a/backend/cmd/dev-config.toml +++ b/backend/cmd/dev-config.toml @@ -38,6 +38,8 @@ keep_alive_ms = 0 # Keep-alive interval in milliseconds (0 to disab [udp] ring_buffer_size = 64 # Size of the ring buffer for incoming packets (number of packets, not bytes) packet_chan_size = 16 # Size of the channel buffer +keep_alive_check_interval_ms = 5 # How often the keep-alive checks last-received times (ms) +keep_alive_timeout_ms = 100 # Max time without packets from an IP before it is considered disconnected (ms) # Server Configuration [server.ethernet-view] diff --git a/backend/cmd/setup_transport.go b/backend/cmd/setup_transport.go index f148a27b7..d4d90e996 100644 --- a/backend/cmd/setup_transport.go +++ b/backend/cmd/setup_transport.go @@ -140,7 +140,16 @@ func configureUDPServerTransport( ) { trace.Info().Msg("Starting UDP server") - udpServer := udp.NewServer(adj.Info.Addresses[BACKEND], adj.Info.Ports[UDP], &trace.Logger, config.UDP.RingBufferSize, config.UDP.PacketChanSize) + udpServer := udp.NewServer( + adj.Info.Addresses[BACKEND], + adj.Info.Ports[UDP], + &trace.Logger, + config.UDP.RingBufferSize, + config.UDP.PacketChanSize, + time.Duration(config.UDP.KeepAliveCheckIntervalMs)*time.Millisecond, + time.Duration(config.UDP.KeepAliveTimeoutMs)*time.Millisecond, + transp.SendFault, + ) err := udpServer.Start() if err != nil { trace.Fatal().Err(err).Msg("failed to start UDP server: " + err.Error()) diff --git a/backend/internal/config/config_types.go b/backend/internal/config/config_types.go index 32260e5aa..b0957b0ef 100644 --- a/backend/internal/config/config_types.go +++ b/backend/internal/config/config_types.go @@ -25,8 +25,10 @@ type TCP struct { } type UDP struct { - RingBufferSize int `toml:"ring_buffer_size"` - PacketChanSize int `toml:"packet_chan_size"` + RingBufferSize int `toml:"ring_buffer_size"` + PacketChanSize int `toml:"packet_chan_size"` + KeepAliveCheckIntervalMs int `toml:"keep_alive_check_interval_ms"` + KeepAliveTimeoutMs int `toml:"keep_alive_timeout_ms"` } type Logging struct { diff --git a/backend/pkg/transport/network/udp/server.go b/backend/pkg/transport/network/udp/server.go index f595c7813..155708856 100644 --- a/backend/pkg/transport/network/udp/server.go +++ b/backend/pkg/transport/network/udp/server.go @@ -34,16 +34,38 @@ type Server struct { count int ringMutex sync.Mutex notEmpty *sync.Cond + + lastSeen map[string]time.Time + lastSeenMu sync.Mutex + keepAliveCheckInterval time.Duration + keepAliveTimeout time.Duration + OnDisconnect func() } -func NewServer(address string, port uint16, logger *zerolog.Logger, ringBufferSize int, packetChanSize int) *Server { +const ( + defaultKeepAliveCheckInterval = 5 * time.Millisecond + defaultKeepAliveTimeout = 100 * time.Millisecond +) + +func NewServer(address string, port uint16, logger *zerolog.Logger, ringBufferSize int, packetChanSize int, keepAliveCheckInterval time.Duration, keepAliveTimeout time.Duration, onDisconnect func()) *Server { + if keepAliveCheckInterval <= 0 { + keepAliveCheckInterval = defaultKeepAliveCheckInterval + } + if keepAliveTimeout <= 0 { + keepAliveTimeout = defaultKeepAliveTimeout + } + s := &Server{ - address: address, - port: port, - logger: logger, - packetsCh: make(chan Packet, packetChanSize), - errorsCh: make(chan error, 100), - stopCh: make(chan struct{}), + address: address, + port: port, + logger: logger, + packetsCh: make(chan Packet, packetChanSize), + errorsCh: make(chan error, 100), + stopCh: make(chan struct{}), + lastSeen: make(map[string]time.Time), + keepAliveCheckInterval: keepAliveCheckInterval, + keepAliveTimeout: keepAliveTimeout, + OnDisconnect: onDisconnect, } s.ring = make([]Packet, ringBufferSize) @@ -71,8 +93,9 @@ func (s *Server) Start() error { Uint16("port", s.port). Msg("UDP server started") - go s.readLoop() - go s.dispatchLoop() + go s.readLoop() // Read incoming UDP packets + go s.dispatchLoop() // Dispatch packets from ring buffer to channel + go s.keepAliveLoop() // Monitor keep-alive timeouts return nil } @@ -121,12 +144,51 @@ func (s *Server) readLoop() { Int("size", len(payload)). Msg("received UDP packet") + // Update keep-alive tracking for this source IP + s.lastSeenMu.Lock() + s.lastSeen[addr.IP.String()] = packet.Timestamp + s.lastSeenMu.Unlock() + // Push packet to ring buffer s.push(packet) } } } +// keepAliveLoop checks for clients that have not sent any packets within the keepAliveTimeout duration. +func (s *Server) keepAliveLoop() { + ticker := time.NewTicker(s.keepAliveCheckInterval) + defer ticker.Stop() + + for { + select { + case <-s.stopCh: + return + case now := <-ticker.C: + var timedOut []string + + s.lastSeenMu.Lock() + for ip, last := range s.lastSeen { + if now.Sub(last) > s.keepAliveTimeout { + timedOut = append(timedOut, ip) + delete(s.lastSeen, ip) + } + } + s.lastSeenMu.Unlock() + + for _, ip := range timedOut { + s.logger.Error(). + Str("ip", ip). + Dur("timeout", s.keepAliveTimeout). + Msg("keep-alive timeout: no UDP packets received") + if s.OnDisconnect != nil { + s.OnDisconnect() + } + } + } + } +} + func (s *Server) GetPackets() <-chan Packet { return s.packetsCh }