MAM archive storage fault must not drop live message delivery

Monal-iOS reconnect-looped with connection-closed reason=
protocol-or-transport-error on every message. The cause was a damaged
MAM archive redo baseline ('archive redo baseline changed in place'): the
native archive at data/<host>/archive/<user>.list no longer matched the
anchor recorded in its .rredo journal, so every archive_for_user() call
returned Err and the '?' propagated up through authenticated_transport,
turning a storage fault into a dropped session.

Delivery is an attribute of MAM, not a prerequisite. Make archival
best-effort at the three inbound/outbound message sites (sender,
recipient, incoming federation): on storage failure the original stanza
is still routed, both sessions stay alive, and a sanitized
'mam-archive-skipped' log is emitted instead of a connection-closed.

Offline storage-checkpoint now repairs the damaged journal: prepare_rewrite
detects an untrusted baseline (changed in place, shorter, missing, or
journal overflows bound), treats the durably synced native as authoritative
and retires the untrusted journal, so MAM history reads and future appends
work again after handoff. Live appends remain fail-closed (the journal is
preserved, never silently dropped).

Tests: redo_rewrite_tests::runtime_append_fails_closed... and
offline_checkpoint_retires_untrusted_redo_baseline; new network regression
mam_broken_baseline_network.py proves message delivery and both sessions
survive a broken baseline and that storage-checkpoint restores MAM reads.

.gitignore

@@ -2,3 +2,4 @@
 *.local.*
 /demo/
 /bin/
+**pycache**

src/archive.rs

