Restore shell compatibility and skip converged scans

Author: myobie

CHANGELOG.md

@@ -60,6 +60,19 @@ EXPERIMENTAL, so on-disk formats and the CLI may change without notice.
 
 ### Fixed
 
+- **Remote shell upgrades remain mixed-version compatible.** Resumable shell
+  sessions now use their own `fabric/shell/1` ALPN. The established
+  `fabric/shell/0` ALPN keeps its original raw shell framing in both directions,
+  and new clients explicitly fall back to it when an older peer does not
+  advertise `shell/1`. Transport reconnect still resumes the same remote PTY,
+  and signal exits restore the caller's exact terminal settings.
+
+- **Converged periodic syncs no longer hash the full tree twice.** An inbound
+  peer whose manifest exactly matches the local node can bypass both folder
+  scans when the local content store is complete. Differing manifests, missing
+  content, and every potentially mutating reconcile retain the guarded
+  pre-merge and completion scans that protect local archive/delete intent.
+
 - **Inherited catalog tombstones no longer leave folders divergent.** Under
   catalog policy, any surviving physical copy now advances an inherited
   Tombstone to a higher Present version and supplies its content over the wire,

README.md

@@ -694,7 +694,10 @@ fabric shell <peer>
 Open an interactive remote shell on a trusted peer over fabric. The server side
 must have been started with `fabric up --allow-shell`; a default `fabric up`
 refuses shell requests. The shell runs as the remote daemon's user and uses the
-remote user's `$SHELL`.
+remote user's `$SHELL`. Current peers negotiate resumable `fabric/shell/1`, so
+the same remote PTY survives a transient transport drop. A new client
+automatically falls back to the byte-compatible one-shot `fabric/shell/0`
+protocol when the peer is running an older Fabric release.
 
 Enabling shell is a security-sensitive opt-in: every trusted peer in
 `peers.toml` can obtain a remote shell while `--allow-shell` is active. Keep the

src/daemon.rs

