fix(move): read the host left behind until every contact has followed

app/desktop/src-tauri/src/ports/mod.rs

@@ -105,11 +105,20 @@ impl DesktopPlatformPorts {
         let profile_manager = crate::platform::profile::ProfileManager::from_inner(
             self.transport.profile_inner.clone(),
         );
-        let credential = self
+        // A request to the host a move left behind carries that host's
+        // runtime id, and goes under the credential kept for it.
+        let previous = self
             .runtime_auth
-            .ensure(&profile_manager, force_refresh)
-            .await?
-            .ok_or_else(|| anyhow::anyhow!("runtime_auth_error:enrollment_required"))?;
+            .ensure_previous(&profile_manager, runtime_id)
+            .await?;
+        let credential = match previous {
+            Some(credential) => credential,
+            None => self
+                .runtime_auth
+                .ensure(&profile_manager, force_refresh)
+                .await?
+                .ok_or_else(|| anyhow::anyhow!("runtime_auth_error:enrollment_required"))?,
+        };
         if credential.runtime_id != *runtime_id || credential.device_id != *device_id {
             anyhow::bail!("runtime_auth_error:runtime_mismatch");
         }

app/desktop/src-tauri/src/runtime_auth.rs

@@ -644,6 +644,61 @@ impl RuntimeAuthManager {
         Ok(Some(refreshed))
     }
 
+    /// The credential for the host a move left behind, which the engine still
+    /// reads, refreshed there when it nears expiry. `None` unless the profile
+    /// keeps one for `runtime_id` and still names that host.
+    pub async fn ensure_previous(
+        &self,
+        profile_manager: &ProfileManager,
+        runtime_id: &str,
+    ) -> Result<Option<DeviceRuntimeAuth>> {
+        let (credential, deployment, identity) = {
+            let inner = profile_manager.inner.read().await;
+            let Some(profile) = inner.active_profile.as_ref() else {
+                return Ok(None);
+            };
+            let Some(credential) = profile
+                .load_previous_runtime_credential()
+                .context("load previous runtime credential")?
+            else {
+                return Ok(None);
+            };
+            if credential.runtime_id != runtime_id {
+                return Ok(None);
+            }
+            let snapshot = profile
+                .load_snapshot()
+                .context("load snapshot for previous runtime authorization")?;
+            let Some(previous) = snapshot
+                .deployment
+                .and_then(|deployment| deployment.previous_inbox)
+            else {
+                return Ok(None);
+            };
+            let Some(identity) = snapshot.local_identity.map(|value| value.state) else {
+                return Ok(None);
+            };
+            (credential, previous.deployment_bundle, identity)
+        };
+        if !runtime_credential_needs_refresh(Some(&credential), now_ms()?) {
+            return Ok(Some(credential));
+        }
+        let refreshed = request_runtime_credential_with_retry(
+            &deployment,
+            &identity,
+            ProofPurpose::Refresh,
+            None,
+        )
+        .await?;
+        let inner = profile_manager.inner.read().await;
+        if let Some(profile) = inner.active_profile.as_ref() {
+            profile
+                .save_previous_runtime_credential(Some(refreshed.clone()))
+                .context("persist refreshed previous runtime credential")?;
+        }
+        Ok(Some(refreshed))
+    }
+
     pub async fn enroll(
         &self,
         profile_manager: &ProfileManager,

src/cli/app.rs

@@ -2244,6 +2244,7 @@ fn load_driver(profile: &Profile) -> Result<CoreDriver> {
     });
     let mut driver = CoreDriver::from_snapshot(snapshot, contact_share_url)?;
     driver.set_runtime_credential(profile.load_runtime_credential()?);
+    driver.set_previous_runtime_credential(profile.load_previous_runtime_credential()?);
     driver.attach_store(profile.storage_session());
     Ok(driver)
 }

src/cli/driver.rs

@@ -72,6 +72,7 @@ pub struct DriverRuntime {
     recent_appends: Vec<Envelope>,
     recent_messages: Vec<(String, MessageType)>,
     runtime_credential: Option<DeviceRuntimeAuth>,
+    previous_runtime_credential: Option<DeviceRuntimeAuth>,
     store: Option<Arc<ProfileStorageSession>>,
 }
 
