Try to re-request missed sequences

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

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<uint8_t>((value >> 24) & 0xff);
+    dest[1] = static_cast<uint8_t>((value >> 16) & 0xff);
+    dest[2] = static_cast<uint8_t>((value >> 8) & 0xff);
+    dest[3] = static_cast<uint8_t>(value & 0xff);
+}
+
+uint32_t decodeUint32BE(const uint8_t *src)
+{
+    return (static_cast<uint32_t>(src[0]) << 24) | (static_cast<uint32_t>(src[1]) << 16) | (static_cast<uint32_t>(src[2]) << 8) |
+           static_cast<uint32_t>(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<const uint8_t *>(msg), strlen(msg));
+            sendFrameToPeer(getFrom(&mp), mp.channel, meshtastic_DMShell_OpCode_ERROR, frame.sessionId, 0,
+                            reinterpret_cast<const uint8_t *>(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<const uint8_t *>(msg), strlen(msg));
+        sendFrameToPeer(getFrom(&mp), mp.channel, meshtastic_DMShell_OpCode_ERROR, frame.sessionId, 0,
+                        reinterpret_cast<const uint8_t *>(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<size_t>(bytesRead), session.txSeq++);
+            sendControl(meshtastic_DMShell_OpCode_OUTPUT, outBuf, static_cast<size_t>(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<uint32_t>(session.childPid) << 8) & 0xff0000) |
                      (static_cast<uint32_t>(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<const uint8_t *>(reason), reasonLen, session.txSeq++);
+        sendControl(meshtastic_DMShell_OpCode_CLOSED, reinterpret_cast<const uint8_t *>(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<const uint8_t *>(message), len, session.txSeq++);
+    sendControl(meshtastic_DMShell_OpCode_ERROR, reinterpret_cast<const uint8_t *>(message), len);
 }
 
 meshtastic_MeshPacket *DMShellModule::buildFramePacket(meshtastic_DMShell_OpCode op, uint32_t sessionId, uint32_t seq,

src/modules/DMShell.h

@@ -7,6 +7,7 @@
 #include "configuration.h"
 #include "mesh/generated/meshtastic/mesh.pb.h"
 #include <Arduino.h>
+#include <array>
 #include <functional>
 
 #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<SentFrame, 10> 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,