From e1601f03c5f8724aca55c192a1dc46966391475d Mon Sep 17 00:00:00 2001 From: Javier Ribal del Rio Date: Wed, 15 Jul 2026 13:25:45 +0200 Subject: [PATCH 1/2] feat(backend): remove TCP fault for UDP --- backend/cmd/config.toml | 5 +- backend/cmd/dev-config.toml | 5 +- backend/cmd/setup_transport.go | 42 +++--- backend/internal/config/config_types.go | 9 +- backend/pkg/transport/network/tcp/conn.go | 4 - backend/pkg/transport/network/udp/server.go | 151 +++++++++++--------- backend/pkg/transport/transport.go | 63 ++++---- 7 files changed, 146 insertions(+), 133 deletions(-) diff --git a/backend/cmd/config.toml b/backend/cmd/config.toml index fde463245..eee3c8c95 100644 --- a/backend/cmd/config.toml +++ b/backend/cmd/config.toml @@ -35,8 +35,9 @@ 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) +# Disabled: UDP keep-alive settings - Javier Ribal del Río (2026-07-15) +# 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 c236c5dab..c3eea59f0 100644 --- a/backend/cmd/dev-config.toml +++ b/backend/cmd/dev-config.toml @@ -37,8 +37,9 @@ 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) +# Disabled: UDP keep-alive settings - Javier Ribal del Río (2026-07-15) +# 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 9611b7cc1..dba1f1111 100644 --- a/backend/cmd/setup_transport.go +++ b/backend/cmd/setup_transport.go @@ -141,22 +141,25 @@ func configureUDPServerTransport( ) { trace.Info().Msg("Starting UDP server") - onKeepAliveTimeout := func(ip string) { - transp.SendFault() - board, ok := transp.TargetFromIp(ip) - if !ok { - board = "unknown" - } - err := fmt.Errorf("UDP keep-alive timeout: no packets received from board %s (%s) for %dms, fault sent", board, ip, config.UDP.KeepAliveTimeoutMs) - transp.ReportError(err) - - // Close the board's TCP connection ourselves: the fault we just - // broadcast leaves unacked data on a dead peer, which suppresses the - // TCP keep-alive and would delay disconnect detection by minutes. - if ok { - transp.DisconnectTarget(board, err) - } - } + // Disabled: UDP keep-alive callback — on timeout it broadcast a fault, + // reported the board to the GUI and force-closed its TCP connection + // - Javier Ribal del Río (2026-07-15) + // onKeepAliveTimeout := func(ip string) { + // transp.SendFault() + // board, ok := transp.TargetFromIp(ip) + // if !ok { + // board = "unknown" + // } + // err := fmt.Errorf("UDP keep-alive timeout: no packets received from board %s (%s) for %dms, fault sent", board, ip, config.UDP.KeepAliveTimeoutMs) + // transp.ReportError(err) + // + // // Close the board's TCP connection ourselves: the fault we just + // // broadcast leaves unacked data on a dead peer, which suppresses the + // // TCP keep-alive and would delay disconnect detection by minutes. + // if ok { + // transp.DisconnectTarget(board, err) + // } + // } udpServer := udp.NewServer( adj.Info.Addresses[BACKEND], @@ -164,9 +167,10 @@ func configureUDPServerTransport( &trace.Logger, config.UDP.RingBufferSize, config.UDP.PacketChanSize, - time.Duration(config.UDP.KeepAliveCheckIntervalMs)*time.Millisecond, - time.Duration(config.UDP.KeepAliveTimeoutMs)*time.Millisecond, - onKeepAliveTimeout, + // Disabled: UDP keep-alive arguments - Javier Ribal del Río (2026-07-15) + // time.Duration(config.UDP.KeepAliveCheckIntervalMs)*time.Millisecond, + // time.Duration(config.UDP.KeepAliveTimeoutMs)*time.Millisecond, + // onKeepAliveTimeout, ) err := udpServer.Start() if err != nil { diff --git a/backend/internal/config/config_types.go b/backend/internal/config/config_types.go index b0957b0ef..e8fbdb648 100644 --- a/backend/internal/config/config_types.go +++ b/backend/internal/config/config_types.go @@ -25,10 +25,11 @@ type TCP struct { } type UDP struct { - 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"` + RingBufferSize int `toml:"ring_buffer_size"` + PacketChanSize int `toml:"packet_chan_size"` + // Disabled: UDP keep-alive settings - Javier Ribal del Río (2026-07-15) + // 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/tcp/conn.go b/backend/pkg/transport/network/tcp/conn.go index 6dff6feb3..dba140674 100644 --- a/backend/pkg/transport/network/tcp/conn.go +++ b/backend/pkg/transport/network/tcp/conn.go @@ -44,10 +44,6 @@ func (conn *connWithErr) Close() error { return conn.Conn.Close() } -// CloseWithError closes conn, delivering reason to its error channel first so -// the connection handler blocked on that channel wakes up. Closing a -// connection directly would not do this, since Read/Write filter out -// net.ErrClosed. func CloseWithError(conn net.Conn, reason error) error { if cwe, ok := conn.(*connWithErr); ok { select { diff --git a/backend/pkg/transport/network/udp/server.go b/backend/pkg/transport/network/udp/server.go index 7453e1278..0b4892cd1 100644 --- a/backend/pkg/transport/network/udp/server.go +++ b/backend/pkg/transport/network/udp/server.go @@ -35,37 +35,44 @@ type Server struct { ringMutex sync.Mutex notEmpty *sync.Cond - lastSeen map[string]time.Time - lastSeenMu sync.Mutex - keepAliveCheckInterval time.Duration - keepAliveTimeout time.Duration - OnDisconnect func(ip string) + // Disabled: state for the UDP keep-alive, which checked that every source + // IP kept sending packets and reported disconnections - Javier Ribal del Río (2026-07-15) + // lastSeen map[string]time.Time + // lastSeenMu sync.Mutex + // keepAliveCheckInterval time.Duration + // keepAliveTimeout time.Duration + // OnDisconnect func(ip string) } -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(ip string)) *Server { - if keepAliveCheckInterval <= 0 { - keepAliveCheckInterval = defaultKeepAliveCheckInterval - } - if keepAliveTimeout <= 0 { - keepAliveTimeout = defaultKeepAliveTimeout - } +// Disabled: default intervals for the UDP keep-alive - Javier Ribal del Río (2026-07-15) +// const ( +// defaultKeepAliveCheckInterval = 5 * time.Millisecond +// defaultKeepAliveTimeout = 100 * time.Millisecond +// ) + +func NewServer(address string, port uint16, logger *zerolog.Logger, ringBufferSize int, packetChanSize int) *Server { + // Disabled: UDP keep-alive parameters and their defaulting; the previous + // signature also took keepAliveCheckInterval, keepAliveTimeout and an + // onDisconnect callback - Javier Ribal del Río (2026-07-15) + // 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{}), - lastSeen: make(map[string]time.Time), - keepAliveCheckInterval: keepAliveCheckInterval, - keepAliveTimeout: keepAliveTimeout, - OnDisconnect: onDisconnect, + address: address, + port: port, + logger: logger, + packetsCh: make(chan Packet, packetChanSize), + errorsCh: make(chan error, 100), + stopCh: make(chan struct{}), + // Disabled: UDP keep-alive state initialization - Javier Ribal del Río (2026-07-15) + // lastSeen: make(map[string]time.Time), + // keepAliveCheckInterval: keepAliveCheckInterval, + // keepAliveTimeout: keepAliveTimeout, + // OnDisconnect: onDisconnect, } s.ring = make([]Packet, ringBufferSize) @@ -93,9 +100,10 @@ func (s *Server) Start() error { Uint16("port", s.port). Msg("UDP server started") - go s.readLoop() // Read incoming UDP packets - go s.dispatchLoop() // Dispatch packets from ring buffer to channel - go s.keepAliveLoop() // Monitor keep-alive timeouts + go s.readLoop() // Read incoming UDP packets + go s.dispatchLoop() // Dispatch packets from ring buffer to channel + // Disabled: goroutine that monitored UDP keep-alive timeouts - Javier Ribal del Río (2026-07-15) + // go s.keepAliveLoop() return nil } @@ -144,10 +152,11 @@ 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() + // Disabled: recorded when each source IP was last seen, feeding + // the UDP keep-alive check - Javier Ribal del Río (2026-07-15) + // s.lastSeenMu.Lock() + // s.lastSeen[addr.IP.String()] = packet.Timestamp + // s.lastSeenMu.Unlock() // Push packet to ring buffer s.push(packet) @@ -155,42 +164,44 @@ func (s *Server) readLoop() { } } -// 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 { - // Run the callback on its own goroutine so a slow - // handler (e.g. blocking TCP writes) cannot stall - // timeout detection for the remaining IPs - go s.OnDisconnect(ip) - } - } - } - } -} +// Disabled: UDP keep-alive loop — checked every keepAliveCheckInterval whether +// any source IP had gone more than keepAliveTimeout without sending a packet, +// and fired OnDisconnect for it - Javier Ribal del Río (2026-07-15) +// 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 { +// // Run the callback on its own goroutine so a slow +// // handler (e.g. blocking TCP writes) cannot stall +// // timeout detection for the remaining IPs +// go s.OnDisconnect(ip) +// } +// } +// } +// } +// } func (s *Server) GetPackets() <-chan Packet { return s.packetsCh diff --git a/backend/pkg/transport/transport.go b/backend/pkg/transport/transport.go index 7fb9c008c..9c65cbeb0 100644 --- a/backend/pkg/transport/transport.go +++ b/backend/pkg/transport/transport.go @@ -224,11 +224,11 @@ func (transport *Transport) readLoopTCPConn(conn net.Conn, logger zerolog.Logger for { packet, err := transport.decoder.DecodeNext(conn) if err != nil { - // net.ErrClosed means we closed the connection on purpose - // (e.g. UDP keep-alive timeout), and a fault was already sent - if errors.Is(err, net.ErrClosed) { - return - } + // Disabled: skipped the fault when we closed the connection on + // purpose from the UDP keep-alive - Javier Ribal del Río (2026-07-15) + // if errors.Is(err, net.ErrClosed) { + // return + // } logger.Error().Stack().Err(err).Msg("decode") transport.errChan <- err transport.SendFault() @@ -485,33 +485,32 @@ func (transport *Transport) consumeErrors() { } } -// ReportError forwards an error to the API as an error notification so it -// shows up in the GUI message log. -func (transport *Transport) ReportError(err error) { - transport.errChan <- err -} - -// TargetFromIp returns the board (transport target) registered for the given IP. -func (transport *Transport) TargetFromIp(ip string) (abstraction.TransportTarget, bool) { - target, ok := transport.ipToTarget[ip] - return target, ok -} - -// DisconnectTarget forcefully closes the TCP connection to target, if any. -// The connection handler wakes up with reason, cleans up and notifies the -// disconnection, and the client reconnection loop takes over. -func (transport *Transport) DisconnectTarget(target abstraction.TransportTarget, reason error) bool { - transport.connectionsMx.RLock() - conn, ok := transport.connections[target] - transport.connectionsMx.RUnlock() - if !ok { - return false - } - - transport.logger.Warn().Str("target", string(target)).Err(reason).Msg("forcefully disconnecting target") - tcp.CloseWithError(conn, reason) - return true -} +// Disabled: helpers for the UDP keep-alive callback — ReportError surfaced an +// error in the GUI message log, TargetFromIp mapped a source IP to its board, +// and DisconnectTarget force-closed a board's TCP connection so the handler +// woke up, cleaned up and the reconnection loop took over +// - Javier Ribal del Río (2026-07-15) +// func (transport *Transport) ReportError(err error) { +// transport.errChan <- err +// } +// +// func (transport *Transport) TargetFromIp(ip string) (abstraction.TransportTarget, bool) { +// target, ok := transport.ipToTarget[ip] +// return target, ok +// } +// +// func (transport *Transport) DisconnectTarget(target abstraction.TransportTarget, reason error) bool { +// transport.connectionsMx.RLock() +// conn, ok := transport.connections[target] +// transport.connectionsMx.RUnlock() +// if !ok { +// return false +// } +// +// transport.logger.Warn().Str("target", string(target)).Err(reason).Msg("forcefully disconnecting target") +// tcp.CloseWithError(conn, reason) +// return true +// } func (transport *Transport) SendFault() { err := transport.SendMessage(NewPacketMessage(data.NewPacket(0))) From dd2f24c2bc8794f07851f1f485241f54f8c2d9c9 Mon Sep 17 00:00:00 2001 From: Javier Ribal del Rio Date: Wed, 15 Jul 2026 16:04:04 +0200 Subject: [PATCH 2/2] feat(backend, ADJ): implement TCP adj commit check --- backend/cmd/main.go | 1 + backend/pkg/adj/adj.go | 9 +++++-- backend/pkg/adj/git.go | 29 +++++++++++++++++++++ backend/pkg/transport/transport.go | 42 ++++++++++++++++++++++++++++++ 4 files changed, 79 insertions(+), 2 deletions(-) diff --git a/backend/cmd/main.go b/backend/cmd/main.go index 5ce0e5126..e421ed617 100644 --- a/backend/cmd/main.go +++ b/backend/cmd/main.go @@ -97,6 +97,7 @@ func main() { // <--- transport ---> transp := transport.NewTransport(trace.Logger) transp.SetpropagateFault(config.Transport.PropagateFault) + transp.SetADJHash(adj.Commit) // <--- vehicle ---> err = configureVehicle( diff --git a/backend/pkg/adj/adj.go b/backend/pkg/adj/adj.go index 8a496f423..156eea188 100644 --- a/backend/pkg/adj/adj.go +++ b/backend/pkg/adj/adj.go @@ -96,9 +96,14 @@ func downloadADJ(AdjSettings config.Adj) (string, json.RawMessage, json.RawMessa return "local", nil, nil, err } - // If not downloading use local ADJ + // If not downloading, recover the commit hash persisted by the last + // successful download. Without it the backend must not start: boards + // receive this hash on connection and could not validate their ADJ. if commitHash == "" { - commitHash = "local" + commitHash, err = readCommitFile() + if err != nil { + return "", nil, nil, fmt.Errorf("using local ADJ but no stored commit hash was found (run once with an adj branch and internet access): %w", err) + } } // After downloading adj apply adj validator diff --git a/backend/pkg/adj/git.go b/backend/pkg/adj/git.go index ab17f1f28..9e7cb208a 100644 --- a/backend/pkg/adj/git.go +++ b/backend/pkg/adj/git.go @@ -1,8 +1,10 @@ package adj import ( + "fmt" "os" "path/filepath" + "strings" trace "github.com/rs/zerolog/log" @@ -10,6 +12,29 @@ import ( "github.com/go-git/go-git/v5/plumbing" ) +// commitFile persists the commit hash of the last downloaded ADJ. It lives +// next to the repo (not inside it) so wiping the repo before a clone does not +// lose it, and local runs (no branch or no internet) can still report it. +var commitFile = filepath.Join(filepath.Dir(filepath.Clean(RepoPath)), "adj_commit") + +// writeCommitFile stores the commit hash of the freshly cloned ADJ +func writeCommitFile(hash string) error { + return os.WriteFile(commitFile, []byte(hash), 0644) +} + +// readCommitFile recovers the commit hash persisted by the last download +func readCommitFile() (string, error) { + raw, err := os.ReadFile(commitFile) + if err != nil { + return "", err + } + hash := strings.TrimSpace(string(raw)) + if hash == "" { + return "", fmt.Errorf("commit file %s is empty", commitFile) + } + return hash, nil +} + // updateRepo ensures that the local ADJ repository matches the specified remote branch. // It first performs a test clone to verify remote accessibility (including internet // connectivity). If the remote branch is accessible, the local repository is completely @@ -85,5 +110,9 @@ func updateRepo(AdjBranch string) (string, error) { } } + if err = writeCommitFile(commitHash); err != nil { + return "", err + } + return commitHash, nil } diff --git a/backend/pkg/transport/transport.go b/backend/pkg/transport/transport.go index 9c65cbeb0..ef6c8c896 100644 --- a/backend/pkg/transport/transport.go +++ b/backend/pkg/transport/transport.go @@ -2,6 +2,7 @@ package transport import ( "bytes" + "encoding/binary" "errors" "fmt" "io" @@ -36,6 +37,8 @@ type Transport struct { propagateFault bool + onConnectOrderValue [8]byte + api abstraction.TransportAPI logger zerolog.Logger @@ -132,6 +135,8 @@ func (transport *Transport) handleTCPConn(conn net.Conn) error { transport.api.ConnectionUpdate(target, true) defer transport.api.ConnectionUpdate(target, false) + transport.sendOnConnectOrder(conn, connectionLogger) + transport.readLoopTCPConn(conn, connectionLogger) err = <-errChan @@ -142,6 +147,43 @@ func (transport *Transport) handleTCPConn(conn net.Conn) error { return err } +// Hardcoded order sent to a board right after its TCP connection is +// established, carrying the short ADJ hash. TODO: set the real id +const onConnectOrderId uint16 = 65535 + +// SetADJHash stores the ADJ commit hash to be sent on each new TCP connection. +// The short hash is the first 8 characters of the commit, sent as ASCII text. +// A hash shorter than 8 characters is sent as zeroes. +func (transport *Transport) SetADJHash(hash string) { + if len(hash) < 8 { + transport.logger.Warn().Str("hash", hash).Msg("adj hash too short, on-connect order will send zeroes") + return + } + copy(transport.onConnectOrderValue[:], hash[:8]) +} + +// sendOnConnectOrder writes the on-connect order (carrying the short ADJ hash) +// to the board using the same wire format as regular orders: packet id (little +// endian) followed by the 8 ASCII characters of the hash. +func (transport *Transport) sendOnConnectOrder(conn net.Conn, logger zerolog.Logger) { + buf := new(bytes.Buffer) + binary.Write(buf, binary.LittleEndian, onConnectOrderId) + binary.Write(buf, binary.LittleEndian, transport.onConnectOrderValue) + + data := buf.Bytes() + totalWritten := 0 + for totalWritten < len(data) { + n, err := conn.Write(data[totalWritten:]) + totalWritten += n + if err != nil { + logger.Error().Stack().Err(err).Msg("write on-connect order") + transport.errChan <- err + return + } + } + logger.Info().Uint16("id", onConnectOrderId).Msg("sent on-connect order") +} + // configureTCPConn sets TCP-level options like linger and no-delay. func (transport *Transport) configureTCPConn(conn net.Conn) { tcpConn, ok := conn.(*net.TCPConn)