fix(ultrabeam): match replies to requests by seq (stop stream desync)

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.
2026-08-04 23:56:42 +02:00
parent 058f164ab6
commit 98f11ee3d0
3 changed files with 150 additions and 45 deletions
@@ -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."
    ]
  },
  {
@@ -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,42 +388,39 @@ 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
-	var buffer []byte
-	for {
-		b, err := c.reader.ReadByte()
-		if err != nil {
-			return nil, fmt.Errorf("failed to read: %w", err)
+	// 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
 		}
-
-		buffer = append(buffer, b)
-
-		// Check if we have a complete packet
-		if b == ETX && len(buffer) > 0 && buffer[0] == STX {
-			break
+		replySeq, replyCmd, payload, perr := parsePacket(buffer)
+		if perr != nil {
+			log.Printf("Ultrabeam: discarding malformed reply: %v", perr)
+			continue
 		}
-
-		// 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)
+		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
@@ -449,6 +445,33 @@ func (c *Client) sendCommand(cmd byte, data []byte) ([]byte, error) {
 			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)
+		if b == ETX {
+			return buffer, nil
+		}
+		if len(buffer) > 256 {
+			return nil, fmt.Errorf("packet too long")
+		}
+	}
+}

 // queryStatus queries general status (command 1)
 func (c *Client) queryStatus() (*Status, error) {
@@ -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)
+	}
+}