fix(backend): add time limit udp

Author: JavierRibaldelRio

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

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]

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())

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 {

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
 }