@@ -999,10 +999,10 @@ impl DaemonState {
         let listener_cancel = CancellationToken::new();
         let lease = DialListenerLease::new(self.active_dial_listeners.clone());
 
-        // Built-in exec remains a one-shot raw framed stream. Built-in shell
-        // rides the resumable tunnel path so its PTY outlives transient iroh
-        // attaches; local reconnect notices are encoded as shell status frames.
-        let listener_task = if alpn == exec::EXEC_ALPN {
+        // Built-in exec and legacy shell/0 remain one-shot raw framed streams.
+        // Resumable shell/1 negotiates its own tunnel path and falls back to
+        // shell/0 when the peer does not advertise the new ALPN.
+        let listener_task = if alpn == exec::EXEC_ALPN || alpn == shell::SHELL_ALPN {
             tokio::spawn(run_raw_dial_socket(
                 listener,
                 self.endpoint_rx(),
@@ -1014,8 +1014,21 @@ impl DaemonState {
                 self.dial_slots.clone(),
                 lease,
             ))
+        } else if alpn == shell::RESUMABLE_SHELL_ALPN {
+            tokio::spawn(run_shell_dial_socket(
+                listener,
+                self.endpoint_rx(),
+                self.home.clone(),
+                peer.to_string(),
+                peer_addr.clone(),
+                listener_cancel.clone(),
+                self.cancel.clone(),
+                self.tunnel_drop_rx(),
+                self.dial_failures.clone(),
+                self.dial_slots.clone(),
+                lease,
+            ))
         } else {
-            let notices = (alpn == shell::SHELL_ALPN).then(shell_client_notices);
             tokio::spawn(run_dial_socket(
                 listener,
                 self.endpoint_rx(),
@@ -1028,7 +1041,7 @@ impl DaemonState {
                 self.dial_failures.clone(),
                 self.dial_slots.clone(),
                 lease,
-                notices,
+                None,
             ))
         };
         sockets.insert(
@@ -2437,7 +2450,7 @@ async fn process_control_request(
                 .dial_alpn(
                     &peer,
                     shell::SHELL_PROTOCOL,
-                    shell::SHELL_ALPN.to_vec(),
+                    shell::RESUMABLE_SHELL_ALPN.to_vec(),
                     false,
                 )
                 .await?;
@@ -2584,8 +2597,14 @@ async fn process_incoming_iroh(incoming: Incoming, state: Arc<DaemonState>) -> R
     }
     if alpn == shell::SHELL_ALPN {
         let connection = accepting.await?;
-        log_connection_paths("builtin_shell_accept", &connection);
-        handle_builtin_shell(connection, state).await?;
+        log_connection_paths("builtin_legacy_shell_accept", &connection);
+        handle_builtin_legacy_shell(connection, state).await?;
+        return Ok(());
+    }
+    if alpn == shell::RESUMABLE_SHELL_ALPN {
+        let connection = accepting.await?;
+        log_connection_paths("builtin_resumable_shell_accept", &connection);
+        handle_builtin_resumable_shell(connection, state).await?;
         return Ok(());
     }
     if alpn == exec::EXEC_ALPN {
@@ -2639,7 +2658,26 @@ async fn handle_builtin_echo(connection: Connection, state: Arc<DaemonState>) ->
     Ok(())
 }
 
-async fn handle_builtin_shell(connection: Connection, state: Arc<DaemonState>) -> Result<()> {
+async fn handle_builtin_legacy_shell(
+    connection: Connection,
+    state: Arc<DaemonState>,
+) -> Result<()> {
+    let peer = connection.remote_id().to_string();
+    let (mut send, mut recv) = connection.accept_bi().await?;
+    if state.allow_shell {
+        shell::serve_shell_session(&mut recv, &mut send, &peer).await?;
+    } else {
+        shell::serve_shell_disabled(&mut send).await?;
+    }
+    send.finish()?;
+    connection.closed().await;
+    Ok(())
+}
+
+async fn handle_builtin_resumable_shell(
+    connection: Connection,
+    state: Arc<DaemonState>,
+) -> Result<()> {
     let peer_id = connection.remote_id();
     let (send, recv) = connection.accept_bi().await?;
     tunnel::serve_connection(
@@ -2680,10 +2718,12 @@ async fn handle_sync(connection: Connection, state: Arc<DaemonState>) -> Result<
     let (send, recv) = connection.accept_bi().await?;
     let stream = tokio::io::join(recv, send);
     let resolver_engine = engine.clone();
-    let outcome = sync::wire::run_server(stream, move |name| {
+    let outcome = sync::wire::run_server(stream, move |name, remote_manifest| {
         let engine = resolver_engine.clone();
         async move {
-            let prepared = engine.prepare_inbound(&name).await?;
+            let prepared = engine
+                .prepare_inbound_for_manifest(&name, &remote_manifest)
+                .await?;
             Ok(prepared.map(|prepared| (prepared.node(), prepared)))
         }
     })
@@ -2706,9 +2746,10 @@ async fn handle_sync(connection: Connection, state: Arc<DaemonState>) -> Result<
 }
 
 fn accepted_alpns(exposures: &HashMap<Vec<u8>, Exposure>) -> Vec<Vec<u8>> {
-    let mut alpns = Vec::with_capacity(exposures.len() + 4);
+    let mut alpns = Vec::with_capacity(exposures.len() + 5);
     alpns.push(BUILTIN_ECHO_ALPN.to_vec());
     alpns.push(shell::SHELL_ALPN.to_vec());
+    alpns.push(shell::RESUMABLE_SHELL_ALPN.to_vec());
     alpns.push(exec::EXEC_ALPN.to_vec());
     alpns.push(SYNC_ALPN.to_vec());
     alpns.extend(exposures.keys().cloned());
@@ -2718,6 +2759,7 @@ fn accepted_alpns(exposures: &HashMap<Vec<u8>, Exposure>) -> Vec<Vec<u8>> {
 fn matches_reserved_alpn(alpn: &[u8]) -> bool {
     alpn == BUILTIN_ECHO_ALPN
         || alpn == shell::SHELL_ALPN
+        || alpn == shell::RESUMABLE_SHELL_ALPN
         || alpn == exec::EXEC_ALPN
         || alpn == SYNC_ALPN
 }
@@ -3133,6 +3175,140 @@ async fn run_dial_socket(
     }
 }
 
+async fn run_shell_dial_socket(
+    listener: UnixListener,
+    endpoint_rx: watch::Receiver<CurrentEndpoint>,
+    home: FabricHome,
+    peer: String,
+    peer_addr: EndpointAddr,
+    listener_cancel: CancellationToken,
+    daemon_cancel: CancellationToken,
+    drop_rx: watch::Receiver<u64>,
+    dial_failures: Arc<FailureBackoff>,
+    dial_slots: Arc<Semaphore>,
+    _lease: DialListenerLease,
+) {
+    loop {
+        tokio::select! {
+            biased;
+            _ = listener_cancel.cancelled() => break,
+            _ = daemon_cancel.cancelled() => break,
+            accepted = listener.accept() => {
+                let Ok((local, _)) = accepted else {
+                    break;
+                };
+                let permit = tokio::select! {
+                    biased;
+                    _ = listener_cancel.cancelled() => break,
+                    _ = daemon_cancel.cancelled() => break,
+                    permit = dial_slots.clone().acquire_owned() => {
+                        let Ok(permit) = permit else {
+                            break;
+                        };
+                        permit
+                    }
+                };
+                let endpoint_rx = endpoint_rx.clone();
+                let home = home.clone();
+                let peer = peer.clone();
+                let peer_addr = peer_addr.clone();
+                let cancel = daemon_cancel.clone();
+                let drop_rx = drop_rx.clone();
+                let dial_failures = dial_failures.clone();
+                tokio::spawn(async move {
+                    let _permit = permit;
+                    if !dial_failures.wait(&cancel).await {
+                        return;
+                    }
+                    match handle_shell_dial_socket_connection(
+                        local,
+                        endpoint_rx,
+                        home,
+                        peer,
+                        peer_addr,
+                        cancel,
+                        drop_rx,
+                    )
+                    .await
+                    {
+                        Ok(()) => dial_failures.record_success().await,
+                        Err(error) => {
+                            dial_failures
+                                .record_failure("shell dial socket connection failed", &error)
+                                .await;
+                        }
+                    }
+                });
+            }
+        }
+    }
+}
+
+async fn handle_shell_dial_socket_connection(
+    local: UnixStream,
+    endpoint_rx: watch::Receiver<CurrentEndpoint>,
+    home: FabricHome,
+    peer: String,
+    peer_addr: EndpointAddr,
+    cancel: CancellationToken,
+    drop_rx: watch::Receiver<u64>,
+) -> Result<()> {
+    let endpoint = endpoint_rx.borrow().endpoint.clone();
+    match endpoint
+        .connect(peer_addr.clone(), shell::RESUMABLE_SHELL_ALPN)
+        .await
+    {
+        Ok(connection) => {
+            tunnel::run_client_connection_with_initial(
+                local,
+                endpoint_rx,
+                home,
+                peer,
+                shell::RESUMABLE_SHELL_ALPN.to_vec(),
+                cancel,
+                drop_rx,
+                Some(shell_client_notices()),
+                connection,
+            )
+            .await
+        }
+        Err(error) => {
+            let error = anyhow::Error::new(error);
+            if !shell_resumable_alpn_unsupported(&error) {
+                // A transient initial failure should retain the resumable
+                // attach loop and its backoff. Only a definitive ALPN mismatch
+                // is allowed to downgrade the session to legacy shell/0.
+                return tunnel::run_client_connection(
+                    local,
+                    endpoint_rx,
+                    home,
+                    peer,
+                    shell::RESUMABLE_SHELL_ALPN.to_vec(),
+                    cancel,
+                    drop_rx,
+                    Some(shell_client_notices()),
+                )
+                .await;
+            }
+            handle_raw_dial_socket_connection(
+                local,
+                endpoint,
+                peer_addr,
+                shell::SHELL_ALPN.to_vec(),
+            )
+            .await
+        }
+    }
+}
+
+fn shell_resumable_alpn_unsupported(error: &anyhow::Error) -> bool {
+    error.chain().any(|cause| {
+        let message = cause.to_string().to_ascii_lowercase();
+        message.contains("peer doesn't support any known protocol")
+            || message.contains("no application protocol")
+    })
+}
+
 fn shell_client_notices() -> tunnel::ClientConnectionNotices {
     tunnel::ClientConnectionNotices::new(|event| {
         let encoded = match event {
@@ -3757,9 +3933,9 @@ mod tests {
         let state = node.state();
         let cases = [
             ("peer-a", exec::EXEC_PROTOCOL, exec::EXEC_ALPN),
-            ("peer-a", shell::SHELL_PROTOCOL, shell::SHELL_ALPN),
+            ("peer-a", shell::SHELL_PROTOCOL, shell::RESUMABLE_SHELL_ALPN),
             ("peer-b", exec::EXEC_PROTOCOL, exec::EXEC_ALPN),
-            ("peer-b", shell::SHELL_PROTOCOL, shell::SHELL_ALPN),
+            ("peer-b", shell::SHELL_PROTOCOL, shell::RESUMABLE_SHELL_ALPN),
         ];
 
         // More replacements than the production macOS soft FD limit. The old

src/shell.rs

@@ -11,8 +11,12 @@ use tokio::{
 };
 use tokio_util::sync::CancellationToken;
 
+/// Legacy one-shot shell framing. This ALPN is wire-compatible with every
+/// released Fabric shell and must never carry generic tunnel frames.
 pub const SHELL_ALPN: &[u8] = b"fabric/shell/0";
 pub const SHELL_PROTOCOL: &str = "fabric/shell/0";
+/// Resumable shell framing carried by the generic tunnel session protocol.
+pub const RESUMABLE_SHELL_ALPN: &[u8] = b"fabric/shell/1";
 
 const MAX_FRAME_LEN: usize = 1024 * 1024;
 const CLIENT_STDIN: u8 = 1;