@@ -465,10 +465,7 @@ pub fn checkpoint_all(root: &Path) -> Result<usize> {
 /// Retire redo before an atomic snapshot replacement. The old native snapshot
 /// is synced first, so a crash after unlink and before rename retains every
 /// committed record. After rename there is no old redo to resurrect deletions.
-fn prepare_rewrite(p: &Path) -> Result<()> {
-    if recover_redo(p)?.is_none() {
-        return Ok(());
-    }
+fn retire_redo(p: &Path) -> Result<()> {
     std::fs::File::open(p)?.sync_all()?;
     std::fs::remove_file(p.with_extension("rredo"))?;
     std::fs::File::open(p.parent().unwrap())?.sync_all()?;
@@ -479,6 +476,34 @@ fn prepare_rewrite(p: &Path) -> Result<()> {
         .remove(p);
     Ok(())
 }
+/// True when the redo journal no longer matches its native baseline after an
+/// external in-place write or replacement. In that state the journal is no
+/// longer authoritative and must not be replayed over the native file.
+fn is_untrusted_redo(error: &anyhow::Error) -> bool {
+    let text = format!("{error:#}");
+    text.contains("archive redo baseline changed in place")
+        || text.contains("archive shorter than redo baseline")
+        || text.contains("missing archive redo baseline")
+        || text.contains("archive redo journal exceeds bound")
+}
+/// Retire redo before an atomic snapshot replacement or with an explicit offline
+/// handoff (storage-checkpoint). A journal whose baseline was externally
+/// rewritten in place can no longer be trusted: because the caller is about to
+/// replace the native file (rewrite) or to hand control of it to a native
+/// writer (checkpoint), the current durably synced native becomes authoritative
+/// and the untrusted journal is retired rather than failing the operation.
+fn prepare_rewrite(p: &Path) -> Result<()> {
+    match recover_redo(p) {
+        Ok(Some(_)) => retire_redo(p),
+        Ok(None) => Ok(()),
+        Err(error) if is_untrusted_redo(&error) => {
+            // The journal no longer matches its baseline. Treat the native file
+            // as authoritative, sync it, and retire the untrusted journal.
+            retire_redo(p)
+        }
+        Err(error) => Err(error),
+    }
+}
 fn scan_unlocked<F: FnMut(Entry) -> Result<()>>(
     root: &Path,
     host: &str,
@@ -1012,4 +1037,120 @@ mod redo_rewrite_tests {
                 .is_empty()
         );
     }
+
+    /// Build an archive large enough to have rolled its redo journal, so the
+    /// baseline is non-empty (`base > 0`) and the anchor guard is meaningful.
+    fn rolled_fixture() -> (tempfile::TempDir, std::path::PathBuf) {
+        let root = tempfile::tempdir().unwrap();
+        // One more than REDO_FRAMES forces a redo rotation at frame 128, which
+        // snapshots the current native length as the new non-empty baseline.
+        for i in 0..=REDO_FRAMES as u64 {
+            append_new(
+                root.path(),
+                "example.org",
+                Some("alice"),
+                "archive",
+                Entry {
+                    id: format!("roll-{i}"),
+                    when: i as f64,
+                    peer: String::new(),
+                    xml: "<message><body>x</body></message>".into(),
+                },
+            )
+            .unwrap();
+        }
+        let path = storage::path(root.path(), "example.org", Some("alice"), "archive")
+            .with_extension("list");
+        // Recovery must see a non-empty baseline (rotation happened).
+        let data = std::fs::read(path.with_extension("rredo")).unwrap();
+        let base = u64::from_le_bytes(data[8..16].try_into().unwrap());
+        assert!(base > 0, "fixture should have rolled a non-empty baseline");
+        (root, path)
+    }
+
+    /// An external writer that rewrote the native archive in place over the
+    /// committed baseline without a handoff leaves the redo journal untrusted.
+    fn corrupt_baseline(path: &std::path::Path) -> u64 {
+        let data = std::fs::read(path.with_extension("rredo")).unwrap();
+        let base = u64::from_le_bytes(data[8..16].try_into().unwrap());
+        let mut native = std::fs::read(path).unwrap();
+        // Flip the trailing bytes of the committed baseline so its anchor changes.
+        let start = (base.saturating_sub(64)) as usize;
+        for i in start..native.len().min(base as usize) {
+            native[i] ^= 0x5a;
+        }
+        let mut file = std::fs::OpenOptions::new()
+            .write(true)
+            .truncate(false)
+            .open(path)
+            .unwrap();
+        std::io::Write::write_all(&mut file, &native).unwrap();
+        file.sync_all().unwrap();
+        base
+    }
+
+    #[test]
+    fn runtime_append_fails_closed_on_untrusted_redo_baseline() {
+        let (root, path) = rolled_fixture();
+        corrupt_baseline(&path);
+        // A live append must fail closed (not silently overwrite the native
+        // file): the journal no longer matches its baseline.
+        let err = append_new(
+            root.path(),
+            "example.org",
+            Some("alice"),
+            "archive",
+            Entry {
+                id: "post-corruption".into(),
+                when: 100000.,
+                peer: String::new(),
+                xml: "<message><body>new</body></message>".into(),
+            },
+        )
+        .expect_err("live append must fail on an untrusted redo baseline");
+        assert!(
+            is_untrusted_redo(&err),
+            "expected an untrusted-baseline error, got: {err:#}"
+        );
+        // The journal is preserved for explicit offline handoff.
+        assert!(path.with_extension("rredo").exists());
+    }
+
+    #[test]
+    fn offline_checkpoint_retires_untrusted_redo_baseline() {
+        let (root, path) = rolled_fixture();
+        let base = corrupt_baseline(&path);
+        assert!(base > 0);
+        assert!(path.with_extension("rredo").exists());
+        // An external rewrite removed the trusted baseline; an explicit offline
+        // handoff accepts the current synced native as authoritative and retires
+        // the untrusted journal so the archive becomes writable again.
+        let retired = checkpoint_all(root.path()).unwrap();
+        assert_eq!(retired, 1);
+        assert!(
+            !path.with_extension("rredo").exists(),
+            "offline checkpoint must retire the untrusted journal"
+        );
+        // The archive accepts new appends: with the journal retired the current
+        // (now authoritative) native length becomes the fresh baseline.
+        let id = append_new(
+            root.path(),
+            "example.org",
+            Some("alice"),
+            "archive",
+            Entry {
+                id: "after-handoff".into(),
+                when: 200000.,
+                peer: String::new(),
+                xml: "<message><body>new</body></message>".into(),
+            },
+        )
+        .unwrap();
+        assert_eq!(id, "after-handoff");
+        // The original baseline tail was externally rewritten, so the native
+        // holds whatever the external writer put there; checkpoint only retires
+        // the journal (handoff), it does not try to re-encode damaged records.
+        // A fresh valid journal is (re)created on the next append.
+        assert!(path.with_extension("rredo").exists());
+    }
 }

src/server.rs

@@ -2887,6 +2887,43 @@ fn archive_policy(rt: &Runtime) -> &str {
         _ => "always",
     }
 }
+/// Sanitized, payload-free label for a session fault: values come from the
+/// error chain only and never contain stanza payload, tokens or passwords.
+fn sanitized_label(error: &anyhow::Error) -> String {
+    error
+        .chain()
+        .map(|x| x.to_string())
+        .find(|s| s.chars().all(|c| c.is_ascii() && !c.is_control()))
+        .unwrap_or_default()
+        .chars()
+        .take(160)
+        .collect()
+}
+/// Best-effort MAM archival so a storage fault (for example an untrusted redo
+/// baseline after an external write to the data directory) cannot take down
+/// message delivery. History is an attribute of delivery, not a prerequisite:
+/// on failure the original stanza is returned so the session stays up and the
+/// message is still routed. The fault is logged without payload.
+fn archive_message_best_effort(
+    rt: &Runtime,
+    ctx: &mut ConnectionLog,
+    owner: &str,
+    peer: &str,
+    xml: &str,
+    time: f64,
+) -> String {
+    match crate::mam::archive_for_user(&rt.data, &rt.domain, owner, peer, xml, time) {
+        Ok((_, stamped)) => stamped,
+        Err(error) => {
+            ctx.event(
+                Level::Warn,
+                "mam-archive-skipped",
+                &[("err", &sanitized_label(&error))],
+            );
+            xml.to_string()
+        }
+    }
+}
 fn routed(n: roxmltree::Node<'_, '_>, from: &str, to: &str) -> String {
     protocol::address(n, from, to)
 }
@@ -3647,15 +3684,14 @@ async fn authenticated_transport<S: AsyncRead + AsyncWrite + Unpin>(
                         boolean(&rt.config, &rt.domain, "mam_smart_enable", false),
                     )?
                 {
-                    let (_, stamped) = crate::mam::archive_for_user(
-                        &rt.data,
-                        &rt.domain,
+                    sent_xml = archive_message_best_effort(
+                        &rt,
+                        ctx,
                         &user,
                         protocol::bare(to),
                         &xml,
                         now,
-                    )?;
-                    sent_xml = stamped;
+                    );
                 } else {
                     sent_xml = xml.clone();
                 }
@@ -3677,10 +3713,7 @@ async fn authenticated_transport<S: AsyncRead + AsyncWrite + Unpin>(
                             boolean(&rt.config, &rt.domain, "mam_smart_enable", false),
                         )?
                     {
-                        let (_, stamped) = crate::mam::archive_for_user(
-                            &rt.data, &rt.domain, u, &owner, &xml, now,
-                        )?;
-                        xml = stamped;
+                        xml = archive_message_best_effort(&rt, ctx, u, &owner, &xml, now);
                     }
                     message_phases[2] = recipient_mam_started.elapsed();
                 }
