Desktop: key queued TX at the slot boundary so fast decodes don't drop the reply

desktop/src-tauri/src/engine.rs

@@ -58,6 +58,55 @@ fn reply_tx_audio_hz(delta_freq: u32) -> Option<i32> {
     let df = i32::try_from(delta_freq).ok()?;
     (MIN_TX_AUDIO_HZ..=MAX_TX_AUDIO_HZ).contains(&df).then_some(df)
 }
+
+/// The core keying gate `maybe_transmit` enforces, factored out so it is pure and
+/// unit-testable and shared with the run loop's boundary trigger. We may key iff:
+///   * a QSO is active,
+///   * a locked TX parity (set when answering a CQ) matches this slot's parity —
+///     `None` leaves us eligible in any slot and locks on the first transmission,
+///   * we have not already transmitted in this slot, and
+///   * we are early enough in the cycle to fit the full 12.64 s waveform before the
+///     next boundary (`into_cycle_ms <= TX_LATEST_MS`); starting later clips the
+///     leading Costas array — audible but undecodable.
+fn tx_slot_eligible(
+    active: bool,
+    tx_parity: Option<i64>,
+    slot_id: i64,
+    txed_slot: i64,
+    into_cycle_ms: i64,
+) -> bool {
+    if !active {
+        return false;
+    }
+    if let Some(p) = tx_parity {
+        if p != slot_id.rem_euclid(2) {
+            return false;
+        }
+    }
+    if txed_slot == slot_id {
+        return false;
+    }
+    into_cycle_ms <= TX_LATEST_MS
+}
+
+/// Whether the run loop should key a queued transmission at this tick. Unlike an
+/// operator command (which keys immediately), the per-tick boundary trigger waits
+/// until the current slot's decodes have been processed (`!awaiting_decode`) so it
+/// never keys a stale message ahead of a slow decode; once they are in,
+/// `handle_decoded` has set the fresh `tx_message` this fires on. This is what lets
+/// a reply computed early in the previous cycle (fast CPU) still go out at the top
+/// of its slot — `maybe_transmit` is otherwise only reached on decode arrival,
+/// which on a quick decode lands mid-slot, past the TX window, dropping the reply.
+fn boundary_tx_ready(
+    active: bool,
+    awaiting_decode: bool,
+    tx_parity: Option<i64>,
+    slot_id: i64,
+    txed_slot: i64,
+    into_cycle_ms: i64,
+) -> bool {
+    !awaiting_decode && tx_slot_eligible(active, tx_parity, slot_id, txed_slot, into_cycle_ms)
+}
 // Live-waterfall FFT parameters (window/size/averaging + display constants)
 // live in `crate::wf` and are runtime-configurable via SetWaterfallConfig.
 // Input RMS at/below this (dBFS) counts as silence — no audio reaching the app.
