From 8d3f9222ffc090132da0adfba1d4236d322efc30 Mon Sep 17 00:00:00 2001 From: Jonathan Bennett Date: Wed, 8 Apr 2026 17:52:47 -0500 Subject: [PATCH] Don't firehose missing packets --- bin/dmshell_client.py | 57 ++++++++++++++++++++++++++++------------- src/modules/DMShell.cpp | 54 +++++++++++++++++++++++--------------- src/modules/DMShell.h | 3 ++- 3 files changed, 74 insertions(+), 40 deletions(-) diff --git a/bin/dmshell_client.py b/bin/dmshell_client.py index df082eaf7..a9d718c48 100644 --- a/bin/dmshell_client.py +++ b/bin/dmshell_client.py @@ -202,13 +202,14 @@ class SessionState: next_seq: int = 1 last_rx_seq: int = 0 next_expected_rx_seq: int = 1 + highest_seen_rx_seq: int = 0 active: bool = False stopped: bool = False opened_event: threading.Event = field(default_factory=threading.Event) closed_event: threading.Event = field(default_factory=threading.Event) event_queue: "queue.Queue[str]" = field(default_factory=queue.Queue) tx_lock: threading.Lock = field(default_factory=threading.Lock) - tx_history: deque[SentShellFrame] = field(default_factory=lambda: deque(maxlen=10)) + tx_history: deque[SentShellFrame] = field(default_factory=lambda: deque(maxlen=50)) def alloc_seq(self) -> int: with self.tx_lock: @@ -225,17 +226,32 @@ class SessionState: if seq == 0: return ("process", None) if seq < self.next_expected_rx_seq: + if self.highest_seen_rx_seq >= self.next_expected_rx_seq: + return ("gap", self.next_expected_rx_seq) return ("duplicate", None) if seq > self.next_expected_rx_seq: + if seq > self.highest_seen_rx_seq: + self.highest_seen_rx_seq = seq return ("gap", self.next_expected_rx_seq) self.last_rx_seq = seq self.next_expected_rx_seq = seq + 1 + if seq > self.highest_seen_rx_seq: + self.highest_seen_rx_seq = seq + if self.highest_seen_rx_seq < self.next_expected_rx_seq: + self.highest_seen_rx_seq = 0 return ("process", None) + def pending_missing_seq(self) -> Optional[int]: + with self.tx_lock: + if self.highest_seen_rx_seq >= self.next_expected_rx_seq: + return self.next_expected_rx_seq + return None + def set_receive_cursor(self, seq: int) -> None: with self.tx_lock: self.last_rx_seq = seq self.next_expected_rx_seq = seq + 1 + self.highest_seen_rx_seq = seq def remember_sent_frame(self, frame: SentShellFrame) -> None: if frame.seq == 0 or frame.op == self.pb2.mesh.DMShell.ACK: @@ -247,7 +263,7 @@ class SessionState: if ack_seq <= 0: return with self.tx_lock: - self.tx_history = deque((frame for frame in self.tx_history if frame.seq > ack_seq), maxlen=10) + self.tx_history = deque((frame for frame in self.tx_history if frame.seq > ack_seq), maxlen=50) def replay_frames_from(self, start_seq: int) -> list[SentShellFrame]: with self.tx_lock: @@ -332,24 +348,24 @@ def send_ack_frame(sock: socket.socket, state: SessionState, replay_from: Option def replay_frames_from(sock: socket.socket, state: SessionState, start_seq: int) -> None: - frames = state.replay_frames_from(start_seq) - if not frames: + frame = next((f for f in state.replay_frames_from(start_seq) if f.seq == start_seq), None) + if frame is None: state.event_queue.put(f"replay unavailable from seq={start_seq}") return - for frame in frames: - send_shell_frame( - sock, - state, - frame.op, - payload=frame.payload, - cols=frame.cols, - rows=frame.rows, - session_id=frame.session_id, - ack_seq=frame.ack_seq, - seq=frame.seq, - flags=frame.flags, - remember=False, - ) + state.event_queue.put(f"replay frame seq={start_seq}") + send_shell_frame( + sock, + state, + frame.op, + payload=frame.payload, + cols=frame.cols, + rows=frame.rows, + session_id=frame.session_id, + ack_seq=frame.ack_seq, + seq=frame.seq, + flags=frame.flags, + remember=False, + ) def wait_for_config_complete(sock: socket.socket, pb2, timeout: float, verbose: bool) -> None: @@ -398,6 +414,7 @@ def reader_loop(sock: socket.socket, state: SessionState) -> None: continue state.prune_sent_frames(shell.ack_seq) if shell.op == state.pb2.mesh.DMShell.ACK: + state.event_queue.put("peer requested replay") replay_from = decode_replay_request(shell.payload) if replay_from is not None: state.event_queue.put(f"peer requested replay from seq={replay_from}") @@ -437,6 +454,10 @@ def reader_loop(sock: socket.socket, state: SessionState) -> None: return elif shell.op == state.pb2.mesh.DMShell.PONG: state.event_queue.put("pong") + + missing_seq = state.pending_missing_seq() + if missing_seq is not None: + send_ack_frame(sock, state, replay_from=missing_seq) elif state.verbose and variant: state.event_queue.put(f"fromradio {variant}") diff --git a/src/modules/DMShell.cpp b/src/modules/DMShell.cpp index 36a1b0247..001d001aa 100644 --- a/src/modules/DMShell.cpp +++ b/src/modules/DMShell.cpp @@ -103,7 +103,7 @@ ProcessMessage DMShellModule::handleReceived(const meshtastic_MeshPacket &mp) if (session.active && frame.sessionId == session.sessionId && getFrom(&mp) == session.peer) { handleAckFrame(frame); } - return ProcessMessage::STOP; + return ProcessMessage::CONTINUE; } if (frame.op >= 64) { @@ -319,6 +319,7 @@ bool DMShellModule::openSession(const meshtastic_MeshPacket &mp, const DMShellFr session.nextTxSeq = 1; session.lastAckedRxSeq = frame.seq; session.nextExpectedRxSeq = frame.seq + 1; + session.highestSeenRxSeq = frame.seq; session.lastActivityMs = millis(); uint8_t payload[sizeof(uint32_t)] = {0}; @@ -434,27 +435,23 @@ void DMShellModule::resendFramesFrom(uint32_t startSeq) return; } - uint32_t lastSeq = startSeq - 1; - for (size_t sentCount = 0; sentCount < session.txHistory.size(); ++sentCount) { - DMShellSession::SentFrame *best = nullptr; - for (auto &entry : session.txHistory) { - if (!entry.valid || entry.seq < startSeq || entry.seq <= lastSeq) { - continue; - } - if (!best || entry.seq < best->seq) { - best = &entry; - } + DMShellSession::SentFrame *match = nullptr; + for (auto &entry : session.txHistory) { + if (!entry.valid || entry.seq != startSeq) { + continue; } - - if (!best) { - break; - } - - LOG_INFO("DMShell: replaying frame seq=%u op=%d", best->seq, best->op); - sendFrameToPeer(session.peer, session.channel, best->op, best->sessionId, best->seq, best->payload, best->payloadLen, - best->cols, best->rows, best->ackSeq, best->flags, false); - lastSeq = best->seq; + match = &entry; + break; } + + if (!match) { + LOG_WARN("DMShell: replay request for seq=%u not found in history", startSeq); + return; + } + + LOG_INFO("DMShell: replaying frame seq=%u op=%d", match->seq, match->op); + sendFrameToPeer(session.peer, session.channel, match->op, match->sessionId, match->seq, match->payload, match->payloadLen, + match->cols, match->rows, match->ackSeq, match->flags, false); } void DMShellModule::sendAck(uint32_t replayFromSeq) @@ -495,17 +492,32 @@ bool DMShellModule::shouldProcessIncomingFrame(const DMShellFrame &frame) } if (frame.seq < session.nextExpectedRxSeq) { - sendAck(); + if (session.highestSeenRxSeq >= session.nextExpectedRxSeq) { + sendAck(session.nextExpectedRxSeq); + } else { + sendAck(); + } return false; } if (frame.seq > session.nextExpectedRxSeq) { + if (frame.seq > session.highestSeenRxSeq) { + session.highestSeenRxSeq = frame.seq; + } sendAck(session.nextExpectedRxSeq); return false; } session.lastAckedRxSeq = frame.seq; session.nextExpectedRxSeq = frame.seq + 1; + if (frame.seq > session.highestSeenRxSeq) { + session.highestSeenRxSeq = frame.seq; + } + if (session.highestSeenRxSeq >= session.nextExpectedRxSeq) { + sendAck(session.nextExpectedRxSeq); + } else { + session.highestSeenRxSeq = 0; + } return true; } diff --git a/src/modules/DMShell.h b/src/modules/DMShell.h index d528dfbd3..302c6cac2 100644 --- a/src/modules/DMShell.h +++ b/src/modules/DMShell.h @@ -34,6 +34,7 @@ struct DMShellSession { uint32_t nextTxSeq = 1; uint32_t lastAckedRxSeq = 0; uint32_t nextExpectedRxSeq = 1; + uint32_t highestSeenRxSeq = 0; uint32_t lastActivityMs = 0; struct SentFrame { bool valid = false; @@ -47,7 +48,7 @@ struct DMShellSession { uint8_t payload[meshtastic_Constants_DATA_PAYLOAD_LEN] = {0}; size_t payloadLen = 0; }; - std::array txHistory = {}; + std::array txHistory = {}; size_t txHistoryNext = 0; };