latest test changes

2026-05-31 17:02:08 -04:00
parent f25e3e893d
commit 11a14bdda5
4 changed files with 88 additions and 37 deletions
@@ -24,7 +24,12 @@ START1 = 0x94
 START2 = 0xC3
 HEADER_LEN = 4
 DEFAULT_API_PORT = 4403
-DEFAULT_HOP_LIMIT = 0
+# Must be > 0: the firmware sets hop_start = hop_limit when it originates the packet, and the
+# receiver's pre-hop drop policy (classifyHopStart, which runs *before* decryption so it can't see
+# the bitfield) discards packets that arrive with hop_start == 0. hop_limit=0 therefore gets us
+# silently dropped at the far end. On a direct 2-node link any non-zero value works; the
+# destination receives it directly and does not rebroadcast a DM addressed to itself.
+DEFAULT_HOP_LIMIT = 3
 LOCAL_ESCAPE_BYTE = b"\x1d"  # Ctrl+]
 MISSING_SEQ_RETRY_INTERVAL_SEC = 1.0
 INPUT_BATCH_WINDOW_SEC = .5
@@ -41,7 +46,7 @@ HEARTBEAT_POLL_INTERVAL_SEC = 0.25
 FLAG_GRANT = 0x01  # I am handing you the turn; you may transmit now
 FLAG_MORE = 0x02   # I yielded under a budget but still have data queued (grant back promptly)
 FLAG_RTS = 0x04    # (server->client) I have output but no turn; please grant me one
-CLIENT_TURN_BUDGET = 4   # max frames the client sends per turn before yielding
+CLIENT_TURN_BUDGET = 8   # max frames the client sends per turn before yielding
 TURN_RECLAIM_SEC = 8.0   # reclaim the token if a grant goes unanswered this long (anti-deadlock)


@@ -67,6 +72,12 @@ def parse_args() -> argparse.Namespace:
    parser.add_argument("--baud", type=int, default=115200, help="serial baud rate when using --serial")
    parser.add_argument("--to", required=True, help="destination node number, e.g. !170896f7 or 0x170896f7")
    parser.add_argument("--channel", type=int, default=0, help="channel index to use")
+    parser.add_argument(
+        "--hop-limit",
+        type=int,
+        default=DEFAULT_HOP_LIMIT,
+        help="hop limit (must be > 0 or the far end drops our packets via the pre-hop policy)",
+    )
    parser.add_argument("--cols", type=int, default=None, help="initial terminal columns (default: detect local terminal)")
    parser.add_argument("--rows", type=int, default=None, help="initial terminal rows (default: detect local terminal)")
    parser.add_argument("--command", action="append", default=[], help="send a command line after opening")
@@ -329,6 +340,7 @@ class SessionState:
    target: int
    channel: int
    verbose: bool
+    hop_limit: int = DEFAULT_HOP_LIMIT
    session_id: int = field(default_factory=lambda: random.randint(1, 0x7FFFFFFF))
    next_seq: int = 1
    last_rx_seq: int = 0
@@ -376,6 +388,12 @@ class SessionState:
            self.last_inbound_time = time.monotonic()
            if flags & FLAG_GRANT:
                self.has_token = True
+            if flags & FLAG_RTS:
+                # The server only sends RTS when it has output and does NOT hold the token, so the
+                # token is (or should be) ours. Take it so we can grant. This also self-heals the
+                # case where our previous grant or the server's yield-grant was lost, which would
+                # otherwise leave neither side holding the token and the server flooding RTS forever.
+                self.has_token = True
            if flags & (FLAG_MORE | FLAG_RTS):
                self.grant_requested = True
            self.turn_cv.notify_all()
@@ -555,13 +573,19 @@ def make_toradio_packet(pb2, state: SessionState, shell_msg) -> object:
    # The 'from' field is a reserved keyword in Python, so use setattr
    setattr(packet, "from", 0)
    packet.channel = state.channel
-    packet.hop_limit = DEFAULT_HOP_LIMIT
+    packet.hop_limit = state.hop_limit
    packet.want_ack = False
    packet.decoded.portnum = pb2.portnums.REMOTE_SHELL_APP
    packet.decoded.payload = shell_msg.SerializeToString()
    packet.decoded.want_response = False
    packet.decoded.dest = state.target
    packet.decoded.source = 0
+    # Mark has_bitfield (proto3 optional → assigning even 0 sets presence) and leave OK_TO_MQTT clear,
+    # which is right for a private shell. Note: this is NOT what prevents the far-end pre-hop drop —
+    # that check runs before decryption and can't read the (encrypted) bitfield; a non-zero hop_limit
+    # is what keeps us alive there. The firmware also sets this for locally-originated packets, so this
+    # is mostly belt-and-suspenders for the post-decryption consumers (e.g. getHopsAway).
+    packet.decoded.bitfield = 0

    toradio = pb2.mesh.ToRadio()
    toradio.packet.CopyFrom(packet)