@@ -4053,15 +4086,29 @@ async fn incoming_federated(rt: Runtime, stanza: crate::federation::FederatedSta
                 boolean(&rt.config, &rt.domain, "mam_smart_enable", false),
             )?
         {
-            delivery_xml = crate::mam::archive_for_user(
+            let time = chrono::Utc::now().timestamp_millis() as f64 / 1000.0;
+            delivery_xml = match crate::mam::archive_for_user(
                 &rt.data,
                 &rt.domain,
                 &user,
                 protocol::bare(from),
                 &delivery_xml,
-                chrono::Utc::now().timestamp_millis() as f64 / 1000.0,
-            )?
-            .1;
+                time,
+            ) {
+                Ok((_, stamped)) => stamped,
+                Err(error) => {
+                    // A storage fault must not drop an inbound federated message:
+                    // deliver it unarchived so it still reaches the recipient.
+                    rt.logger.event(
+                        Level::Warn,
+                        "s2s",
+                        None,
+                        "mam-archive-skipped",
+                        &[("err", &sanitized_label(&error))],
+                    );
+                    delivery_xml
+                }
+            };
         }
         let targets = rt.hub.message_targets_for_type(to, typ);
         let mut delivered = 0;

tests/mam_broken_baseline_network.py

@@ -0,0 +1,338 @@
+#!/usr/bin/env python3
+"""A damaged MAM redo baseline must not take down the live message session.
+
+Usage: python3 tests/mam_broken_baseline_network.py target/release/rust-xmpp
+
+Reproduces the production symptom reported with Monal: the native archive at
+`data/<host>/archive/<user>.list` no longer matches the anchor recorded in its
+`.rredo` journal ("archive redo baseline changed in place"). On every message the
+previous build drained the session with `connection-closed reason=
+protocol-or-transport-error`, so the client reconnect-looped and its message was
+never delivered.
+
+The archive store is a durable attribute of delivery, not a prerequisite. This
+test seeds an archive for `load-0`, corrupts its redo baseline offline to force
+exactly that recovery error, then verifies that:
+  * a fresh message is still routed load-0 -> load-1,
+  * BOTH sessions survive (ping round-trip), no `connection-closed`,
+  * the runtime stays fail-closed (the untrusted journal is NOT silently dropped),
+  * the offline `storage-checkpoint` repair retires the untrusted journal and
+    restores MAM history reads on the now-authoritative native file.
+"""
+
+import argparse
+import hashlib
+import json
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import xml.etree.ElementTree as ET
+from pathlib import Path
+
+from _wire_client import Client, local
+
+REDO_MAGIC = b"XARRED01"
+REDO_HEADER = 96
+BODY = "must-survive-broken-baseline"
+
+
+def start(binary, config, port, log):
+    server = subprocess.Popen(
+        [binary, "serve", str(config), "--experimental"], stdout=log, stderr=log
+    )
+    deadline = time.monotonic() + 60
+    while time.monotonic() < deadline:
+        assert server.poll() is None, "server stopped during startup"
+        try:
+            with socket.create_connection(("127.0.0.1", port), timeout=0.1):
+                return server
+        except OSError:
+            time.sleep(0.05)
+    server.kill()
+    server.wait()
+    raise AssertionError("server did not start")
+
+
+def stop(server):
+    server.terminate()
+    try:
+        code = server.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        server.kill()
+        server.wait()
+        raise AssertionError("SIGTERM did not complete orderly shutdown")
+    assert code == 0, f"SIGTERM shutdown exited {code}"
+
+
+def corrupt_baseline(path):
+    """Force the `archive redo baseline changed in place` recovery error.
+
+    Rebuilds a rotated-style journal header leaving the native file untouched:
+    the header records the native length as a non-empty baseline but stores a
+    wrong anchor. `recover_redo` then rejects it exactly as when an external
+    writer rewrote the native file in place after the journal rotated.
+    """
+    native = path.read_bytes()
+    length = len(native)
+    assert length > 0, "seed produced an empty native MAM archive"
+    header = bytearray(REDO_HEADER)
+    header[:8] = REDO_MAGIC
+    header[8:16] = length.to_bytes(8, "little")
+    header[16:32] = b"\0" * 16  # dev/ino are only a cache identity, not validated
+    tail = native[-min(length, 4096) :]
+    real_anchor = hashlib.sha256(tail).digest()
+    wrong = bytearray(real_anchor)
+    wrong[0] ^= 0xFF
+    header[32:64] = wrong
+    header[64:96] = hashlib.sha256(header[:64]).digest()
+    path.with_suffix(".rredo").write_bytes(header)
+    assert hashlib.sha256(header[:64]).digest() == header[64:96], (
+        "rebuilt header checksum"
+    )
+    assert header[32:64] != real_anchor, "anchor did not diverge from native"
+
+
+def run(binary):
+    with tempfile.TemporaryDirectory(prefix="xmpp-mam-broken-") as tmp:
+        root = Path(tmp)
+        subprocess.run(
+            [
+                "openssl",
+                "req",
+                "-x509",
+                "-newkey",
+                "rsa:2048",
+                "-nodes",
+                "-days",
+                "1",
+                "-keyout",
+                str(root / "key.pem"),
+                "-out",
+                str(root / "cert.pem"),
+                "-subj",
+                "/CN=localhost",
+                "-addext",
+                "subjectAltName=DNS:localhost",
+            ],
+            check=True,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        with socket.socket() as probe:
+            probe.bind(("127.0.0.1", 0))
+            port = probe.getsockname()[1]
+        config = root / "config.json"
+        config.write_text(
+            json.dumps(
+                {
+                    "global": {
+                        "data_path": str(root / "data"),
+                        "interfaces": ["127.0.0.1"],
+                        "c2s_ports": [port],
+                        "s2s_ports": [],
+                        "http_ports": [],
+                        "https_ports": [],
+                        "modules_enabled": [
+                            "tls",
+                            "saslauth",
+                            "smacks",
+                            "ping",
+                            "mam",
+                            "carbons",
+                        ],
+                        "smacks_hibernation_time": 600,
+                        "archive_expires_after": "never",
+                        "default_archive_policy": True,
+                        "ssl": {
+                            "key": str(root / "key.pem"),
+                            "certificate": str(root / "cert.pem"),
+                        },
+                    },
+                    "hosts": {
+                        "localhost": {
+                            "options": {
+                                "authentication": "internal_hashed",
+                                "password_hash": "SHA-256",
+                            }
+                        }
+                    },
+                }
+            )
+        )
+        for number in range(2):
+            subprocess.run(
+                [
+                    binary,
+                    "create-user",
+                    str(config),
+                    f"load-{number}",
+                    "--password-stdin",
+                ],
+                input="load-test-password\n",
+                text=True,
+                check=True,
+                capture_output=True,
+            )
+
+        with (root / "server.log").open("w+") as log:
+            server = None
+            try:
+                # Seed: leave a real, valid MAM entry for load-0 so its archive
+                # carries a genuine .list + .rredo pair with a non-zero baseline.
+                server = start(binary, config, port, log)
+                alice = Client(
+                    port, root / "cert.pem", "load-0", "load-test-password", "desktop"
+                )
+                bob = Client(
+                    port, root / "cert.pem", "load-1", "load-test-password", "desktop"
+                )
+                alice.available()
+                bob.available()
+                alice.send(
+                    f"<message to='{bob.jid}' type='chat' id='seed'><body>seed</body></message>"
+                )
+                bob.body("seed")
+                alice.sync()
+                stop(server)
+                server = None
+                for client in (alice, bob):
+                    client.close()
+            finally:
+                if server is not None and server.poll() is None:
+                    server.kill()
+                    server.wait()
+
+        # The sender localpart is percent-encoded on disk (`load-0` -> `load%2d0`).
+        sender_archive = root / "data" / "localhost" / "archive" / "load%2d0.list"
+        assert sender_archive.exists(), "seed did not create load-0 MAM archive"
+        assert sender_archive.with_suffix(".rredo").exists(), (
+            "seed did not attach load-0 redo journal"
+        )
+        corrupt_baseline(sender_archive)
+        damaged_journal = sender_archive.with_suffix(".rredo").read_bytes()
+
+        with (root / "server.log").open("a") as log:
+            server = None
+            try:
+                server = start(binary, config, port, log)
+                alice = Client(
+                    port, root / "cert.pem", "load-0", "load-test-password", "desktop"
+                )
+                bob = Client(
+                    port, root / "cert.pem", "load-1", "load-test-password", "desktop"
+                )
+                alice.available()
+                bob.available()
+                # A damaged MAM baseline must not drop the dispatch: deliver and
+                # keep both sessions alive.
+                alice.send(
+                    f"<message to='{bob.jid}' type='chat' id='live'><body>{BODY}</body></message>"
+                )
+                bob.body(BODY)
+                # Ping round-trips prove neither session was closed by the archive fault.
+                alice.sync()
+                bob.sync()
+                alice.send(
+                    f"<message to='{bob.jid}' type='chat' id='second'><body>second</body></message>"
+                )
+                bob.body("second")
+                alice.sync()
+                bob.sync()
+            finally:
+                if server is not None and server.poll() is None:
+                    server.kill()
+                    server.wait()
+
+        log_text = (root / "server.log").read_text()
+        skipped = [l for l in log_text.splitlines() if "mam-archive-skipped" in l]
+        assert skipped, (
+            "server did not log the best-effort archive skip (isolation inactive?)"
+        )
+        assert not any(
+            "connection-closed" in l and "protocol-or-transport-error" in l
+            for l in log_text.splitlines()
+        ), "a session was closed by the archive fault"
+        assert not any("session-error-label" in l for l in log_text.splitlines()), (
+            "unhandled archive fault labelled a session error"
+        )
+
+        # Runtime must stay fail-closed: the untrusted journal is preserved.
+        assert sender_archive.with_suffix(".rredo").read_bytes() == damaged_journal, (
+            "runtime dropped the untrusted redo journal instead of failing closed"
+        )
+
+        # Offline handoff must repair: checkpoint retires the untrusted journal
+        # so the now-authoritative native file becomes a clean one-journal base.
+        result = subprocess.run(
+            [binary, "storage-checkpoint", str(config)],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        assert "Durable native archive handoff:" in result.stdout
+        assert not sender_archive.with_suffix(".rredo").exists(), (
+            "storage-checkpoint did not retire the untrusted redo journal"
+        )
+
+        # After repair MAM history is readable again (endpoint reads now succeed).
+        phase3_offset = len((root / "server.log").read_text().splitlines())
+        with (root / "server.log").open("a") as log:
+            server = None
+            try:
+                server = start(binary, config, port, log)
+                alice = Client(
+                    port, root / "cert.pem", "load-0", "load-test-password", "desktop"
+                )
+                bob = Client(
+                    port, root / "cert.pem", "load-1", "load-test-password", "desktop"
+                )
+                alice.available()
+                bob.available()
+                # A MAM query to load-0's own archive must serve history.
+                # Query the bare account JID (not the full resource).
+                alice.send(
+                    f"<iq type='set' id='mam' to='load-0@localhost'>"
+                    f"<query xmlns='urn:xmpp:mam:2'/>"
+                    f"<x xmlns='jabber:x:data' type='submit'><field var='FORM_TYPE'>"
+                    f"<value>urn:xmpp:mam:2</value></field></x></iq>"
+                )
+                result_iq = alice.iq("mam")
+                assert (
+                    result_iq.find(
+                        ".//{urn:ietf:params:xml:ns:xmpp-stanzas}internal-server-error"
+                    )
+                    is None
+                ), ET.tostring(result_iq)
+                # After reconnect the sender archive is fresh (journal was retired);
+                # a new message archives cleanly.
+                alice.send(
+                    f"<message to='{bob.jid}' type='chat' id='post-repair'><body>post-repair</body></message>"
+                )
+                bob.body("post-repair")
+                alice.sync()
+            finally:
+                if server is not None and server.poll() is None:
+                    server.kill()
+                    server.wait()
+
+        post_log = (root / "server.log").read_text().splitlines()[phase3_offset:]
+        fresh_errors = [
+            l for l in post_log if "archive redo baseline changed in place" in l
+        ]
+        assert not fresh_errors, (
+            "archive fault returned after offline repair:\n" + "\n".join(fresh_errors)
+        )
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--binary", default="target/release/rust-xmpp")
+    args = parser.parse_args()
+    run(args.binary)
+    print("PASS: damaged MAM redo baseline does not drop live message delivery")
+
+
+if __name__ == "__main__":
+    main()