From 98f11ee3d0b8e6635a2d441e1a16bd10df9befa7 Mon Sep 17 00:00:00 2001 From: Gregory Salaun Date: Tue, 4 Aug 2026 23:56:42 +0200 Subject: [PATCH] fix(ultrabeam): match replies to requests by seq (stop stream desync) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sendCommand discarded the reply's sequence number and accepted whatever frame came back. On a slow remote link a command that timed out left its late reply in the stream, and the NEXT command read it as its own — crossing STATUS with READ_BANDS/PROGRESS. That surfaced as phantom frequency jumps (a spurious follow-loop re-tune), intermittent "reply too short" disconnects, and wrong element-length readings (READ_BANDS getting the status frame). Now every reply is matched to its request seq and stale/malformed frames are drained (bounded), with readPacket resyncing to the next STX. Tested with a net.Pipe that injects a stale reply ahead of the real one. --- changelog.json | 6 +- internal/ultrabeam/ultrabeam.go | 109 ++++++++++++++++----------- internal/ultrabeam/ultrabeam_test.go | 80 ++++++++++++++++++++ 3 files changed, 150 insertions(+), 45 deletions(-) create mode 100644 internal/ultrabeam/ultrabeam_test.go diff --git a/changelog.json b/changelog.json index dc61de0..180aef3 100644 --- a/changelog.json +++ b/changelog.json @@ -3,10 +3,12 @@ "version": "0.23.6", "date": "", "en": [ - "Log viewer: the window now keeps about four times more history (1 MB instead of 256 KB, ~6500 lines). During a busy trace the oldest lines used to scroll out of the buffer while you were still reading them; the larger window holds them." + "Log viewer: the window now keeps about four times more history (1 MB instead of 256 KB, ~6500 lines). During a busy trace the oldest lines used to scroll out of the buffer while you were still reading them; the larger window holds them.", + "Ultrabeam over a remote link: fixed intermittent connection drops, phantom frequency jumps and wrong element-length readings. Replies weren't matched to the command that asked for them, so on a slow link a status query could pick up a lingering reply meant for another command — reading, say, the antenna's status where element lengths were expected. Every reply is now matched to its request by sequence number, and stale replies are discarded." ], "fr": [ - "Visionneuse de log : la fenêtre conserve environ quatre fois plus d'historique (1 Mo au lieu de 256 Ko, ~6500 lignes). Lors d'une trace chargée, les plus vieilles lignes défilaient hors du buffer pendant qu'on les lisait encore ; la fenêtre agrandie les garde." + "Visionneuse de log : la fenêtre conserve environ quatre fois plus d'historique (1 Mo au lieu de 256 Ko, ~6500 lignes). Lors d'une trace chargée, les plus vieilles lignes défilaient hors du buffer pendant qu'on les lisait encore ; la fenêtre agrandie les garde.", + "Ultrabeam en remote : coupures de connexion intermittentes, sauts de fréquence fantômes et longueurs d'éléments erronées corrigés. Les réponses n'étaient pas associées à la commande qui les demandait : sur un lien lent, une requête de statut pouvait récupérer une réponse traînante destinée à une autre commande — lisant par exemple le statut de l'antenne là où on attendait les longueurs d'éléments. Chaque réponse est désormais appariée à sa requête par numéro de séquence, et les réponses périmées sont ignorées." ] }, { diff --git a/internal/ultrabeam/ultrabeam.go b/internal/ultrabeam/ultrabeam.go index 248e9a9..b533e23 100644 --- a/internal/ultrabeam/ultrabeam.go +++ b/internal/ultrabeam/ultrabeam.go @@ -301,10 +301,9 @@ func quoteByte(b byte) []byte { return []byte{b} } -// buildPacket creates a complete packet with checksum and escaping -func (c *Client) buildPacket(cmd byte, data []byte) []byte { - seq := c.getNextSeq() - +// buildPacket creates a complete packet with checksum and escaping. The seq is +// supplied by the caller so sendCommand can match the reply against it. +func (c *Client) buildPacket(seq, cmd byte, data []byte) []byte { // Calculate checksum on unquoted data payload := append([]byte{seq, cmd}, data...) chk := calculateChecksum(payload) @@ -389,65 +388,89 @@ func (c *Client) sendCommand(cmd byte, data []byte) ([]byte, error) { return nil, fmt.Errorf("not connected") } - // Build and send packet - packet := c.buildPacket(cmd, data) + // Build and send packet, remembering the seq so the reply can be matched to it. + seq := c.getNextSeq() + packet := c.buildPacket(seq, cmd, data) _, err := c.conn.Write(packet) if err != nil { return nil, fmt.Errorf("failed to write: %w", err) } - // Read reply with timeout (generous — tolerates remote-link latency). + // Read reply with timeout (generous — tolerates remote-link latency). The + // deadline bounds the WHOLE exchange, including draining any stale replies. c.conn.SetReadDeadline(time.Now().Add(ubReadTimeout)) - // Read until we get a complete packet + // Match the reply to our seq. A command that timed out earlier can leave its + // late reply sitting in the stream; without this check THIS command would read + // that stale reply as its own — crossing STATUS with READ_BANDS/PROGRESS, which + // showed up as phantom frequencies (a spurious follow-loop re-tune) and dropped + // connections. So read frames until one carries our seq, discarding the rest. + // Bounded, so a stream that never yields our reply errors instead of looping. + const maxDrain = 16 + for attempts := 0; attempts < maxDrain; attempts++ { + buffer, rerr := c.readPacket() + if rerr != nil { + return nil, rerr + } + replySeq, replyCmd, payload, perr := parsePacket(buffer) + if perr != nil { + log.Printf("Ultrabeam: discarding malformed reply: %v", perr) + continue + } + if replySeq != seq { + log.Printf("Ultrabeam: draining stale reply (seq %d, want %d) — likely a prior timed-out command", replySeq, seq) + continue + } + + // Log for debugging unknown codes + if replyCmd != UB_OK && replyCmd != UB_BAD && replyCmd != UB_PAR && replyCmd != UB_ERR { + log.Printf("Ultrabeam: Unknown reply code %d (0x%02X), raw packet: %v", replyCmd, replyCmd, buffer) + } + + // Check for errors + switch replyCmd { + case UB_BAD: + return nil, fmt.Errorf("invalid command") + case UB_PAR: + return nil, fmt.Errorf("bad parameters") + case UB_ERR: + return nil, fmt.Errorf("execution error") + case UB_OK: + return payload, nil + default: + // Unknown codes might indicate "busy" or "in progress" + // Treat as non-fatal, return empty payload + log.Printf("Ultrabeam: Unusual reply code %d, treating as busy/in-progress", replyCmd) + return []byte{}, nil + } + } + return nil, fmt.Errorf("no reply matching seq %d after draining %d frames", seq, maxDrain) +} + +// readPacket reads one complete STX…ETX frame, skipping any leading bytes until +// an STX so a partial/garbage remnant left in the stream can't derail the parse. +// STX/ETX/DLE are all high-bit (0xF5/0xFA/0xF6) and the escaping clears the MSB, +// so a raw ETX only ever appears as the real terminator. Caller holds connMu and +// has set a read deadline. +func (c *Client) readPacket() ([]byte, error) { var buffer []byte for { b, err := c.reader.ReadByte() if err != nil { return nil, fmt.Errorf("failed to read: %w", err) } - + if len(buffer) == 0 && b != STX { + continue // resync to the start of a frame + } buffer = append(buffer, b) - - // Check if we have a complete packet - if b == ETX && len(buffer) > 0 && buffer[0] == STX { - break + if b == ETX { + return buffer, nil } - - // Prevent infinite loop if len(buffer) > 256 { return nil, fmt.Errorf("packet too long") } } - - // Parse reply - _, replyCmd, payload, err := parsePacket(buffer) - if err != nil { - return nil, fmt.Errorf("failed to parse reply: %w", err) - } - - // Log for debugging unknown codes - if replyCmd != UB_OK && replyCmd != UB_BAD && replyCmd != UB_PAR && replyCmd != UB_ERR { - log.Printf("Ultrabeam: Unknown reply code %d (0x%02X), raw packet: %v", replyCmd, replyCmd, buffer) - } - - // Check for errors - switch replyCmd { - case UB_BAD: - return nil, fmt.Errorf("invalid command") - case UB_PAR: - return nil, fmt.Errorf("bad parameters") - case UB_ERR: - return nil, fmt.Errorf("execution error") - case UB_OK: - return payload, nil - default: - // Unknown codes might indicate "busy" or "in progress" - // Treat as non-fatal, return empty payload - log.Printf("Ultrabeam: Unusual reply code %d, treating as busy/in-progress", replyCmd) - return []byte{}, nil - } } // queryStatus queries general status (command 1) diff --git a/internal/ultrabeam/ultrabeam_test.go b/internal/ultrabeam/ultrabeam_test.go new file mode 100644 index 0000000..ebf3f52 --- /dev/null +++ b/internal/ultrabeam/ultrabeam_test.go @@ -0,0 +1,80 @@ +package ultrabeam + +import ( + "bufio" + "bytes" + "net" + "testing" +) + +// A stale reply left in the stream by an earlier timed-out command must be +// DRAINED, not taken for this command's reply. Without seq matching, a STATUS +// query would return a lingering READ_BANDS/PROGRESS payload — which showed up +// on a remote link as phantom frequencies (a spurious follow-loop re-tune) and +// dropped connections. +func TestSendCommandDrainsStaleReplyAndMatchesSeq(t *testing.T) { + srvConn, cliConn := net.Pipe() + defer srvConn.Close() + defer cliConn.Close() + + c := &Client{conn: cliConn, reader: bufio.NewReader(cliConn)} + srv := &Client{reader: bufio.NewReader(srvConn)} + + errc := make(chan error, 1) + go func() { + // Read the request, learn its seq, then answer with a stale reply (wrong + // seq) FIRST, then the real one. + req, err := srv.readPacket() + if err != nil { + errc <- err + return + } + seq, _, _, err := parsePacket(req) + if err != nil { + errc <- err + return + } + staleSeq := (seq + 1) % 128 + if _, err := srvConn.Write(c.buildPacket(staleSeq, UB_OK, []byte{0xAA, 0xBB})); err != nil { + errc <- err + return + } + if _, err := srvConn.Write(c.buildPacket(seq, UB_OK, []byte{0x11, 0x22, 0x33})); err != nil { + errc <- err + return + } + errc <- nil + }() + + payload, err := c.sendCommand(CMD_STATUS, nil) + if err != nil { + t.Fatalf("sendCommand: %v", err) + } + if !bytes.Equal(payload, []byte{0x11, 0x22, 0x33}) { + t.Fatalf("payload = % X, want 11 22 33 — stale reply not drained or seq not matched", payload) + } + if err := <-errc; err != nil { + t.Fatalf("server: %v", err) + } +} + +// readPacket must resynchronise to the next STX, dropping any partial/garbage +// bytes left in the stream, so one corrupt frame can't misalign every frame +// after it. +func TestReadPacketResyncsToSTX(t *testing.T) { + frame := (&Client{}).buildPacket(5, UB_OK, []byte{0x01, 0x02}) + stream := append([]byte{0x11, 0x22, 0x33}, frame...) // leading garbage, then a real frame + c := &Client{reader: bufio.NewReader(bytes.NewReader(stream))} + + got, err := c.readPacket() + if err != nil { + t.Fatalf("readPacket: %v", err) + } + seq, cmd, payload, err := parsePacket(got) + if err != nil { + t.Fatalf("parsePacket: %v", err) + } + if seq != 5 || cmd != UB_OK || !bytes.Equal(payload, []byte{0x01, 0x02}) { + t.Fatalf("seq=%d cmd=%d payload=% X, want 5 / OK / 01 02", seq, cmd, payload) + } +}