fix(ultrabeam): flush stale replies (not seq-match) + instant moving flag
The antenna does NOT echo our sequence number — its replies carry their own counter — so the previous seq-matching drained every reply and stalled status updates. Revert to reading one reply per command, but flush any bytes left in the stream before each command (drainStale): a reply left by a timed-out command is discarded so the next read stays 1:1. readPacket also resyncs to the next STX. This fixes the intermittent disconnects, phantom frequency jumps and wrong element-length readings without depending on the seq. Also: report motion for a short window right after a commanded move, so the "moving" indicator and the Flex TX-inhibit fire the instant a band/pattern is clicked instead of a poll (~2 s) later; the real motor state takes over once polled.
This commit is contained in:
@@ -88,8 +88,22 @@ type Client struct {
|
||||
// yet (Frequency==0) — otherwise the deadband is bypassed and every small QSY
|
||||
// re-tunes the motors.
|
||||
lastSetKHz int
|
||||
|
||||
// moveCmdAt is when a move (frequency or direction) was last COMMANDED. The
|
||||
// status poll runs every couple of seconds, so without this the "moving" flag
|
||||
// — which drives the UI indicator and the Flex TX-inhibit — appeared up to a
|
||||
// poll late even though the elements start moving at once. GetStatus reports
|
||||
// motion during a short window after a command so both react immediately.
|
||||
moveCmdAt time.Time
|
||||
}
|
||||
|
||||
// ubMoveOptimisticWindow is how long after a commanded move GetStatus reports
|
||||
// "moving" before the status poll has had a chance to read the real motor state.
|
||||
// It only needs to bridge one poll interval; once a poll sees real motion, that
|
||||
// takes over. Bounded, so if the antenna never reports motion the flag still
|
||||
// clears rather than latching the TX-inhibit on for ever.
|
||||
const ubMoveOptimisticWindow = 3 * time.Second
|
||||
|
||||
// LastSetKHz returns the frequency (kHz) most recently commanded to the antenna,
|
||||
// or 0 if none yet.
|
||||
func (c *Client) LastSetKHz() int {
|
||||
@@ -270,7 +284,19 @@ func (c *Client) GetStatus() (*Status, error) {
|
||||
return &Status{Connected: false}, nil
|
||||
}
|
||||
|
||||
return c.lastStatus, nil
|
||||
// Copy so the optimistic-motion tweak below never mutates the cached status
|
||||
// the poll goroutine owns.
|
||||
st := *c.lastStatus
|
||||
// Optimistic motion: right after a commanded move, report "moving" until a
|
||||
// status poll can read the real motor state (~one poll interval). The elements
|
||||
// start moving the instant the operator clicks a band/pattern, so this makes
|
||||
// the UI indicator and the Flex TX-inhibit react immediately instead of a poll
|
||||
// later. Real polled motion takes over once seen; the window is bounded so the
|
||||
// flag can't latch the inhibit on for ever.
|
||||
if st.Connected && st.MotorsMoving == 0 && !c.moveCmdAt.IsZero() && time.Since(c.moveCmdAt) < ubMoveOptimisticWindow {
|
||||
st.MotorsMoving = 1 // sentinel: optimistically moving (read only as != 0)
|
||||
}
|
||||
return &st, nil
|
||||
}
|
||||
|
||||
// getNextSeq returns the next sequence number
|
||||
@@ -388,64 +414,67 @@ func (c *Client) sendCommand(cmd byte, data []byte) ([]byte, error) {
|
||||
return nil, fmt.Errorf("not connected")
|
||||
}
|
||||
|
||||
// Build and send packet, remembering the seq so the reply can be matched to it.
|
||||
// Flush anything already sitting in the stream before we send. The antenna
|
||||
// does NOT echo our sequence number — its replies carry their own counter — so
|
||||
// a reply cannot be matched to its request. Instead we keep the exchange 1:1:
|
||||
// a reply left behind by an earlier timed-out command is discarded here, so
|
||||
// the reply we read next belongs to THIS command. Reading a stale reply as the
|
||||
// current one crossed STATUS with READ_BANDS/PROGRESS — phantom frequencies (a
|
||||
// spurious follow-loop re-tune), dropped connections and wrong element lengths.
|
||||
c.drainStale()
|
||||
|
||||
seq := c.getNextSeq()
|
||||
packet := c.buildPacket(seq, cmd, data)
|
||||
|
||||
_, err := c.conn.Write(packet)
|
||||
if err != nil {
|
||||
if _, err := c.conn.Write(packet); err != nil {
|
||||
return nil, fmt.Errorf("failed to write: %w", err)
|
||||
}
|
||||
|
||||
// Read reply with timeout (generous — tolerates remote-link latency). The
|
||||
// deadline bounds the WHOLE exchange, including draining any stale replies.
|
||||
// Read the reply with a timeout generous enough for a remote link.
|
||||
c.conn.SetReadDeadline(time.Now().Add(ubReadTimeout))
|
||||
buffer, err := c.readPacket()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, replyCmd, payload, err := parsePacket(buffer)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to parse reply: %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
|
||||
}
|
||||
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)
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
// drainStale discards any bytes already waiting in the stream — a reply left
|
||||
// behind by a command that timed out. A short read deadline lets it consume what
|
||||
// is there and stop quickly when the stream is clean. Caller holds connMu.
|
||||
func (c *Client) drainStale() {
|
||||
c.conn.SetReadDeadline(time.Now().Add(5 * time.Millisecond))
|
||||
buf := make([]byte, 256)
|
||||
for {
|
||||
n, err := c.reader.Read(buf)
|
||||
if n == 0 || err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
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
|
||||
@@ -560,6 +589,7 @@ func (c *Client) SetFrequency(freqKhz int, direction int) error {
|
||||
c.statusMu.Lock()
|
||||
c.pendingDir, c.pendingDirAt, c.pendingDirSet = direction, time.Now(), true
|
||||
c.lastSetKHz = freqKhz
|
||||
c.moveCmdAt = time.Now() // start reporting motion at once — see ubMoveOptimisticWindow
|
||||
if c.lastStatus != nil {
|
||||
c.lastStatus.Direction = direction // reflect immediately
|
||||
}
|
||||
|
||||
@@ -5,14 +5,14 @@ import (
|
||||
"bytes"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 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) {
|
||||
// sendCommand must flush any bytes already in the stream (a reply left by an
|
||||
// earlier timed-out command) before reading, then return the reply to THIS
|
||||
// command. The antenna does not echo our sequence number, so keeping the stream
|
||||
// clean is the only way to stay in sync.
|
||||
func TestSendCommandFlushesStaleThenReadsReply(t *testing.T) {
|
||||
srvConn, cliConn := net.Pipe()
|
||||
defer srvConn.Close()
|
||||
defer cliConn.Close()
|
||||
@@ -22,8 +22,14 @@ func TestSendCommandDrainsStaleReplyAndMatchesSeq(t *testing.T) {
|
||||
|
||||
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.
|
||||
// Leftover from a "previous" command still sitting in the stream. The Write
|
||||
// blocks until drainStale consumes it, so it is guaranteed flushed before
|
||||
// the real exchange.
|
||||
if _, err := srvConn.Write(c.buildPacket(9, UB_OK, []byte{0xDE, 0xAD})); err != nil {
|
||||
errc <- err
|
||||
return
|
||||
}
|
||||
// Now serve the actual request.
|
||||
req, err := srv.readPacket()
|
||||
if err != nil {
|
||||
errc <- err
|
||||
@@ -34,16 +40,8 @@ func TestSendCommandDrainsStaleReplyAndMatchesSeq(t *testing.T) {
|
||||
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
|
||||
_, err = srvConn.Write(c.buildPacket(seq, UB_OK, []byte{0x11, 0x22, 0x33}))
|
||||
errc <- err
|
||||
}()
|
||||
|
||||
payload, err := c.sendCommand(CMD_STATUS, nil)
|
||||
@@ -51,7 +49,7 @@ func TestSendCommandDrainsStaleReplyAndMatchesSeq(t *testing.T) {
|
||||
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)
|
||||
t.Fatalf("payload = % X, want 11 22 33 — stale reply not flushed", payload)
|
||||
}
|
||||
if err := <-errc; err != nil {
|
||||
t.Fatalf("server: %v", err)
|
||||
@@ -78,3 +76,19 @@ func TestReadPacketResyncsToSTX(t *testing.T) {
|
||||
t.Fatalf("seq=%d cmd=%d payload=% X, want 5 / OK / 01 02", seq, cmd, payload)
|
||||
}
|
||||
}
|
||||
|
||||
// A commanded move must report motion immediately (before the next status poll),
|
||||
// then fall back to the real motor state once the window elapses.
|
||||
func TestOptimisticMotionAfterCommand(t *testing.T) {
|
||||
c := &Client{lastStatus: &Status{Connected: true, MotorsMoving: 0}}
|
||||
|
||||
c.moveCmdAt = time.Now()
|
||||
if st, _ := c.GetStatus(); st.MotorsMoving == 0 {
|
||||
t.Fatal("just after a move command, GetStatus should report motion")
|
||||
}
|
||||
|
||||
c.moveCmdAt = time.Now().Add(-ubMoveOptimisticWindow - time.Second)
|
||||
if st, _ := c.GetStatus(); st.MotorsMoving != 0 {
|
||||
t.Fatal("past the window with motors idle, GetStatus must report no motion")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user