@@ -1092,6 +1116,7 @@ def main() -> int:
        target=parse_node_num(args.to),
        channel=args.channel,
        verbose=args.verbose,
+        hop_limit=args.hop_limit,
        turn_taking=args.turn_taking,
        turn_budget=args.turn_budget,
        turn_reclaim_sec=args.turn_reclaim,
@@ -132,7 +132,7 @@ inline bool shouldDropPacketForPreHop(const meshtastic_MeshPacket &p)
    if (isFromUs(&p)) {
        return false; // local-originated packets should never be dropped by pre-hop drop policy
    }
-    return classifyHopStart(p) != HopStartStatus::VALID;
+    return classifyHopStart(p) == HopStartStatus::INVALID;
 #endif
 }

@@ -38,8 +38,14 @@ constexpr size_t MAX_MESSAGE_SIZE = 200;
 constexpr uint32_t TURN_FLAG_GRANT = 0x01; // I am handing you the turn; you may transmit now
 constexpr uint32_t TURN_FLAG_MORE = 0x02;  // I yielded under a budget but still have data queued
 constexpr uint32_t TURN_FLAG_RTS = 0x04;   // I have output but no turn; please grant me one
-constexpr size_t TURN_BUDGET_FRAMES = 4;   // max output frames per granted turn before yielding
-constexpr uint32_t RTS_RETRY_MS = 1000;    // min interval between request-to-send frames
+constexpr size_t TURN_BUDGET_FRAMES = 8;   // max output frames per granted turn before yielding (bounds interrupt latency)
+constexpr uint32_t RTS_RETRY_MS = 250;     // min interval between request-to-send frames
+// After being granted the turn we keep it for a short "linger" window, continuing to drain shell
+// output as it appears, instead of yielding the instant the PTY drains. This lets a command's
+// output (and the next prompt) ride the same turn as the keystroke that triggered it, avoiding a
+// full RTS->grant round-trip per command. The turn still ends promptly once the PTY is idle this
+// long, or once TURN_BUDGET_FRAMES is hit (so the client can interject, e.g. Ctrl-C).
+constexpr uint32_t TURN_LINGER_MS = 120;
 } // namespace

 DMShellModule::DMShellModule()
