Preserve encrypted receive history when cache writes fail

crates/chatd/src/adapters/tiny.rs

@@ -73,6 +73,11 @@ pub fn prove_cache_history_limit() -> Result<(), String> {
     e2ee::prove_cache_history_limit()
 }
 
+/// The lab's durable receive/cache commit proof.
+pub fn prove_opened_history_cache_commit() -> Result<(), String> {
+    e2ee::prove_opened_history_cache_commit()
+}
+
 /// The lab's reauthentication history proof, using real encrypted device/cache files.
 pub fn prove_reauthenticated_history() -> Result<(), String> {
     e2ee::prove_reauthenticated_history()

crates/chatd/src/adapters/tiny/e2ee/receive_tests.rs

@@ -304,6 +304,148 @@ fn add(
     (bundle, proposals)
 }
 
+/// A failed sealed-cache append must not leave only a consumed receive ratchet. After storage is
+/// repaired, the same ciphertext opens and survives a cold restart with the same keys.
+#[test]
+fn opened_history_survives_a_failed_cache_append_and_restart() {
+    assert_eq!(prove_opened_history_cache_commit(), Ok(()));
+}
+
+/// Cache durable + state save lagging: restart serves the sealed row; a mismatched claim still
+/// spends no key and does not replace authenticated cache metadata.
+#[test]
+fn sealed_cache_survives_state_save_lag_and_rejects_mismatched_replay() {
+    let fixture = tempfile::tempdir().unwrap();
+    let a_dir = fixture.path().join("sender");
+    let b_dir = fixture.path().join("receiver");
+    private_dir(&a_dir).unwrap();
+    private_dir(&b_dir).unwrap();
+    let alice = UserId::random().unwrap();
+    let bea = UserId::random().unwrap();
+    let mut a = Engine::create(
+        alice,
+        Box::new(FileStore::new(&a_dir)),
+        StateKey::generate().unwrap(),
+        STATE,
+    )
+    .unwrap();
+    let state_key = StateKey::generate().unwrap();
+    let cache_key = CacheKey::generate().unwrap();
+    let mut b = Engine::create(
+        bea,
+        Box::new(FileStore::new(&b_dir)),
+        state_key.clone(),
+        STATE,
+    )
+    .unwrap();
+    a.verify_devices(bea, &[record(&b, None, 1)]).unwrap();
+    b.verify_devices(alice, &[record(&a, None, 1)]).unwrap();
+    let conv = ConversationId::random().unwrap();
+    let server = DeviceKey::generate().unwrap();
+    let public = server.public_key();
+    a.create_group(conv, &public).unwrap();
+    let (bundle, _) = add(&mut a, conv, &server, &b.key_packages(1).unwrap().remove(0));
+    b.handle_welcome(conv, &bundle.welcomes[0].data, Some(&public))
+        .unwrap();
+    let sent = a
+        .encrypt(
+            conv,
+            Kind::Message,
+            None,
+            &Envelope::message("lagged state history"),
+        )
+        .unwrap();
+    let id = conv.to_string();
+    let before = std::fs::read(b_dir.join(STATE)).unwrap();
+    let saved = Saved {
+        base_url: "https://diagnostic.invalid".into(),
+        token: "synthetic".into(),
+        user_id: bea.to_string(),
+        email: None,
+        display_name: "Fixture".into(),
+        role: "member".into(),
+        plain_http_ok: false,
+        e2ee: None,
+    };
+    let author = alice.to_string();
+    let from = a.device_id().to_string();
+    let ciphertext = b64(&sent.ciphertext);
+    let row = Row {
+        id: "m1",
+        author_id: &author,
+        device_id: &from,
+        kind: "message",
+        ciphertext: Some(&ciphertext),
+        epoch: sent.epoch,
+        generation: None,
+        group_id: None,
+        target_id: None,
+        client_id: None,
+        created_at: "2026-01-01T00:00:00Z",
+    };
+    let mut device = {
+        let mut device = Device::new(&saved);
+        device.dir = b_dir.clone();
+        device.engine = Some(b);
+        device.cache_key = Some(cache_key.clone());
+        device.state = DeviceState::Ready;
+        device
+    };
+    let wrong_epoch = Row { epoch: sent.epoch + 1, ..row };
+    let refused = device.read(&id, std::slice::from_ref(&wrong_epoch));
+    assert!(matches!(refused.get("m1"), Some(Read::Undecryptable)),
+        "wire context must match before the receive is durably accepted");
+    assert_eq!(std::fs::read(b_dir.join(STATE)).unwrap(), before,
+        "a mismatched row must not consume the genuine ciphertext");
+    assert!(device.cache(&id).unwrap().rows.get("m1").is_none());
+    let opened = device.read(&id, std::slice::from_ref(&row));
+    assert!(
+        matches!(opened.get("m1"), Some(Read::Opened(o)) if o.text == "lagged state history")
+    );
+    // Simulate the opposite cut: durable cache present, on-disk ratchet not yet advanced.
+    std::fs::write(b_dir.join(STATE), &before).unwrap();
+    drop(device);
+    let b = Engine::open(Box::new(FileStore::new(&b_dir)), state_key.clone(), STATE)
+        .unwrap()
+        .unwrap();
+    let mut restarted = {
+        let mut device = Device::new(&saved);
+        device.dir = b_dir.clone();
+        device.engine = Some(b);
+        device.cache_key = Some(cache_key.clone());
+        device.state = DeviceState::Ready;
+        device
+    };
+    let after = restarted.read(&id, std::slice::from_ref(&row));
+    assert!(
+        matches!(after.get("m1"), Some(Read::Opened(o)) if o.text == "lagged state history"),
+        "sealed cache must serve history when the ratchet snapshot lags"
+    );
+    let other = UserId::random().unwrap().to_string();
+    let mismatched = Row {
+        id: "m1",
+        author_id: &other,
+        device_id: &from,
+        kind: "message",
+        ciphertext: Some(&ciphertext),
+        epoch: sent.epoch,
+        generation: None,
+        group_id: None,
+        target_id: None,
+        client_id: None,
+        created_at: "2026-01-01T00:00:00Z",
+    };
+    let refused = restarted.read(&id, std::slice::from_ref(&mismatched));
+    assert!(
+        matches!(refused.get("m1"), Some(Read::Undecryptable)),
+        "cached auth must refuse a mismatched claim without replacing the sealed row"
+    );
+    let again = restarted.read(&id, std::slice::from_ref(&row));
+    assert!(
+        matches!(again.get("m1"), Some(Read::Opened(o)) if o.text == "lagged state history")
+    );
+}
+
 #[test]
 fn mixed_generation_receive_does_not_spend_current_welcome_on_blocked_historical_rows() {
     let alice = UserId::parse("11111111-1111-4111-8111-111111111111").unwrap();

crates/chatd/src/main.rs

@@ -127,6 +127,15 @@ fn main() -> io::Result<()> {
         eprintln!("prove-cache-append-tail:0");
         return Ok(());
     }
+    // A failed sealed-cache append must not leave only a consumed receive ratchet.
+    if std::env::var("TINYCLIENT_PROVE").ok().as_deref() == Some("opened-history-cache-commit") {
+        if let Err(err) = chatd::adapters::tiny::prove_opened_history_cache_commit() {
+            eprintln!("opened-history-cache-commit: {err}");
+            std::process::exit(1);
+        }
+        eprintln!("prove-opened-history-cache-commit:0");
+        return Ok(());
+    }
     // The lab's keep-opened-sealed driver. A build without this returns to the server below and stays up.
     if std::env::var("TINYCLIENT_PROVE").ok().as_deref() == Some("keep-opened-sealed") {
         if let Err(err) = chatd::adapters::tiny::prove_keep_opened_sealed() {

crates/chatd/src/recorder.rs

@@ -254,7 +254,14 @@ mod tests {
             std::thread::sleep(Duration::from_millis(5));
         }
         drop(hub);
-        std::fs::remove_dir_all(dir).unwrap();
+        // Recorded is published before the reader exits and drops its Hub/SQLite handle.
+        // Windows cannot delete the database until that asynchronous cleanup finishes.
+        let cleanup_deadline = Instant::now() + Duration::from_secs(5);
+        while let Err(err) = std::fs::remove_dir_all(&dir) {
+            assert!(cfg!(windows) && err.raw_os_error() == Some(32)
+                && Instant::now() < cleanup_deadline, "recorder fixture cleanup: {err}");
+            std::thread::sleep(Duration::from_millis(5));
+        }
     }
 
     fn recording(since: u64) -> Recording {

crates/e2ee/src/messaging.rs

@@ -243,6 +243,50 @@ impl Engine {
         ciphertext: &[u8],
         claimed: &Claimed<'_>,
     ) -> Result<Incoming> {
+        // Existing callers keep the lenient save: a receive that also writes opened history must
+        // use [`Engine::decrypt_committed`] so a failed durable commit cannot leave only a consumed
+        // ratchet.
+        let (incoming, checkpoint) = self.decrypt_open(conv, ciphertext, claimed)?;
+        drop(checkpoint);
+        self.persist(false)?;
+        Ok(incoming)
+    }
+
+    /// Like [`Engine::decrypt`], but runs `commit` while the receive checkpoint is still
+    /// reversible and before the advanced ratchet is written. If `commit` fails, the pre-read
+    /// state is restored and the receive spends no key. After a successful commit, the caller
+    /// must already have durable recovery data. State saving then follows [`Engine::decrypt`]:
+    /// a failed save leaves the accepted receive consumed in memory and reports the error on
+    /// [`Engine::flush`]. It must never rewind an accepted receive to match a lagging disk.
+    pub fn decrypt_committed<F>(
+        &mut self,
+        conv: ConversationId,
+        ciphertext: &[u8],
+        claimed: &Claimed<'_>,
+        commit: F,
+    ) -> Result<Incoming>
+    where
+        F: FnOnce(&Incoming) -> Result<()>,
+    {
+        let (incoming, checkpoint) = self.decrypt_open(conv, ciphertext, claimed)?;
+        if let Err(error) = commit(&incoming) {
+            self.provider.restore(checkpoint);
+            self.groups.remove(&conv);
+            return Err(error);
+        }
+        drop(checkpoint);
+        self.persist(false)?;
+        Ok(incoming)
+    }
+
+    /// Authenticate and open a ciphertext, leaving the provider checkpoint for the caller to
+    /// drop (accept) or restore (refuse).
+    fn decrypt_open(
+        &mut self,
+        conv: ConversationId,
+        ciphertext: &[u8],
+        claimed: &Claimed<'_>,
+    ) -> Result<(Incoming, crate::provider::Checkpoint)> {
         self.repair_receive_allowed(conv)?;
         self.usable(conv)?;
         self.load_group(conv)?;
@@ -339,17 +383,18 @@ impl Engine {
                 return Err(error);
             }
         };
-        drop(checkpoint);
         let sender_trusted = trust.is_trusted(sender.user, sender.device, &sender.signature_key);
-        self.persist(false)?;
-        Ok(Incoming {
-            sender,
-            sender_trusted,
-            epoch,
-            kind: claimed.kind,
-            target_id: claimed.target_id.map(str::to_owned),
-            envelope,
-        })
+        Ok((
+            Incoming {
+                sender,
+                sender_trusted,
+                epoch,
+                kind: claimed.kind,
+                target_id: claimed.target_id.map(str::to_owned),
+                envelope,
+            },
+            checkpoint,
+        ))
     }
 
     /// A ciphertext under this device's leaf, at the generation its next real message will use,

crates/e2ee/tests/store.rs

@@ -177,6 +177,105 @@ fn encrypting_and_making_key_packages_fail_when_the_state_cannot_be_written() {
     assert_eq!(store.saves.load(Ordering::SeqCst), before + 1);
 }
 
+#[test]
+fn decrypt_committed_restores_ratchet_when_commit_refuses() {
+    let mut world = World::new();
+    let a = world.person("a");
+    let b = world.person("b");
+    world.sync_trust();
+    world.start(a);
+    world.change(&[a], a, &[b], &[]);
+    let conv = world.conv();
+    let sent = world.send(a, "must stay openable after a refused commit");
+    let claimed = Claimed {
+        author: Some(world.clients[a].user()),
+        device: Some(world.clients[a].device()),
+        kind: Kind::Message,
+        target_id: None,
+    };
+    let err = world.clients[b]
+        .engine
+        .decrypt_committed(conv, &sent, &claimed, |_| Err(Error::Storage("cache full".into())))
+        .expect_err("refused commit must surface");
+    assert!(matches!(err, Error::Storage(_)));
+    assert_eq!(
+        world.clients[b]
+            .engine
+            .decrypt(conv, &sent, &claimed)
+            .unwrap()
+            .envelope
+            .text,
+        "must stay openable after a refused commit"
+    );
+}
+
+#[test]
+fn decrypt_committed_keeps_plaintext_when_state_save_lags() {
+    let store = Shared::default();
+    let key = StateKey::generate().unwrap();
+    let mut world = World::new();
+    let a = world.person("a");
+    let mut bob =
+        Engine::create(UserId::random().unwrap(), Box::new(store.clone()), key.clone(), "s")
+            .unwrap();
+    world.sync_trust();
+    let alice_user = world.clients[a].user();
+    bob.verify_devices(alice_user, &world.directory[&alice_user].clone())
+        .unwrap();
+    world.start(a);
+    let conv = world.conv();
+    let bob_record = e2ee::identity::DeviceRecord {
+        device_id: bob.device_id(),
+        signature_key: bob.signature_key(),
+        cross_signature: None,
+        created_at: 1,
+    };
+    world.clients[a]
+        .engine
+        .verify_devices(bob.user_id(), &[bob_record])
+        .unwrap();
+    let package = bob.key_packages(1).unwrap().remove(0);
+    world.clients[a]
+        .engine
+        .handle_proposals(conv, &world.ds.proposals(&[&package], &[], 0))
+        .unwrap();
+    let bundle = world.clients[a]
+        .engine
+        .commit_pending(conv)
+        .unwrap()
+        .unwrap();
+    world.clients[a].engine.confirm_commit(conv).unwrap();
+    bob.handle_welcome(conv, &bundle.welcomes[0].data, Some(&world.ds.public()))
+        .unwrap();
+    let sent = world.send(a, "opened while state lags");
+    let claimed = Claimed {
+        author: Some(alice_user),
+        device: None,
+        kind: Kind::Message,
+        target_id: None,
+    };
+    let before = store.load("s").unwrap().unwrap();
+    store.fail.store(true, Ordering::SeqCst);
+    let mut committed = false;
+    let opened = bob
+        .decrypt_committed(conv, &sent, &claimed, |_| {
+            committed = true;
+            Ok(())
+        })
+        .expect("plaintext remains available when only the state save fails");
+    assert!(committed);
+    assert_eq!(opened.envelope.text, "opened while state lags");
+    assert_eq!(store.load("s").unwrap().unwrap(), before);
+    assert!(bob.flush().is_err(), "the failed state save must remain visible");
+    store.fail.store(false, Ordering::SeqCst);
+    // An accepted receive must stay consumed in memory even while its state save lags.
+    assert!(matches!(bob.decrypt(conv, &sent, &claimed), Err(Error::Undecryptable)),
+        "state-save failure must not rewind an accepted receive");
+    bob.flush().unwrap();
+    let mut reopened = Engine::open(Box::new(store.clone()), key, "s").unwrap().unwrap();
+    assert!(matches!(reopened.decrypt(conv, &sent, &claimed), Err(Error::Undecryptable)));
+}
+
 #[test]
 fn receiving_survives_a_failed_write_and_reports_it_on_flush() {
     let store = Shared::default();