sl0p: stop recycling client ids, which silently ate CLI output

`zellij action` failed ~20% of the time against a live session: exit 0, empty
stdout, empty stderr, and nothing in the server log. `sl0ppi ls` inherited it,
and attached clients were dropped with "lost connection to the zellij server".

The chain, which took measuring rather than reading to find:

  1. every action first probes liveness (sessions::assert_socket): connect,
     send ConnStatus, disconnect. The probe is a full client and takes an id.
  2. new_client() handed out the LOWEST FREE id, so the action connection that
     follows microseconds later gets the id the probe just released.
  3. the main server thread then processes the queued ConnStatus(id), whose
     handler replies and calls remove_client!(id) -- removing the *action*
     client, which now owns that id.
  4. the action's reply goes through send_to_client, which returns Ok(()) for an
     unknown client. The payload is dropped in silence; the client sees only the
     disconnect and exits 0 with nothing to print.

Fix: hand out ids from an advancing cursor instead of reusing the lowest free
one, so a stale removal cannot land on a fresh connection. ClientId is a u16 and
every CLI invocation burns two, so the cursor wraps and skips ids that are still
in use rather than assuming they are free.

Measured on the 2-vCPU Debian box that surfaced this:

                        before        after
  list-tabs             ~20-25%       0/50 failed
  list-panes            ~20-25%       0/50 failed
  list-tabs (attached)  ~25%          0/30 failed
  sl0ppi ls             most runs     0/15 failed
  5 attach cycles       dropped       5/5 stable

Three regression tests: ids are not reused immediately, they advance and skip
live ones, and they wrap without colliding or hanging. zellij-server: 1502 pass.

The silent-drop in send_to_client is left as-is but is worth revisiting: an
unknown client id is a bug in the caller, and swallowing it is what made this
cost a day instead of a minute.

1 files changed, 96 insertions, 10 deletions

diff --git a/sl0ppij/zellij-server/src/lib.rs b/sl0ppij/zellij-server/src/lib.rs
index 1237ff8..8671b7c 100644
--- a/sl0ppij/zellij-server/src/lib.rs
+++ b/sl0ppij/zellij-server/src/lib.rs
@@ -635,6 +635,12 @@ pub(crate) struct SessionState {
     /// empty synthetic reply so `Screen`'s `forward_in_flight` slot
     /// releases and the queued forwards keep moving.
     forwards_in_flight: HashMap<u32, ClientId>,
+    /// Where to start looking for the next client id.
+    ///
+    /// sl0p fork: ids used to be "lowest free", which made them recycle
+    /// instantly. That is a correctness problem, not a cosmetic one — see
+    /// `new_client`.
+    next_client_id: ClientId,
 }
 
 impl SessionState {
@@ -645,26 +651,56 @@ impl SessionState {
             watchers: HashMap::new(),
             last_active_client: None,
             forwards_in_flight: HashMap::new(),
+            next_client_id: 1,
         }
     }
+    /// Hand out a client id, advancing through the id space instead of reusing
+    /// the lowest free one.
+    ///
+    /// sl0p fork. Recycling ids immediately made `zellij action` fail silently
+    /// about 20% of the time on a 2-vCPU box, with exit 0, empty stdout, no
+    /// stderr and nothing in the server log:
+    ///
+    ///   1. every action first probes liveness (`sessions::assert_socket`),
+    ///      which connects, sends `ConnStatus`, and disconnects. That probe
+    ///      takes a client id.
+    ///   2. the probe's id is freed, so the ACTION connection that follows
+    ///      microseconds later is handed the very same id.
+    ///   3. the main server thread then gets round to the queued
+    ///      `ServerInstruction::ConnStatus(id)`, whose handler replies and then
+    ///      `remove_client!(id)` — removing the *action* client that now owns
+    ///      that id.
+    ///   4. the action's reply is sent with `send_to_client`, which silently
+    ///      returns `Ok(())` for an unknown client, so the payload is dropped
+    ///      and the client sees only the disconnect.
+    ///
+    /// Advancing the cursor means a stale removal can only hit a live client
+    /// after the whole u16 space has been cycled, which is not a race any more.
+    /// Ids must still wrap (`ClientId` is a u16 and every CLI invocation burns
+    /// two), so in-use ids are skipped rather than assumed free.
     pub fn new_client(&mut self) -> ClientId {
-        let all_ids: HashSet<ClientId> = self
+        let in_use: HashSet<ClientId> = self
             .clients
             .keys()
             .copied()
             .chain(self.watchers.keys().copied())
             .collect();
 
-        let mut next_client_id = 1;
-        loop {
-            if all_ids.contains(&next_client_id) {
-                next_client_id += 1;
-            } else {
-                break;
-            }
+        // 0 is never handed out, matching the previous behaviour.
+        let mut candidate = if self.next_client_id == 0 {
+            1
+        } else {
+            self.next_client_id
+        };
+        let mut skipped: u32 = 0;
+        while in_use.contains(&candidate) && skipped <= ClientId::MAX as u32 {
+            candidate = candidate.checked_add(1).unwrap_or(1);
+            skipped += 1;
         }
-        self.clients.insert(next_client_id, None);
-        next_client_id
+
+        self.next_client_id = candidate.checked_add(1).unwrap_or(1);
+        self.clients.insert(candidate, None);
+        candidate
     }
     pub fn associate_pipe_with_client(&mut self, pipe_id: String, client_id: ClientId) {
         self.pipes.insert(pipe_id, client_id);
@@ -787,6 +823,56 @@ impl SessionState {
 mod session_state_tests {
     use super::*;
 
+    /// sl0p fork regression tests: ids must not recycle immediately.
+    ///
+    /// Instant reuse let a liveness probe's queued `ConnStatus` removal delete
+    /// the *next* connection, which silently swallowed `zellij action` output
+    /// about 20% of the time.
+    #[test]
+    fn client_ids_are_not_reused_immediately() {
+        let mut state = SessionState::new();
+        let first = state.new_client();
+        state.remove_client(first);
+        let second = state.new_client();
+        assert_ne!(
+            first, second,
+            "an id freed a moment ago must not be handed straight back out"
+        );
+    }
+
+    #[test]
+    fn client_ids_advance_and_skip_live_ones() {
+        let mut state = SessionState::new();
+        let a = state.new_client();
+        let b = state.new_client();
+        let c = state.new_client();
+        assert!(a < b && b < c, "ids should advance: {a} {b} {c}");
+
+        // freeing a middle id must not make the next allocation collide with a
+        // still-connected client
+        state.remove_client(b);
+        let d = state.new_client();
+        assert_ne!(d, a);
+        assert_ne!(d, c);
+    }
+
+    /// ClientId is a u16 and every CLI invocation burns two, so the cursor has
+    /// to wrap rather than run out.
+    #[test]
+    fn client_ids_wrap_without_colliding_or_hanging() {
+        let mut state = SessionState::new();
+        let held = state.new_client();
+        let mut last = held;
+        for _ in 0..(u16::MAX as u32 + 10) {
+            let id = state.new_client();
+            assert_ne!(id, held, "wrapped onto a still-connected client");
+            assert_ne!(id, 0, "0 is never a valid client id");
+            state.remove_client(id);
+            last = id;
+        }
+        let _ = last;
+    }
+
     fn with_client(id: ClientId) -> SessionState {
         let mut s = SessionState::new();
         s.clients.insert(id, None);