@@ -63,6 +69,7 @@ ProcessMessage DMShellModule::handleReceived(const meshtastic_MeshPacket &mp)

    if (frame.op == meshtastic_RemoteShell_OpCode_ACK) {
        if (session.active && frame.session_id == session.sessionId && getFrom(&mp) == session.peer) {
+            LOG_WARN("DMShell: Received ack from 0x%x 0x%x", getFrom(&mp), session.peer);
            applyTurnFlags(frame);
            if (frame.last_rx_seq > 0) {
                resendFramesFrom(frame.last_rx_seq + 1);
@@ -363,7 +370,7 @@ bool DMShellModule::openSession(const meshtastic_MeshPacket &mp, const meshtasti
    session.nextExpectedRxSeq = frame.seq + 1;
    session.highestSeenRxSeq = frame.seq;
    session.lastActivityMs = millis();
-    session.turnManaged = false;
+    session.turnManaged = true;
    session.hasToken = false;
    session.lastRtsMs = 0;

@@ -637,6 +644,11 @@ void DMShellModule::applyTurnFlags(const meshtastic_RemoteShell &frame)
        session.turnManaged = true; // peer speaks turn-taking; enable gating for this session
    }
    if (frame.flags & TURN_FLAG_GRANT) {
+        if (!session.hasToken) {
+            // Fresh turn: start the linger window and reset the per-turn budget.
+            session.turnDeadlineMs = millis() + TURN_LINGER_MS;
+            session.turnFramesSent = 0;
+        }
        session.hasToken = true;
    }
 }
@@ -701,26 +713,29 @@ void DMShellModule::sendRts()
    sendFrameToPeer(session.peer, frame, false);
 }

-// Called when we hold the turn: send up to TURN_BUDGET_FRAMES chunks of shell output, then hand
-// the turn back to the client. The grant is piggybacked on the last output frame (or sent as a
-// standalone ACK if there was nothing to send), so the token always settles back at the client.
+// Called (every tick) while we hold the turn. Drains available shell output and sends it
+// immediately, then decides whether to keep the turn (linger, to catch output that is about to
+// appear) or hand it back. The turn is yielded once the per-turn budget is hit (so the client can
+// interject, e.g. Ctrl-C) or once the PTY has been idle past the linger window. Output frames go
+// out as soon as they are read (no extra delay); the grant is a trailing ACK.
 void DMShellModule::serviceTurn()
 {
    if (!session.active || !session.turnManaged || !session.hasToken) {
        return;
    }

-    uint8_t chunks[TURN_BUDGET_FRAMES][MAX_MESSAGE_SIZE];
-    size_t chunkLen[TURN_BUDGET_FRAMES] = {0};
-    size_t nChunks = 0;
+    uint8_t buf[MAX_MESSAGE_SIZE];
    bool eof = false;
    bool readError = false;

-    while (nChunks < TURN_BUDGET_FRAMES && session.masterFd >= 0) {
-        const ssize_t n = read(session.masterFd, chunks[nChunks], MAX_MESSAGE_SIZE);
+    // Drain whatever is available right now, up to the remaining per-turn budget.
+    while (session.turnFramesSent < TURN_BUDGET_FRAMES && session.masterFd >= 0) {
+        const ssize_t n = read(session.masterFd, buf, sizeof(buf));
        if (n > 0) {
-            chunkLen[nChunks] = (size_t)n;
-            nChunks++;
+            sendOutputFrame(buf, (size_t)n, 0u);
+            session.turnFramesSent++;
+            session.lastActivityMs = millis();
+            session.turnDeadlineMs = millis() + TURN_LINGER_MS; // extend the linger while output flows
        } else if (n == 0) {
            eof = true;
            break;
@@ -732,27 +747,36 @@ void DMShellModule::serviceTurn()
        }
    }

-    // If we filled the budget, there may still be more output waiting; ask for a prompt re-grant.
-    const bool more = (nChunks == TURN_BUDGET_FRAMES) && ptyHasOutput();
-
-    session.hasToken = false; // we are handing the turn back below
-    if (nChunks > 0) {
-        session.lastActivityMs = millis();
-        for (size_t i = 0; i < nChunks; i++) {
-            const uint32_t flags = (i + 1 == nChunks) ? (TURN_FLAG_GRANT | (more ? TURN_FLAG_MORE : 0u)) : 0u;
-            sendOutputFrame(chunks[i], chunkLen[i], flags);
-        }
-    } else {
-        // Nothing to send: hand the turn straight back so the link goes quiet at the client.
+    if (eof || readError) {
+        session.hasToken = false;
        sendTurnGrant(false);
+        if (eof) {
+            closeSession("pty_eof", true);
+        } else {
+            LOG_WARN("DMShell: PTY read error errno=%d", errno);
+            closeSession("pty_read_error", true);
+        }
+        return;
    }

-    if (eof) {
-        closeSession("pty_eof", true);
-    } else if (readError) {
-        LOG_WARN("DMShell: PTY read error errno=%d", errno);
-        closeSession("pty_read_error", true);
+    const bool morePending = ptyHasOutput();
+
+    if (session.turnFramesSent >= TURN_BUDGET_FRAMES) {
+        // Hit the per-turn budget: yield so the client gets a chance to interject (e.g. Ctrl-C).
+        session.hasToken = false;
+        sendTurnGrant(morePending);
+        return;
    }
+
+    if (!morePending && (int32_t)(millis() - session.turnDeadlineMs) >= 0) {
+        // PTY has been idle past the linger window: hand the turn back.
+        session.hasToken = false;
+        sendTurnGrant(false);
+        return;
+    }
+
+    // Otherwise keep holding the turn: there is more to drain next pass, or we are lingering for
+    // output that may be about to appear. runOnce re-enters serviceTurn on the next tick.
 }

 void DMShellModule::sendError(const char *message, NodeNum peer)
@@ -25,9 +25,11 @@ struct DMShellSession {
    uint32_t highestSeenRxSeq = 0;
    uint32_t lastActivityMs = 0;
    // --- Half-duplex turn-taking ("talking stick") state ---
-    bool turnManaged = false; // becomes true once the peer uses turn-taking flags; enables gating
-    bool hasToken = false;    // do we currently hold the right to transmit?
-    uint32_t lastRtsMs = 0;   // last time we asked the client for a turn (rate-limit)
+    bool turnManaged = false;    // becomes true once the peer uses turn-taking flags; enables gating
+    bool hasToken = false;       // do we currently hold the right to transmit?
+    uint32_t lastRtsMs = 0;      // last time we asked the client for a turn (rate-limit)
+    uint32_t turnDeadlineMs = 0; // while holding the turn, keep it until this time (extended on output)
+    uint32_t turnFramesSent = 0; // output frames sent in the current held turn (vs TURN_BUDGET_FRAMES)
    struct SentFrame {
        bool valid = false;
        meshtastic_RemoteShell_OpCode op = meshtastic_RemoteShell_OpCode_ERROR;