@@ -136,6 +137,7 @@ impl CoreDriver {
                 recent_appends: Vec::new(),
                 recent_messages: Vec::new(),
                 runtime_credential: None,
+                previous_runtime_credential: None,
                 store: None,
             },
             suppress_realtime: false,
@@ -154,10 +156,23 @@ impl CoreDriver {
         self.runtime.latest_snapshot.as_ref()
     }
 
+    /// Install the credential for the current runtime. One for another
+    /// runtime is a move, and the one it replaces is kept for the host left
+    /// behind, which the engine still reads.
     pub fn set_runtime_credential(&mut self, credential: Option<DeviceRuntimeAuth>) {
+        if crate::cli::profile::runtime_changes(
+            self.runtime.runtime_credential.as_ref(),
+            credential.as_ref(),
+        ) {
+            self.runtime.previous_runtime_credential = self.runtime.runtime_credential.take();
+        }
         self.runtime.runtime_credential = credential;
     }
 
+    pub fn set_previous_runtime_credential(&mut self, credential: Option<DeviceRuntimeAuth>) {
+        self.runtime.previous_runtime_credential = credential;
+    }
+
     fn inject_runtime_authorization(
         &self,
         headers: &mut BTreeMap<String, String>,
@@ -170,14 +185,22 @@ impl CoreDriver {
         else {
             return Ok(());
         };
-        let credential = self
-            .runtime
-            .runtime_credential
-            .as_ref()
-            .ok_or_else(|| anyhow!("runtime credential is not available"))?;
-        if credential.runtime_id != *runtime_id || credential.device_id != *device_id {
-            return Err(anyhow!("runtime credential scope mismatch"));
-        }
+        // The current runtime's credential, or the one kept for the host a
+        // move left behind, which the engine still reads.
+        let held = [
+            self.runtime.runtime_credential.as_ref(),
+            self.runtime.previous_runtime_credential.as_ref(),
+        ];
+        if held.iter().all(Option::is_none) {
+            return Err(anyhow!("runtime credential is not available"));
+        }
+        let credential = held
+            .into_iter()
+            .flatten()
+            .find(|credential| {
+                credential.runtime_id == *runtime_id && credential.device_id == *device_id
+            })
+            .ok_or_else(|| anyhow!("runtime credential scope mismatch"))?;
         headers.insert(
             "Authorization".into(),
             format!("Bearer {}", credential.token),

src/cli/profile.rs

@@ -138,6 +138,10 @@ pub struct ProfilePrivateState {
         skip_serializing_if = "Option::is_none"
     )]
     pub runtime_credential: Option<DeviceRuntimeAuth>,
+    /// The credential this device held at the host it moved away from, kept
+    /// while the engine still reads that host (its `previous_inbox`).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub previous_runtime_credential: Option<DeviceRuntimeAuth>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub pending_runtime_provisioning: Option<PendingRuntimeProvisioning>,
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
@@ -146,6 +150,18 @@ pub struct ProfilePrivateState {
     pub attachment_cache: BTreeMap<String, AttachmentCacheEntry>,
 }
 
+/// Whether `replacement` is a credential for another runtime than `held`,
+/// that is, whether storing it is a move.
+pub fn runtime_changes(
+    held: Option<&DeviceRuntimeAuth>,
+    replacement: Option<&DeviceRuntimeAuth>,
+) -> bool {
+    match (held, replacement) {
+        (Some(held), Some(replacement)) => held.runtime_id != replacement.runtime_id,
+        _ => false,
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
 pub struct RuntimeSecrets {
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -243,6 +259,7 @@ impl Default for ProfilePrivateState {
             runtime_secrets: RuntimeSecrets::default(),
             runtime_secret_rotation: None,
             runtime_credential: None,
+            previous_runtime_credential: None,
             pending_runtime_provisioning: None,
             settings: BTreeMap::new(),
             attachment_cache: BTreeMap::new(),
@@ -749,12 +766,31 @@ impl Profile {
         Ok(self.load_private_state()?.runtime_credential)
     }
 
+    /// Store the credential for the current runtime. One for another runtime
+    /// than the credential it replaces is a move, and the replaced one is
+    /// kept as the previous credential, for the host left behind.
     pub fn save_runtime_credential(&self, credential: Option<DeviceRuntimeAuth>) -> Result<()> {
         let mut private = self.load_private_state()?;
+        if runtime_changes(private.runtime_credential.as_ref(), credential.as_ref()) {
+            private.previous_runtime_credential = private.runtime_credential.take();
+        }
         private.runtime_credential = credential;
         self.save_private_state(&private)
     }
 
+    pub fn load_previous_runtime_credential(&self) -> Result<Option<DeviceRuntimeAuth>> {
+        Ok(self.load_private_state()?.previous_runtime_credential)
+    }
+
+    pub fn save_previous_runtime_credential(
+        &self,
+        credential: Option<DeviceRuntimeAuth>,
+    ) -> Result<()> {
+        let mut private = self.load_private_state()?;
+        private.previous_runtime_credential = credential;
+        self.save_private_state(&private)
+    }
+
     pub fn load_pending_runtime_provisioning(&self) -> Result<Option<PendingRuntimeProvisioning>> {
         Ok(self.load_private_state()?.pending_runtime_provisioning)
     }

src/ffi_api/engine/direct.rs

@@ -21,7 +21,7 @@ impl CoreEngine {
         let inbox_moved = previous_inbox
             .as_ref()
             .is_some_and(|endpoint| endpoint != &bundle.inbox_http_endpoint);
-        self.state.deployment_bundle = Some(bundle);
+        let left_behind = self.state.deployment_bundle.replace(bundle);
         // A new deployment is new content for a bundle already published, and
         // a peer accepts a bundle only at a higher revision: at the one it
         // holds it keeps what it has, so an unchanged revision would leave it
@@ -57,17 +57,15 @@ impl CoreEngine {
             .effects
             .extend(self.local_shared_state_publish_effects()?);
         if inbox_moved {
-            // The new host numbers its records from 1. A cursor carried over
-            // would wait for that host's head to pass the old ack before
-            // fetching anything, and the host refuses an ack above its head.
-            // Records held back for a commit cannot resolve either: whatever
-            // they wait for is at the old host, which this device no longer
-            // reads.
-            if let Ok(device_id) = self.local_device_id_required() {
-                self.state
-                    .sync_states
-                    .insert(device_id.clone(), SyncEngine::new_device_state(&device_id));
-                self.state.pending_acks.remove(&device_id);
+            // The new host numbers its records from 1, so its cursor starts
+            // afresh; the old one goes on reading the host left behind, where
+            // contacts write until they apply the lane notice (see
+            // `previous_inbox`).
+            if let (Ok(device_id), Some(left_behind)) =
+                (self.local_device_id_required(), left_behind)
+            {
+                let ops = self.begin_previous_inbox_drain(left_behind, &device_id);
+                output.effects.push(persist_effect(&self.state, ops));
             }
             // The new host's KeyPackage pool starts empty, and until it is
             // filled every claim there falls back to the last-resort package.
@@ -3843,11 +3841,16 @@ impl CoreEngine {
                 .and_then(|conversation| conversation.lanes.as_ref())
                 .is_some()
         {
-            self.switch_outbound_lane(conversation_id, inbound_lane)?;
+            let readdressed = self.switch_outbound_lane(conversation_id, inbound_lane)?;
             output.state_update.conversations_changed = true;
             persist_ops.push(PersistOp::SaveConversation {
                 conversation_id: conversation_id.to_string(),
             });
+            persist_ops.extend(
+                readdressed
+                    .into_iter()
+                    .map(|message_id| PersistOp::SaveOutgoingEnvelope { message_id }),
+            );
         }
         if !persist_ops.is_empty() {
             output

src/ffi_api/engine/lanes.rs

@@ -21,17 +21,28 @@ impl CoreEngine {
     }
 
     pub(super) fn conversation_id_for_lane(&self, lane: &str) -> Option<String> {
-        self.state.lane_index.get(lane).cloned().or_else(|| {
-            self.state
-                .conversations
-                .iter()
-                .find_map(|(id, conversation)| {
-                    conversation.lanes.as_ref().and_then(|lanes| {
-                        (lanes.inbound_lane == lane || lanes.outbound_lane == lane)
-                            .then(|| id.clone())
+        self.state
+            .lane_index
+            .get(lane)
+            .cloned()
+            .or_else(|| {
+                self.state
+                    .conversations
+                    .iter()
+                    .find_map(|(id, conversation)| {
+                        conversation.lanes.as_ref().and_then(|lanes| {
+                            (lanes.inbound_lane == lane || lanes.outbound_lane == lane)
+                                .then(|| id.clone())
+                        })
                     })
-                })
-        })
+            })
+            .or_else(|| {
+                // A lane at the host this device moved away from, still read.
+                self.state
+                    .previous_inbox
+                    .as_ref()
+                    .and_then(|previous| previous.lanes.get(lane).cloned())
+            })
     }
 
     pub(super) fn index_inbound_lane(&mut self, lane: &str, conversation_id: &str) {
@@ -111,7 +122,7 @@ impl CoreEngine {
         &mut self,
         conversation_id: &str,
         new_outbound: String,
-    ) -> CoreResult<()> {
+    ) -> CoreResult<Vec<String>> {
         if !is_opaque_id(&new_outbound) {
             return Err(CoreError::invalid_input("lane must be a 128-bit hex id"));
         }
@@ -123,7 +134,7 @@ impl CoreEngine {
             .map(|lanes| lanes.outbound_lane.clone())
             .ok_or_else(|| CoreError::invalid_state("conversation has no outbound lane"))?;
         if old_outbound == new_outbound {
-            return Ok(());
+            return Ok(Vec::new());
         }
         if let Some(lanes) = self
             .state
@@ -135,7 +146,34 @@ impl CoreEngine {
         }
         self.state.lane_index.remove(&old_outbound);
         self.index_inbound_lane(&new_outbound, conversation_id);
-        Ok(())
+        // What is still queued on the old lane goes to the new one. A lane is
+        // read by the host, not inside the wrap, so the envelope is otherwise
+        // unchanged; its retries start again, since the failures, if any, were
+        // the old host's. An envelope in flight keeps its lane: if it fails,
+        // `flush_outbox` readdresses it before sending it again. Returns the
+        // envelopes readdressed.
+        let mut readdressed = Vec::new();
+        for item in &mut self.state.pending_outbox {
+            if item.envelope.lane == old_outbound && !item.in_flight {
+                item.envelope.lane = new_outbound.clone();
+                item.retries = 0;
+                readdressed.push(item.envelope.mid.clone());
+            }
+        }
+        Ok(readdressed)
+    }
+
+    /// The lane `peer_user_id`'s direct conversation now writes on, if the
+    /// peer has exactly one.
+    pub(super) fn outbound_lane_for_peer(&self, peer_user_id: &str) -> Option<String> {
+        let mut lanes = self
+            .state
+            .conversations
+            .values()
+            .filter(|conversation| conversation.peer_user_id == peer_user_id)
+            .filter_map(|conversation| conversation.lanes.as_ref());
+        let lane = lanes.next()?.outbound_lane.clone();
+        lanes.next().is_none().then_some(lane)
     }
 
     /// Whether `lane` is the inbound lane of a conversation this device holds.
@@ -359,7 +397,8 @@ impl CoreEngine {
                 .conversations
                 .get(&conversation_id)
                 .and_then(|conversation| conversation.lanes.as_ref())
-                .is_some_and(|lanes| lanes.inbound_lane == record.envelope.lane);
+                .is_some_and(|lanes| lanes.inbound_lane == record.envelope.lane)
+                || self.is_previous_inbound_lane(&conversation_id, &record.envelope.lane);
             if !inbound {
                 return Ok(InboundFrameResolution::Rejected);
             }