@@ -304,6 +353,11 @@ struct Engine {
     /// Slot id (rx-corrected clock) most recently handed to the decode worker.
     /// Guards the once-per-slot early decode trigger in the run loop.
     last_decoded_slot: i64,
+    /// True from the moment a slot's audio is handed to the decode worker until its
+    /// decodes come back. Gates the run loop's boundary TX trigger so it never keys
+    /// a stale message ahead of a slow decode (the fresh one is keyed by
+    /// `handle_decoded` when it arrives). See [`boundary_tx_ready`].
+    awaiting_decode: bool,
     /// Audio-capture latency compensation (ms). The RX decode window is sliced
     /// this much later than the UTC cycle boundary so that buffered/late-arriving
     /// input audio lands aligned — drives decoded DT toward 0. Auto-calibrated
@@ -416,6 +470,7 @@ impl Engine {
             tx_audio_hz,
             tx_gain,
             last_decoded_slot: -1,
+            awaiting_decode: false,
             rx_offset_ms,
             last_tick_ms: 0,
             tx_parity: None,
@@ -528,17 +583,42 @@ impl Engine {
                 // front-aligned (signal at sample 0) with the tail zero-padded.
                 let elapsed = (into_rx_cycle * SAMPLE_RATE as i64 / 1000) as usize;
                 let slot = self.accum.take_slot_from_start(elapsed);
-                let _ = self.slot_tx.send(slot);
+                if self.slot_tx.send(slot).is_ok() {
+                    // Hold off the boundary TX trigger until this slot's decodes
+                    // come back, so it can't key a stale message ahead of a slow
+                    // decode (the fresh one is keyed by handle_decoded on arrival).
+                    self.awaiting_decode = true;
+                }
             }
 
             // Act on any decodes the worker has finished (off-loop DSP). This is
             // where steady-state TX is driven: each slot's decodes advance the QSO
             // and then `maybe_transmit` keys up if it's our turn. Immediate keying
             // on an operator tap is handled separately, in the command handlers.
             while let Ok(msgs) = self.dec_rx.try_recv() {
+                self.awaiting_decode = false;
                 self.handle_decoded(msgs, slot_id);
             }
 
+            // Boundary TX trigger: key a queued reply/CQ at the top of its slot even
+            // when this slot's decodes were processed earlier in the previous cycle.
+            // `maybe_transmit` is otherwise only reached on decode arrival, which on a
+            // fast decode lands mid-slot — past the TX window — so the queued message
+            // would never go out (auto-sequence stalls; CQ never retransmits). Gated
+            // on `!awaiting_decode` so a stale message is never keyed before this
+            // slot's decodes are in; `maybe_transmit` re-checks eligibility and is
+            // idempotent per slot (the `txed_slot` guard).
+            if boundary_tx_ready(
+                self.qso.active,
+                self.awaiting_decode,
+                self.tx_parity,
+                slot_id,
+                self.txed_slot,
+                now.rem_euclid(CYCLE_MS),
+            ) {
+                self.maybe_transmit(slot_id);
+            }
+
             // Drop PTT as soon as the TX waveform finishes clocking out (or a
             // stalled device trips the deadline). Polling here — rather than
             // blocking inside `transmit` — is what lets Stop TX take effect
@@ -620,26 +700,21 @@ impl Engine {
     /// (`txed_slot` guard), so calling it from both `handle_decoded` and a command
     /// handler in the same slot transmits at most once.
     fn maybe_transmit(&mut self, slot_id: i64) {
-        if !self.qso.active {
-            return;
-        }
-        let parity = slot_id.rem_euclid(2);
-        // Respect a locked alternation. Answering a CQ pins the parity to the
-        // operator's click slot (set in the command handler) so we always reply
-        // opposite the DX; a CQ / free-text start leaves it `None`, eligible in
-        // any slot, and locks on its first transmission below.
-        if matches!(self.tx_parity, Some(p) if p != parity) {
-            return;
-        }
-        if self.txed_slot == slot_id {
-            return;
-        }
-        // Too late in the cycle to start cleanly — wait for our next eligible
-        // slot rather than transmit a clipped, undecodable signal.
-        if self.now().rem_euclid(CYCLE_MS) > TX_LATEST_MS {
+        // All the keying rules — active QSO, locked-parity alternation, once per
+        // slot, and early enough in the cycle to fit the waveform — live in the pure
+        // `tx_slot_eligible` so the run loop's boundary trigger shares them exactly.
+        if !tx_slot_eligible(
+            self.qso.active,
+            self.tx_parity,
+            slot_id,
+            self.txed_slot,
+            self.now().rem_euclid(CYCLE_MS),
+        ) {
             return;
         }
-        self.tx_parity = Some(parity);
+        // Answering a CQ pins the parity to the operator's click slot so we always
+        // reply opposite the DX; a CQ / free-text start locks it on this first TX.
+        self.tx_parity = Some(slot_id.rem_euclid(2));
         self.txed_slot = slot_id;
         if let Some(msg) = self.qso.tx_message().map(|s| s.to_string()) {
             if self.start_transmit(&msg) {
@@ -931,6 +1006,9 @@ impl Engine {
             EngineCommand::StopDecode => {
                 self.decoding = false;
                 self.input = None;
+                // No slot is in flight once capture stops; clear the wait so a
+                // later CQ can key at its slot boundary without being blocked.
+                self.awaiting_decode = false;
                 // Tell companion apps to wipe their decode band and forget the
                 // replay history — this is a genuine reset, not a per-cycle churn.
                 if self.udp.is_enabled() {
@@ -1241,4 +1319,41 @@ mod tests {
         // And again at the following boundary.
         assert!(wf_boundary_row(2 * CYCLE_MS + 10, &mut last));
     }
+
+    #[test]
+    fn tx_slot_eligible_enforces_active_parity_once_and_window() {
+        use super::{tx_slot_eligible, TX_LATEST_MS};
+        // Inactive QSO never keys.
+        assert!(!tx_slot_eligible(false, None, 10, -1, 0));
+        // Active, no parity lock, fresh slot, at the very top of the cycle → eligible.
+        assert!(tx_slot_eligible(true, None, 10, -1, 0));
+        // The last instant that still fits the waveform is eligible; one ms later is not.
+        assert!(tx_slot_eligible(true, None, 10, -1, TX_LATEST_MS));
+        assert!(!tx_slot_eligible(true, None, 10, -1, TX_LATEST_MS + 1));
+        // Already transmitted in this slot → no second keying.
+        assert!(!tx_slot_eligible(true, None, 10, 10, 0));
+        // A locked parity must match the slot's parity (slot 10 is even, 11 is odd).
+        assert!(tx_slot_eligible(true, Some(0), 10, -1, 0));
+        assert!(!tx_slot_eligible(true, Some(1), 10, -1, 0));
+        assert!(tx_slot_eligible(true, Some(1), 11, -1, 0));
+    }
+
+    #[test]
+    fn boundary_tx_keys_at_slot_top_only_after_this_slots_decode() {
+        use super::{boundary_tx_ready, TX_LATEST_MS};
+        // Fast decode: this slot's decodes are already processed (awaiting=false), so a
+        // reply computed in the previous cycle keys at the top of its slot. This is the
+        // case the decode-arrival trigger used to miss — it fires mid-slot, past the
+        // window, so the queued reply/CQ was silently dropped.
+        assert!(boundary_tx_ready(true, false, None, 4, -1, 0));
+        // Slow decode still pending: must NOT key a (stale) message; handle_decoded
+        // keys the fresh one when the decode lands.
+        assert!(!boundary_tx_ready(true, true, None, 4, -1, 0));
+        // Past the window, already transmitted this slot, wrong parity, or inactive all
+        // block it too (it delegates to tx_slot_eligible).
+        assert!(!boundary_tx_ready(true, false, None, 4, -1, TX_LATEST_MS + 1));
+        assert!(!boundary_tx_ready(true, false, None, 4, 4, 0));
+        assert!(!boundary_tx_ready(true, false, Some(1), 4, -1, 0)); // wants odd, slot 4 even
+        assert!(!boundary_tx_ready(false, false, None, 4, -1, 0));
+    }
 }