From 4a3f44955571d439a51b6621b7561434c0742919 Mon Sep 17 00:00:00 2001 From: Jonathan Bennett Date: Wed, 8 Apr 2026 15:16:33 -0500 Subject: [PATCH] Try to re-request missed sequences --- bin/dmshell_client.py | 149 ++++++++++++++++++++++++++++++-- src/modules/DMShell.cpp | 182 ++++++++++++++++++++++++++++++++++++---- src/modules/DMShell.h | 31 ++++++- 3 files changed, 339 insertions(+), 23 deletions(-) diff --git a/bin/dmshell_client.py b/bin/dmshell_client.py index a5b8705a7..6f76d79a4 100644 --- a/bin/dmshell_client.py +++ b/bin/dmshell_client.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import argparse +import struct import os import queue import random @@ -14,6 +15,7 @@ import termios import threading import time import tty +from collections import deque from dataclasses import dataclass, field from pathlib import Path from typing import Optional @@ -25,6 +27,7 @@ HEADER_LEN = 4 DEFAULT_API_PORT = 4403 DEFAULT_HOP_LIMIT = 3 LOCAL_ESCAPE_BYTE = b"\x1d" # Ctrl+] +REPLAY_REQUEST_SIZE = 4 def parse_args() -> argparse.Namespace: @@ -177,6 +180,18 @@ def send_stream_frame(sock: socket.socket, payload: bytes) -> None: sock.sendall(header + payload) +@dataclass +class SentShellFrame: + op: int + session_id: int + seq: int + ack_seq: int + payload: bytes = b"" + cols: int = 0 + rows: int = 0 + flags: int = 0 + + @dataclass class SessionState: pb2: object # ProtoModules with mesh and portnums attributes @@ -185,16 +200,68 @@ class SessionState: verbose: bool session_id: int = field(default_factory=lambda: random.randint(1, 0x7FFFFFFF)) next_seq: int = 1 + last_rx_seq: int = 0 + next_expected_rx_seq: int = 1 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)) def alloc_seq(self) -> int: - value = self.next_seq - self.next_seq += 1 - return value + with self.tx_lock: + value = self.next_seq + self.next_seq += 1 + return value + + def current_ack_seq(self) -> int: + with self.tx_lock: + return self.last_rx_seq + + def note_received_seq(self, seq: int) -> tuple[str, Optional[int]]: + with self.tx_lock: + if seq == 0: + return ("process", None) + if seq < self.next_expected_rx_seq: + return ("duplicate", None) + if seq > self.next_expected_rx_seq: + return ("gap", self.next_expected_rx_seq) + self.last_rx_seq = seq + self.next_expected_rx_seq = seq + 1 + return ("process", 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 + + def remember_sent_frame(self, frame: SentShellFrame) -> None: + if frame.seq == 0 or frame.op == self.pb2.mesh.DMShell.ACK: + return + with self.tx_lock: + self.tx_history.append(frame) + + def prune_sent_frames(self, ack_seq: int) -> None: + 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) + + def replay_frames_from(self, start_seq: int) -> list[SentShellFrame]: + with self.tx_lock: + return [frame for frame in self.tx_history if frame.seq >= start_seq] + + +def encode_replay_request(start_seq: int) -> bytes: + return struct.pack(">I", start_seq) + + +def decode_replay_request(payload: bytes) -> Optional[int]: + if len(payload) < REPLAY_REQUEST_SIZE: + return None + return struct.unpack(">I", payload[:REPLAY_REQUEST_SIZE])[0] def send_toradio(sock: socket.socket, toradio) -> None: @@ -221,16 +288,68 @@ def make_toradio_packet(pb2, state: SessionState, shell_msg) -> object: return toradio -def send_shell_frame(sock: socket.socket, state: SessionState, op: int, payload: bytes = b"", cols: int = 0, rows: int = 0) -> None: +def send_shell_frame( + sock: socket.socket, + state: SessionState, + op: int, + payload: bytes = b"", + cols: int = 0, + rows: int = 0, + session_id: Optional[int] = None, + ack_seq: Optional[int] = None, + seq: Optional[int] = None, + flags: int = 0, + remember: bool = True, +) -> int: + if seq is None: + seq = 0 if op == state.pb2.mesh.DMShell.ACK else state.alloc_seq() + if ack_seq is None: + ack_seq = state.current_ack_seq() + if session_id is None: + session_id = state.session_id + shell = state.pb2.mesh.DMShell() shell.op = op - shell.session_id = state.session_id - shell.seq = state.alloc_seq() + shell.session_id = session_id + shell.seq = seq + shell.ack_seq = ack_seq shell.cols = cols shell.rows = rows + shell.flags = flags if payload: shell.payload = payload send_toradio(sock, make_toradio_packet(state.pb2, state, shell)) + if remember: + state.remember_sent_frame( + SentShellFrame(op=op, session_id=session_id, seq=seq, ack_seq=ack_seq, payload=payload, cols=cols, rows=rows, flags=flags) + ) + return seq + + +def send_ack_frame(sock: socket.socket, state: SessionState, replay_from: Optional[int] = None) -> None: + payload = b"" if replay_from is None else encode_replay_request(replay_from) + send_shell_frame(sock, state, state.pb2.mesh.DMShell.ACK, payload=payload, seq=0, remember=False) + + +def replay_frames_from(sock: socket.socket, state: SessionState, start_seq: int) -> None: + frames = state.replay_frames_from(start_seq) + if not frames: + 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, + ) def wait_for_config_complete(sock: socket.socket, pb2, timeout: float, verbose: bool) -> None: @@ -277,8 +396,26 @@ def reader_loop(sock: socket.socket, state: SessionState) -> None: shell = decode_shell_packet(state, fromradio.packet) if not shell: continue + state.prune_sent_frames(shell.ack_seq) + if shell.op == state.pb2.mesh.DMShell.ACK: + 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}") + replay_frames_from(sock, state, replay_from) + continue + + action, missing_from = state.note_received_seq(shell.seq) + if action == "duplicate": + send_ack_frame(sock, state) + continue + if action == "gap": + state.event_queue.put(f"missing frame before seq={shell.seq}, requesting replay from seq={missing_from}") + send_ack_frame(sock, state, replay_from=missing_from) + continue + if shell.op == state.pb2.mesh.DMShell.OPEN_OK: state.session_id = shell.session_id + state.set_receive_cursor(shell.seq) state.active = True state.opened_event.set() pid = int.from_bytes(shell.payload, "big") if shell.payload else 0 diff --git a/src/modules/DMShell.cpp b/src/modules/DMShell.cpp index bf7d8679b..fed1a526e 100644 --- a/src/modules/DMShell.cpp +++ b/src/modules/DMShell.cpp @@ -29,6 +29,7 @@ namespace constexpr uint16_t PTY_COLS_DEFAULT = 120; constexpr uint16_t PTY_ROWS_DEFAULT = 40; constexpr size_t MAX_PTY_READ_SIZE = 200; +constexpr size_t REPLAY_REQUEST_SIZE = sizeof(uint32_t); struct BytesDecodeState { uint8_t *buf; @@ -67,6 +68,20 @@ bool encodeBytesField(pb_ostream_t *stream, const pb_field_iter_t *field, void * } return pb_encode_string(stream, state->buf, state->len); } + +void encodeUint32BE(uint8_t *dest, uint32_t value) +{ + dest[0] = static_cast((value >> 24) & 0xff); + dest[1] = static_cast((value >> 16) & 0xff); + dest[2] = static_cast((value >> 8) & 0xff); + dest[3] = static_cast(value & 0xff); +} + +uint32_t decodeUint32BE(const uint8_t *src) +{ + return (static_cast(src[0]) << 24) | (static_cast(src[1]) << 16) | (static_cast(src[2]) << 8) | + static_cast(src[3]); +} } // namespace DMShellModule::DMShellModule() @@ -94,20 +109,35 @@ ProcessMessage DMShellModule::handleReceived(const meshtastic_MeshPacket &mp) return ProcessMessage::STOP; } + if (frame.ackSeq > 0) { + pruneSentFrames(frame.ackSeq); + } + + if (frame.op == meshtastic_DMShell_OpCode_ACK) { + if (session.active && frame.sessionId == session.sessionId && getFrom(&mp) == session.peer) { + handleAckFrame(frame); + } + return ProcessMessage::STOP; + } + if (frame.op == meshtastic_DMShell_OpCode_OPEN) { LOG_WARN("DMShell: received OPEN from 0x%x sessionId=0x%x", mp.from, frame.sessionId); if (!openSession(mp, frame)) { const char *msg = "open_failed"; - sendFrameToPeer(getFrom(&mp), mp.channel, meshtastic_DMShell_OpCode_ERROR, frame.sessionId, frame.seq, - reinterpret_cast(msg), strlen(msg)); + sendFrameToPeer(getFrom(&mp), mp.channel, meshtastic_DMShell_OpCode_ERROR, frame.sessionId, 0, + reinterpret_cast(msg), strlen(msg), 0, 0, frame.seq); } return ProcessMessage::STOP; } if (!session.active || frame.sessionId != session.sessionId || getFrom(&mp) != session.peer) { const char *msg = "invalid_session"; - sendFrameToPeer(getFrom(&mp), mp.channel, meshtastic_DMShell_OpCode_ERROR, frame.sessionId, frame.seq, - reinterpret_cast(msg), strlen(msg)); + sendFrameToPeer(getFrom(&mp), mp.channel, meshtastic_DMShell_OpCode_ERROR, frame.sessionId, 0, + reinterpret_cast(msg), strlen(msg), 0, 0, frame.seq); + return ProcessMessage::STOP; + } + + if (!shouldProcessIncomingFrame(frame)) { return ProcessMessage::STOP; } @@ -130,7 +160,7 @@ ProcessMessage DMShellModule::handleReceived(const meshtastic_MeshPacket &mp) } break; case meshtastic_DMShell_OpCode_PING: - sendControl(meshtastic_DMShell_OpCode_PONG, nullptr, 0, frame.seq); + sendControl(meshtastic_DMShell_OpCode_PONG, nullptr, 0); break; case meshtastic_DMShell_OpCode_CLOSE: closeSession("peer_close", true); @@ -164,7 +194,7 @@ int32_t DMShellModule::runOnce() const ssize_t bytesRead = read(session.masterFd, outBuf, sizeof(outBuf)); if (bytesRead > 0) { LOG_WARN("DMShell: read %d bytes from PTY", bytesRead); - sendControl(meshtastic_DMShell_OpCode_OUTPUT, outBuf, static_cast(bytesRead), session.txSeq++); + sendControl(meshtastic_DMShell_OpCode_OUTPUT, outBuf, static_cast(bytesRead)); session.lastActivityMs = millis(); continue; } @@ -280,7 +310,9 @@ bool DMShellModule::openSession(const meshtastic_MeshPacket &mp, const DMShellFr session.channel = mp.channel; session.masterFd = masterFd; session.childPid = childPid; - session.txSeq = 0; + session.nextTxSeq = 1; + session.lastAckedRxSeq = frame.seq; + session.nextExpectedRxSeq = frame.seq + 1; session.lastActivityMs = millis(); uint8_t payload[sizeof(uint32_t)] = {0}; @@ -288,8 +320,8 @@ bool DMShellModule::openSession(const meshtastic_MeshPacket &mp, const DMShellFr ((static_cast(session.childPid) << 8) & 0xff0000) | (static_cast(session.childPid) >> 24); memcpy(payload, &pidBE, sizeof(payload)); - sendFrameToPeer(session.peer, session.channel, meshtastic_DMShell_OpCode_OPEN_OK, session.sessionId, frame.seq, payload, - sizeof(payload), ws.ws_col, ws.ws_row); + sendFrameToPeer(session.peer, session.channel, meshtastic_DMShell_OpCode_OPEN_OK, session.sessionId, allocTxSeq(), payload, + sizeof(payload), ws.ws_col, ws.ws_row, frame.seq); LOG_INFO("DMShell: opened session=0x%x peer=0x%x pid=%d", session.sessionId, session.peer, session.childPid); return true; @@ -316,7 +348,7 @@ void DMShellModule::closeSession(const char *reason, bool notifyPeer) if (notifyPeer) { const size_t reasonLen = strnlen(reason, 256); - sendControl(meshtastic_DMShell_OpCode_CLOSED, reinterpret_cast(reason), reasonLen, session.txSeq++); + sendControl(meshtastic_DMShell_OpCode_CLOSED, reinterpret_cast(reason), reasonLen); } if (session.masterFd >= 0) { @@ -348,20 +380,142 @@ void DMShellModule::reapChildIfExited() } } -void DMShellModule::sendControl(meshtastic_DMShell_OpCode op, const uint8_t *payload, size_t payloadLen, uint32_t seq) +uint32_t DMShellModule::allocTxSeq() { - sendFrameToPeer(session.peer, session.channel, op, session.sessionId, seq, payload, payloadLen); + return session.nextTxSeq++; +} + +void DMShellModule::rememberSentFrame(meshtastic_DMShell_OpCode op, uint32_t sessionId, uint32_t seq, const uint8_t *payload, + size_t payloadLen, uint32_t cols, uint32_t rows, uint32_t ackSeq, uint32_t flags) +{ + if (seq == 0 || op == meshtastic_DMShell_OpCode_ACK) { + return; + } + + auto &entry = session.txHistory[session.txHistoryNext]; + entry.valid = true; + entry.op = op; + entry.sessionId = sessionId; + entry.seq = seq; + entry.ackSeq = ackSeq; + entry.cols = cols; + entry.rows = rows; + entry.flags = flags; + entry.payloadLen = payloadLen; + if (payload && payloadLen > 0) { + memcpy(entry.payload, payload, payloadLen); + } + + session.txHistoryNext = (session.txHistoryNext + 1) % session.txHistory.size(); +} + +void DMShellModule::pruneSentFrames(uint32_t ackSeq) +{ + if (ackSeq == 0) { + return; + } + + for (auto &entry : session.txHistory) { + if (entry.valid && entry.seq <= ackSeq) { + entry.valid = false; + } + } +} + +void DMShellModule::resendFramesFrom(uint32_t startSeq) +{ + if (startSeq == 0) { + 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; + } + } + + 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; + } +} + +void DMShellModule::sendAck(uint32_t replayFromSeq) +{ + uint8_t payload[REPLAY_REQUEST_SIZE] = {0}; + const uint8_t *payloadPtr = nullptr; + size_t payloadLen = 0; + if (replayFromSeq > 0) { + encodeUint32BE(payload, replayFromSeq); + payloadPtr = payload; + payloadLen = sizeof(payload); + LOG_WARN("DMShell: requesting replay from seq=%u", replayFromSeq); + } + + sendFrameToPeer(session.peer, session.channel, meshtastic_DMShell_OpCode_ACK, session.sessionId, 0, payloadPtr, payloadLen, 0, + 0, session.lastAckedRxSeq, 0, false); +} + +void DMShellModule::sendControl(meshtastic_DMShell_OpCode op, const uint8_t *payload, size_t payloadLen) +{ + sendFrameToPeer(session.peer, session.channel, op, session.sessionId, allocTxSeq(), payload, payloadLen, 0, 0, + session.lastAckedRxSeq); +} + +bool DMShellModule::handleAckFrame(const DMShellFrame &frame) +{ + if (frame.payloadLen >= REPLAY_REQUEST_SIZE) { + const uint32_t replayFromSeq = decodeUint32BE(frame.payload); + resendFramesFrom(replayFromSeq); + } + return true; +} + +bool DMShellModule::shouldProcessIncomingFrame(const DMShellFrame &frame) +{ + if (frame.seq == 0) { + return true; + } + + if (frame.seq < session.nextExpectedRxSeq) { + sendAck(); + return false; + } + + if (frame.seq > session.nextExpectedRxSeq) { + sendAck(session.nextExpectedRxSeq); + return false; + } + + session.lastAckedRxSeq = frame.seq; + session.nextExpectedRxSeq = frame.seq + 1; + return true; } void DMShellModule::sendFrameToPeer(NodeNum peer, uint8_t channel, meshtastic_DMShell_OpCode op, uint32_t sessionId, uint32_t seq, const uint8_t *payload, size_t payloadLen, uint32_t cols, uint32_t rows, uint32_t ackSeq, - uint32_t flags) + uint32_t flags, bool remember) { meshtastic_MeshPacket *packet = buildFramePacket(op, sessionId, seq, payload, payloadLen, cols, rows, ackSeq, flags); if (!packet) { return; } + if (remember) { + rememberSentFrame(op, sessionId, seq, payload, payloadLen, cols, rows, ackSeq, flags); + } + packet->to = peer; packet->channel = channel; packet->want_ack = false; @@ -372,7 +526,7 @@ void DMShellModule::sendFrameToPeer(NodeNum peer, uint8_t channel, meshtastic_DM void DMShellModule::sendError(const char *message) { const size_t len = strnlen(message, meshtastic_Constants_DATA_PAYLOAD_LEN); - sendControl(meshtastic_DMShell_OpCode_ERROR, reinterpret_cast(message), len, session.txSeq++); + sendControl(meshtastic_DMShell_OpCode_ERROR, reinterpret_cast(message), len); } meshtastic_MeshPacket *DMShellModule::buildFramePacket(meshtastic_DMShell_OpCode op, uint32_t sessionId, uint32_t seq, diff --git a/src/modules/DMShell.h b/src/modules/DMShell.h index dde56a3c7..d528dfbd3 100644 --- a/src/modules/DMShell.h +++ b/src/modules/DMShell.h @@ -7,6 +7,7 @@ #include "configuration.h" #include "mesh/generated/meshtastic/mesh.pb.h" #include +#include #include #if defined(ARCH_PORTDUINO) @@ -30,8 +31,24 @@ struct DMShellSession { uint8_t channel = 0; int masterFd = -1; int childPid = -1; - uint32_t txSeq = 0; + uint32_t nextTxSeq = 1; + uint32_t lastAckedRxSeq = 0; + uint32_t nextExpectedRxSeq = 1; uint32_t lastActivityMs = 0; + struct SentFrame { + bool valid = false; + meshtastic_DMShell_OpCode op = meshtastic_DMShell_OpCode_ERROR; + uint32_t sessionId = 0; + uint32_t seq = 0; + uint32_t ackSeq = 0; + uint32_t cols = 0; + uint32_t rows = 0; + uint32_t flags = 0; + uint8_t payload[meshtastic_Constants_DATA_PAYLOAD_LEN] = {0}; + size_t payloadLen = 0; + }; + std::array txHistory = {}; + size_t txHistoryNext = 0; }; class DMShellModule : private concurrency::OSThread, public SinglePortModule @@ -52,14 +69,22 @@ class DMShellModule : private concurrency::OSThread, public SinglePortModule bool parseFrame(const meshtastic_MeshPacket &mp, DMShellFrame &outFrame); bool isAuthorizedPacket(const meshtastic_MeshPacket &mp) const; bool openSession(const meshtastic_MeshPacket &mp, const DMShellFrame &frame); + bool handleAckFrame(const DMShellFrame &frame); + bool shouldProcessIncomingFrame(const DMShellFrame &frame); bool writeSessionInput(const DMShellFrame &frame); void closeSession(const char *reason, bool notifyPeer); void reapChildIfExited(); - void sendControl(meshtastic_DMShell_OpCode op, const uint8_t *payload, size_t payloadLen, uint32_t seq = 0); + uint32_t allocTxSeq(); + void rememberSentFrame(meshtastic_DMShell_OpCode op, uint32_t sessionId, uint32_t seq, const uint8_t *payload, + size_t payloadLen, uint32_t cols, uint32_t rows, uint32_t ackSeq, uint32_t flags); + void pruneSentFrames(uint32_t ackSeq); + void resendFramesFrom(uint32_t startSeq); + void sendAck(uint32_t replayFromSeq = 0); + void sendControl(meshtastic_DMShell_OpCode op, const uint8_t *payload, size_t payloadLen); void sendFrameToPeer(NodeNum peer, uint8_t channel, meshtastic_DMShell_OpCode op, uint32_t sessionId, uint32_t seq, const uint8_t *payload, size_t payloadLen, uint32_t cols = 0, uint32_t rows = 0, uint32_t ackSeq = 0, - uint32_t flags = 0); + uint32_t flags = 0, bool remember = true); void sendError(const char *message); meshtastic_MeshPacket *buildFramePacket(meshtastic_DMShell_OpCode op, uint32_t sessionId, uint32_t seq, const uint8_t *payload, size_t payloadLen, uint32_t cols, uint32_t rows,