- Уменьшено число последовательных синхронизаций до доставки: ordinary

MAM
  append использует один sync вместо четырёх, SM checkpoint — один
  вместо двух.
  Checksummed redo journals сохраняют commit-before-ACK и
  восстанавливают
  native archive/SM JSON exports после аварии.
- Ограничены рост журналов, ротация и tombstone GC. Старый `.rappend` и
  legacy
  JSON импортируются. Retain/save/purge не возвращают удалённые записи.
- Recovery синхронизирует валидный journal перед публикацией состояния;
  проверка
  checksum не заменяет durability barrier. EIO при adoption блокирует
  startup
  или offline handoff. Новые fault fixtures проверяют сохранность
  журнала и retry.
- Replay SM проверяет checksum/JSON всех frames, но создаёт и валидирует
  полный
  Record только для последнего snapshot, сокращая работу с superseded
  queues.
- `message-processing-slow` содержит времена MAM отправителя/получателя,
  доставки, carbons, offline/push и handled, а также число
  ресурсов/копий.
  Payload и секреты не добавляются в логи.
- Добавлена offline команда `storage-checkpoint CONFIG`: durable экспорт
  и
  передача archive writer/подготовка отката на legacy Rust. Не
  запускайте другого
  writer и не удаляйте journals до успешного завершения команды.
- Новые crash, partial-write, corruption, EIO, recovery, handoff и
  latency
  регрессии включены в GitHub Actions через verify-local.sh; отчёты
  сохраняются.

Измерения, ограничения и эксплуатация:
[docs/DURABLE-MESSAGE-IO.md](docs/DURABLE-MESSAGE-IO.md),
[docs/VALIDATION-0.8.3.md](docs/VALIDATION-0.8.3.md).

tests/archive_redo.rs

@@ -0,0 +1,347 @@
+//! Crash fixtures model independent loss of unsynced native writes. The redo
+//! sidecar is the durable commit source until a native checkpoint or handoff.
+use rust_xmpp::{archive, storage};
+use std::{
+    fs,
+    io::Write,
+    path::{Path, PathBuf},
+    process::Command,
+    time::{Duration, Instant},
+};
+
+fn path(root: &Path) -> PathBuf {
+    storage::path(root, "example.org", Some("alice"), "archive").with_extension("list")
+}
+fn entry(id: &str) -> archive::Entry {
+    archive::Entry {
+        id: id.into(),
+        when: 42.,
+        peer: "bob@example.org".into(),
+        xml: "<message xmlns='jabber:client'><body>hello</body></message>".into(),
+    }
+}
+fn append(root: &Path, id: &str) {
+    archive::append_new(root, "example.org", Some("alice"), "archive", entry(id)).unwrap();
+}
+fn ids(root: &Path) -> Vec<String> {
+    archive::load(root, "example.org", "alice")
+        .unwrap()
+        .into_iter()
+        .map(|e| e.id)
+        .collect()
+}
+
+#[test]
+fn committed_redo_recovers_missing_and_partial_native_without_duplicates() {
+    let t = tempfile::tempdir().unwrap();
+    append(t.path(), "one");
+    append(t.path(), "two");
+    let p = path(t.path());
+    let expected = fs::read(&p).unwrap();
+    for length in [0, 1, expected.len() / 2, expected.len() - 1] {
+        fs::OpenOptions::new()
+            .write(true)
+            .open(&p)
+            .unwrap()
+            .set_len(length as u64)
+            .unwrap();
+        assert_eq!(ids(t.path()), ["one", "two"]);
+        assert_eq!(fs::read(&p).unwrap(), expected);
+        assert_eq!(ids(t.path()), ["one", "two"]);
+    }
+    fs::remove_file(&p).unwrap();
+    assert_eq!(ids(t.path()), ["one", "two"]);
+    assert_eq!(fs::read(&p).unwrap(), expected);
+}
+
+#[test]
+fn incomplete_final_frame_and_uncommitted_native_suffix_are_discarded() {
+    for extra in [0, 1, 7, 20, 39, 40, 71, 72, 90] {
+        let t = tempfile::tempdir().unwrap();
+        append(t.path(), "one");
+        let p = path(t.path());
+        let committed = fs::read(&p).unwrap();
+        let journal = p.with_extension("rredo");
+        let committed_journal = fs::metadata(&journal).unwrap().len();
+        append(t.path(), "two");
+        // Construct a crash before the second frame was fully persisted.
+        fs::OpenOptions::new()
+            .write(true)
+            .open(&journal)
+            .unwrap()
+            .set_len(committed_journal + extra)
+            .unwrap();
+        assert_eq!(ids(t.path()), ["one"]);
+        assert_eq!(fs::read(&p).unwrap(), committed);
+        assert_eq!(fs::metadata(&journal).unwrap().len(), committed_journal);
+        append(t.path(), "three");
+        assert_eq!(ids(t.path()), ["one", "three"]);
+    }
+}
+
+#[test]
+fn complete_checksum_corruption_fails_without_modifying_native_or_journal() {
+    for offset in [8, 64, 96, 104, 136, 168] {
+        let t = tempfile::tempdir().unwrap();
+        append(t.path(), "one");
+        let p = path(t.path());
+        let native = fs::read(&p).unwrap();
+        let journal = p.with_extension("rredo");
+        let mut corrupt = fs::read(&journal).unwrap();
+        corrupt[offset] ^= 1;
+        fs::write(&journal, &corrupt).unwrap();
+        assert!(archive::load(t.path(), "example.org", "alice").is_err());
+        assert_eq!(fs::read(&p).unwrap(), native);
+        assert_eq!(fs::read(&journal).unwrap(), corrupt);
+    }
+}
+
+#[test]
+fn bounded_rotation_recovers_suffix_and_rewrites_never_resurrect_deleted_records() {
+    let t = tempfile::tempdir().unwrap();
+    for i in 0..300 {
+        append(t.path(), &format!("id{i}"));
+    }
+    let p = path(t.path());
+    let journal = p.with_extension("rredo");
+    let data = fs::read(&journal).unwrap();
+    assert!(data.len() < 1024 * 1024);
+    let baseline = u64::from_le_bytes(data[8..16].try_into().unwrap());
+    assert!(baseline > 0);
+    fs::OpenOptions::new()
+        .write(true)
+        .open(&p)
+        .unwrap()
+        .set_len(baseline)
+        .unwrap();
+    assert_eq!(ids(t.path()).len(), 300);
+    archive::delete_ids(
+        t.path(),
+        "example.org",
+        Some("alice"),
+        "archive",
+        &["id299".into()],
+    )
+    .unwrap();
+    assert_eq!(ids(t.path()).len(), 299);
+    append(t.path(), "after-delete");
+    assert!(!ids(t.path()).contains(&"id299".into()));
+    archive::save_store(
+        t.path(),
+        "example.org",
+        Some("alice"),
+        "archive",
+        &[entry("replacement")],
+    )
+    .unwrap();
+    assert_eq!(ids(t.path()), ["replacement"]);
+    append(t.path(), "after-save");
+    assert_eq!(ids(t.path()), ["replacement", "after-save"]);
+    archive::expire_before(t.path(), "example.org", Some("alice"), "archive", 43.).unwrap();
+    assert!(ids(t.path()).is_empty());
+    append(t.path(), "after-expire");
+    assert_eq!(ids(t.path()), ["after-expire"]);
+}
+
+#[test]
+fn byte_limit_rotates_before_frame_count_limit() {
+    let t = tempfile::tempdir().unwrap();
+    for i in 0..20 {
+        let mut e = entry(&format!("big{i}"));
+        e.xml = format!("<message><body>{}</body></message>", "x".repeat(100_000));
+        archive::append_new(t.path(), "example.org", Some("alice"), "archive", e).unwrap();
+    }
+    let data = fs::read(path(t.path()).with_extension("rredo")).unwrap();
+    assert!(u64::from_le_bytes(data[8..16].try_into().unwrap()) > 0);
+    assert!(data.len() < 1024 * 1024 + 110_000);
+    assert_eq!(ids(t.path()).len(), 20);
+}
+
+#[test]
+fn explicit_native_atomic_handoff_and_missing_baseline_fail_closed() {
+    let t = tempfile::tempdir().unwrap();
+    append(t.path(), "old");
+    let p = path(t.path());
+    archive::checkpoint_all(t.path()).unwrap();
+    storage::atomic_write(
+        &p,
+        b"item({name='message';attr={};key='native';when=42});\n",
+    )
+    .unwrap();
+    assert_eq!(ids(t.path()), ["native"]);
+    append(t.path(), "rust");
+    assert_eq!(ids(t.path()), ["native", "rust"]);
+    fs::OpenOptions::new()
+        .append(true)
+        .open(&p)
+        .unwrap()
+        .write_all(b"item(bad uncommitted);")
+        .unwrap();
+    assert_eq!(ids(t.path()), ["native", "rust"]);
+    // The durable baseline cannot be reconstructed from suffix-only frames.
+    fs::remove_file(&p).unwrap();
+    assert!(archive::load(t.path(), "example.org", "alice").is_err());
+}
+
+#[test]
+fn native_baseline_in_place_edits_are_rejected() {
+    let t = tempfile::tempdir().unwrap();
+    let p = path(t.path());
+    fs::create_dir_all(p.parent().unwrap()).unwrap();
+    let original = b"item({name='message';attr={};key='native';when=42});\n";
+    fs::write(&p, original).unwrap();
+    append(t.path(), "rust");
+    let mut data = fs::read(&p).unwrap();
+    data[5] ^= 1;
+    fs::write(&p, &data).unwrap();
+    assert!(archive::load(t.path(), "example.org", "alice").is_err());
+    assert_eq!(fs::read(&p).unwrap(), data);
+}
+
+#[test]
+fn redo_process_child() {
+    let Some(root) = std::env::var_os("XMPP_ARCHIVE_REDO_CHILD") else {
+        return;
+    };
+    let root = Path::new(&root);
+    append(root, "accepted-before-kill");
+    fs::write(root.join("ready"), b"ready").unwrap();
+    loop {
+        std::thread::park();
+    }
+}
+
+#[test]
+fn killed_writer_retains_accepted_records_after_native_data_loss() {
+    let t = tempfile::tempdir().unwrap();
+    let mut child = Command::new(std::env::current_exe().unwrap())
+        .args(["--exact", "redo_process_child", "--nocapture"])
+        .env("XMPP_ARCHIVE_REDO_CHILD", t.path())
+        .spawn()
+        .unwrap();
+    let deadline = Instant::now() + Duration::from_secs(10);
+    while !t.path().join("ready").exists() {
+        if Instant::now() >= deadline {
+            let _ = child.kill();
+            let _ = child.wait();
+            panic!("redo child not ready");
+        }
+        assert!(
+            child.try_wait().unwrap().is_none(),
+            "redo child exited early"
+        );
+        std::thread::sleep(Duration::from_millis(10));
+    }
+    child.kill().unwrap();
+    child.wait().unwrap();
+    // Simulate loss of dirty native pages independently of the synced journal.
+    fs::OpenOptions::new()
+        .write(true)
+        .open(path(t.path()))
+        .unwrap()
+        .set_len(0)
+        .unwrap();
+    assert_eq!(ids(t.path()), ["accepted-before-kill"]);
+    assert_eq!(ids(t.path()), ["accepted-before-kill"]);
+}
+
+#[test]
+fn durable_handoff_allows_native_in_place_append_and_next_rust_append() {
+    let t = tempfile::tempdir().unwrap();
+    append(t.path(), "rust-before");
+    let p = path(t.path());
+    assert_eq!(archive::checkpoint_all(t.path()).unwrap(), 1);
+    assert!(!p.with_extension("rredo").exists());
+    fs::OpenOptions::new()
+        .append(true)
+        .open(&p)
+        .unwrap()
+        .write_all(b"item({name='message';attr={};key='native';when=42});\n")
+        .unwrap();
+    assert_eq!(ids(t.path()), ["rust-before", "native"]);
+    append(t.path(), "rust-after");
+    assert_eq!(ids(t.path()), ["rust-before", "native", "rust-after"]);
+    assert_eq!(archive::checkpoint_all(t.path()).unwrap(), 1);
+    assert_eq!(archive::checkpoint_all(t.path()).unwrap(), 0);
+}
+
+#[test]
+fn legacy_rollback_marker_is_migrated_before_redo_append() {
+    let t = tempfile::tempdir().unwrap();
+    let p = path(t.path());
+    fs::create_dir_all(p.parent().unwrap()).unwrap();
+    let original = b"item({name='message';attr={};key='legacy';when=42});\n";
+    fs::write(&p, [original.as_slice(), b"item(broken"].concat()).unwrap();
+    fs::write(
+        p.with_extension("rappend"),
+        (original.len() as u64).to_le_bytes(),
+    )
+    .unwrap();
+    append(t.path(), "redo");
+    assert_eq!(ids(t.path()), ["legacy", "redo"]);
+    assert!(!p.with_extension("rappend").exists());
+    assert!(p.with_extension("rredo").exists());
+}
+
+#[test]
+fn copied_archive_replays_committed_redo_despite_new_native_inode() {
+    for native_baseline in [false, true] {
+        for lose_entire_suffix in [false, true] {
+            let source = tempfile::tempdir().unwrap();
+            let source_path = path(source.path());
+            if native_baseline {
+                fs::create_dir_all(source_path.parent().unwrap()).unwrap();
+                fs::write(
+                    &source_path,
+                    b"item({name='message';attr={};key='baseline';when=42});\n",
+                )
+                .unwrap();
+            }
+            append(source.path(), "accepted-one");
+            append(source.path(), "accepted-two");
+            let expected = fs::read(&source_path).unwrap();
+            let redo = fs::read(source_path.with_extension("rredo")).unwrap();
+            let baseline = u64::from_le_bytes(redo[8..16].try_into().unwrap());
+            let restored = tempfile::tempdir().unwrap();
+            let restored_path = path(restored.path());
+            fs::create_dir_all(restored_path.parent().unwrap()).unwrap();
+            fs::copy(&source_path, &restored_path).unwrap();
+            fs::copy(
+                source_path.with_extension("rredo"),
+                restored_path.with_extension("rredo"),
+            )
+            .unwrap();
+            let length = if lose_entire_suffix {
+                baseline
+            } else {
+                expected.len() as u64 - 1
+            };
+            fs::OpenOptions::new()
+                .write(true)
+                .open(&restored_path)
+                .unwrap()
+                .set_len(length)
+                .unwrap();
+            let recovered = ids(restored.path());
+            assert_eq!(recovered.last().unwrap(), "accepted-two");
+            assert_eq!(recovered.len(), if native_baseline { 3 } else { 2 });
+            assert_eq!(fs::read(&restored_path).unwrap(), expected);
+            append(restored.path(), "after-restore");
+            assert_eq!(ids(restored.path()).last().unwrap(), "after-restore");
+        }
+    }
+}
+
+#[test]
+fn atomic_native_replace_without_handoff_cannot_discard_committed_frames() {
+    let t = tempfile::tempdir().unwrap();
+    append(t.path(), "accepted");
+    let p = path(t.path());
+    storage::atomic_write(
+        &p,
+        b"item({name='message';attr={};key='replacement';when=42});\n",
+    )
+    .unwrap();
+    // A replacement without handoff is not authority to discard committed WAL.
+    assert_eq!(ids(t.path()), ["accepted"]);
+}

tests/fixtures/slow_sync.c

@@ -0,0 +1,74 @@
+/* Test-only LD_PRELOAD shim: delay, then perform the real durable operation.
+ * Scope is strictly the canonical fixture data root selected by the harness.
+ * No host mount or durability settings are changed. Log contains no file data.
+ * Exceptional XMPP_SYNC_FAIL_SUFFIX=.wal|.rredo injects EIO for matching
+ * fixture regular files. Without that explicit fault mode all syncs are real.
+ */
+#define _GNU_SOURCE
+#include <dlfcn.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <limits.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <time.h>
+#include <unistd.h>
+
+static long long ns(void) {
+    struct timespec t;
+    clock_gettime(CLOCK_MONOTONIC, &t);
+    return (long long)t.tv_sec * 1000000000LL + t.tv_nsec;
+}
+static int sync_call(int fd, const char *operation) {
+    int (*real_sync)(int) = dlsym(RTLD_NEXT, operation);
+    if (!real_sync) { errno = ENOSYS; return -1; }
+    const char *prefix = getenv("XMPP_SLOW_SYNC_ROOT");
+    const char *log = getenv("XMPP_SLOW_SYNC_LOG");
+    char link[64], path[PATH_MAX];
+    snprintf(link, sizeof link, "/proc/self/fd/%d", fd);
+    ssize_t length = readlink(link, path, sizeof path - 1);
+    if (!prefix || !log || length < 0) return real_sync(fd);
+    path[length] = '\0';
+    size_t size = strlen(prefix);
+    if (strncmp(path, prefix, size) || (path[size] && path[size] != '/'))
+        return real_sync(fd);
+    struct stat st;
+    if (fstat(fd, &st) || (!S_ISREG(st.st_mode) && !S_ISDIR(st.st_mode)))
+        return real_sync(fd);
+    const char *delay = getenv("XMPP_SLOW_SYNC_MS");
+    long ms = delay ? strtol(delay, NULL, 10) : 100;
+    if (ms < 0 || ms > 10000) ms = 100;
+    long long begin = ns();
+    struct timespec wait = {.tv_sec = ms / 1000, .tv_nsec = (ms % 1000) * 1000000};
+    while (nanosleep(&wait, &wait) && errno == EINTR) {}
+    const char *suffix = getenv("XMPP_SYNC_FAIL_SUFFIX");
+    int injected = 0;
+    if (suffix && (!strcmp(suffix, ".wal") || !strcmp(suffix, ".rredo")) && S_ISREG(st.st_mode)) {
+        size_t suffix_length = strlen(suffix);
+        injected = (size_t)length >= suffix_length &&
+            !strcmp(path + length - suffix_length, suffix);
+    }
+    int result;
+    if (injected) { errno = EIO; result = -1; }
+    else result = real_sync(fd);
+    int saved_errno = errno;
+    long long end = ns();
+    /* One append write per operation avoids partial concurrent log records.
+     * Paths are not logged: only a path-derived store category. */
+    const char *store = strstr(path, "/rust_smacks") ? "sm" :
+        strstr(path, "/archive") ? "mam" :
+        strstr(path, "/pep") ? "pep" : "other";
+    char row[512];
+    int count = snprintf(row, sizeof row,
+        "{\"op\":\"%s\",\"kind\":\"%s\",\"store\":\"%s\",\"begin_ns\":%lld,\"end_ns\":%lld,\"result\":%d,\"injected\":%s}\n",
+        operation, S_ISDIR(st.st_mode) ? "directory" : "file", store, begin, end, result,
+        injected ? "true" : "false");
+    int output = open(log, O_WRONLY | O_CREAT | O_APPEND | O_CLOEXEC, 0600);
+    if (output >= 0) { ssize_t ignored = write(output, row, count); (void)ignored; close(output); }
+    errno = saved_errno;
+    return result;
+}
+int fsync(int fd) { return sync_call(fd, "fsync"); }
+int fdatasync(int fd) { return sync_call(fd, "fdatasync"); }

tests/journal_recovery_sync_network.py

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+"""Fail recovery journal syncs explicitly; never publish a false durable handoff.
+
+Uses independent real TLS fixtures for .wal and .rredo so archive native inode
+identity remains genuine. SIGKILL leaves complete valid journals. The scoped
+test interposer returns EIO only for the selected suffix beneath fixture data.
+"""
+import argparse
+import asyncio
+import hashlib
+import json
+import os
+from pathlib import Path
+import shutil
+import signal
+import socket
+import ssl
+import subprocess
+import tempfile
+import time
+
+from muc_restart_network import stop
+from sm_restart_network import durable_slot
+from slow_storage_latency_network import CARBONS, MeasuredClient, check_shim_scope, iq, sync_rows
+from slow_storage_crash_network import committed_archive, start
+
+
+async def seed(root, binary, config, port, log, env):
+    context = ssl.create_default_context(cafile=str(root/'cert.pem'))
+    clients, server = [], None
+    identifier = 'recovery-sync-body'
+    try:
+        server = await asyncio.to_thread(start, binary, config, port, log, env)
+        for number in range(4):
+            client = await MeasuredClient.connect(number, port, context, 2)
+            clients.append(client)
+            client.auto_ack = False
+            await client.enable()
+            await iq(client, f'carbons-{number}', f"<enable xmlns='{CARBONS}'/>")
+            await client.stanza('<presence/>')
+            await client.fence(f'ready-{number}')
+        alice, bob, _, _ = clients
+        await alice.stanza(f"<message to='{bob.jid}' type='chat' id='{identifier}'><body>{identifier}</body></message>")
+        await alice.handled()
+        for client, kind in zip(clients[1:], ('direct', 'sent', 'received')):
+            await client.wait(lambda c=client,k=kind: c.arrivals[(identifier,k)], 'message/carbon before crash')
+        server.send_signal(signal.SIGKILL)
+        assert await asyncio.to_thread(server.wait, 10) == -signal.SIGKILL
+        records = {client.token: durable_slot(root, client.token) for client in clients}
+        assert len(records) == 4
+        for client in clients[1:]:
+            assert sum(identifier in xml for xml in records[client.token]['pending']) == 1
+        archives = {p: committed_archive(p) for p in (root/'data'/'localhost'/'archive').glob('*.list')}
+        assert len(archives) == 2
+        assert all(sum(identifier in row for row in content.splitlines()) == 1 for content in archives.values())
+        return records, archives
+    finally:
+        if server is not None and server.poll() is None:
+            await asyncio.to_thread(stop, server)
+        await asyncio.gather(*(c.close_raw() for c in clients), return_exceptions=True)
+
+
+def journal_bytes(root, suffix):
+    return {p: p.read_bytes() for p in (root/'data').rglob('*'+suffix)}
+
+
+def assert_serve_rejects(binary, config, port, log, env):
+    server = subprocess.Popen([str(binary), 'serve', str(config), '--experimental'],
+                              env=env, stdout=log, stderr=log)
+    listened = False
+    try:
+        deadline = time.monotonic()+15
+        while server.poll() is None and time.monotonic() < deadline:
+            try:
+                with socket.create_connection(('127.0.0.1', port), timeout=.05):
+                    listened = True
+                    break
+            except OSError:
+                time.sleep(.01)
+        assert not listened, 'SM recovery sync failure exposed a listener'
+        assert server.wait(timeout=5) != 0, 'SM recovery sync failure reported successful serve'
+    finally:
+        if server.poll() is None:
+            server.kill()
+            server.wait()
+
+
+def scope_fault(work, shim):
+    root = work/'fault-scope'
+    root.mkdir()
+    code = """import errno,json,os,pathlib,sys
+root,outside,suffix=pathlib.Path(sys.argv[1]),pathlib.Path(sys.argv[2]),sys.argv[3]
+statuses=[]
+for path in (root/('inside'+suffix),root/'unmatched.txt',outside):
+ with path.open('wb') as stream:
+  stream.write(b'fixture');stream.flush()
+  for sync in (os.fsync,os.fdatasync):
+   try:sync(stream.fileno());statuses.append(0)
+   except OSError as error:statuses.append(error.errno)
+assert statuses==[errno.EIO,errno.EIO,0,0,0,0],statuses
+"""
+    for suffix in ('.wal', '.rredo'):
+        log = work/('scope-'+suffix[1:]+'.jsonl')
+        env = dict(os.environ, LD_PRELOAD=str(shim), XMPP_SLOW_SYNC_ROOT=str(root),
+                   XMPP_SLOW_SYNC_LOG=str(log), XMPP_SLOW_SYNC_MS='0', XMPP_SYNC_FAIL_SUFFIX=suffix)
+        subprocess.run(['python3', '-c', code, str(root), str(work/('outside'+suffix)), suffix],
+                       env=env, check=True, capture_output=True)
+        rows = sync_rows(log)
+        assert len(rows) == 4 and sum(row['injected'] for row in rows) == 2
+
+
+def run_case(work, binary, shim, suffix):
+    root = work/suffix[1:]
+    root.mkdir()
+    subprocess.run(['openssl', 'req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:P-256',
+                    '-nodes', '-days', '1', '-subj', '/CN=localhost', '-addext', 'subjectAltName=DNS:localhost',
+                    '-keyout', str(root/'key.pem'), '-out', str(root/'cert.pem')], check=True, capture_output=True)
+    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([str(binary), 'create-user', str(config), f'load-{number}', '--password-stdin'],
+                       input='load-test-password\n', text=True, check=True, capture_output=True)
+    sync_log = root/'sync.jsonl'
+    env = dict(os.environ, LD_PRELOAD=str(shim), XMPP_SLOW_SYNC_ROOT=str(root/'data'),
+               XMPP_SLOW_SYNC_LOG=str(sync_log), XMPP_SLOW_SYNC_MS='0')
+    env.pop('XMPP_SYNC_FAIL_SUFFIX', None)
+    with (root/'server.log').open('w+') as log:
+        records, archives = asyncio.run(seed(root, binary, config, port, log, env))
+        journals = journal_bytes(root, suffix)
+        assert journals, 'network fixture did not leave requested journals'
+        fault_env = dict(env, XMPP_SYNC_FAIL_SUFFIX=suffix)
+        before = len(sync_rows(sync_log))
+        command = [str(binary), 'storage-checkpoint', str(config)]
+        result = subprocess.run(command, env=fault_env, capture_output=True, text=True)
+        assert result.returncode != 0, f'{suffix} recovery skipped journal sync and falsely reported successful handoff'
+        assert 'Durable native archive handoff:' not in result.stdout, 'failed checkpoint advertised handoff success'
+        attempts = [row for row in sync_rows(sync_log)[before:] if row.get('injected')]
+        assert attempts and all(row['result'] == -1 for row in attempts), 'recovery did not attempt faulted journal sync'
+        assert journal_bytes(root, suffix) == journals, 'failed checkpoint removed or changed authoritative journal'
+        if suffix == '.wal':
+            before_serve = len(sync_rows(sync_log))
+            assert_serve_rejects(binary, config, port, log, fault_env)
+            assert any(row.get('injected') and row['store'] == 'sm'
+                       for row in sync_rows(sync_log)[before_serve:]), 'serve failed without attempting SM recovery sync'
+            assert journal_bytes(root, suffix) == journals, 'failed serve changed authoritative SM journals'
+        subprocess.run(command, env=env, check=True, capture_output=True, text=True)
+        assert not journal_bytes(root, '.wal') and not journal_bytes(root, '.rredo'), 'successful handoff left journals'
+        for token, record in records.items():
+            paths = list((root/'data').rglob(token+'.json'))
+            assert len(paths) == 1
+            exported = json.loads(paths[0].read_text())
+            assert {k:v for k,v in exported.items() if k != 'deadline_ms'} == {
+                k:v for k,v in record.items() if k != 'deadline_ms'}, 'handoff changed acknowledged SM state'
+        assert all(path.read_text() == content for path, content in archives.items()), 'handoff lost acknowledged MAM body'
+        return {'suffix': suffix, 'status': 'pass', 'checkpoint_rejected_sync_failure': True,
+                'journals_preserved_on_failure': len(journals),
+                'serve_rejected_before_listeners': suffix == '.wal',
+                'clean_retry_preserved_sessions': len(records), 'clean_retry_preserved_archives': len(archives)}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--binary', default='target/release/rust-xmpp')
+    parser.add_argument('--output', default='/tmp/journal-recovery-sync.json')
+    parser.add_argument('--suffix', nargs='+', choices=['.wal', '.rredo'], default=['.wal', '.rredo'])
+    args = parser.parse_args()
+    report = {'status': 'running', 'cases': []}
+    try:
+        with tempfile.TemporaryDirectory(prefix='xmpp-journal-recovery-sync-') as directory:
+            work = Path(directory)
+            binary = work/'rust-xmpp'
+            shutil.copy2(Path(args.binary).resolve(), binary)
+            report['binary_sha256'] = hashlib.sha256(binary.read_bytes()).hexdigest()
+            shim = work/'slow_sync.so'
+            subprocess.run(['cc', '-shared', '-fPIC', '-O2', '-Wall', '-Wextra', '-Werror',
+                str(Path(__file__).parent/'fixtures'/'slow_sync.c'), '-o', str(shim), '-ldl'], check=True)
+            check_shim_scope(work, shim)
+            scope_fault(work, shim)
+            for suffix in args.suffix:
+                report['cases'].append(run_case(work, binary, shim, suffix))
+            report['status'] = 'pass'
+    except BaseException as error:
+        report.update(status='fail', failure=str(error))
+        raise
+    finally:
+        Path(args.output).write_text(json.dumps(report, indent=2)+'\n')
+        print(json.dumps(report, indent=2))
+
+
+if __name__ == '__main__':
+    main()

tests/muc_native.rs

@@ -25,6 +25,8 @@ fn actual_prosody_muc_mam_reads_rust_and_writes_native_archive() {
         "<message type='groupchat' to='room@rooms.example.org' id='rust-message'><body>Rust archived body</body></message>",
     );
     drop(muc);
+    // Transfer archive write ownership before native Prosody appends in place.
+    assert_eq!(rust_xmpp::archive::checkpoint_all(root.path()).unwrap(), 1);
     let bootstrap = std::fs::read_to_string(concat!(
         env!("CARGO_MANIFEST_DIR"),
         "/scripts/prosody_compat/oracle.lua"

tests/reference_prosody.rs

@@ -47,6 +47,8 @@ fn real_prosody_storage_and_auth_roundtrip() {
         for (i, id) in ["rust-one", "rust-two"].iter().enumerate() {
             archive::append_new(root,host,Some("alice"),"archive",archive::Entry{id:id.to_string(),when:1767225600.+i as f64,peer:"bob@example.org".into(),xml:"<message xmlns='jabber:client' xml:lang='ru' xmlns:o='urn:oracle' o:flag='yes'><body>Rust body</body></message>".into()}).unwrap();
         }
+        // Transfer write ownership only after durable native checkpoint.
+        archive::checkpoint_all(root).unwrap();
         oracle(root, "roundtrip", hash);
         let v = storage::load(root, host, Some("alice"), "oracle")
             .unwrap()
@@ -97,6 +99,7 @@ fn real_prosody_storage_and_auth_roundtrip() {
         )
         .unwrap();
         assert!(!p.exists());
+        archive::checkpoint_all(root).unwrap();
         oracle(root, "verify-final", hash);
     }
 }

tests/slow_storage_crash_network.py

@@ -0,0 +1,287 @@
+#!/usr/bin/env python3
+"""SIGKILL after sender SM acknowledgement under real delayed durable IO.
+
+Four STARTTLS resources/two accounts, MAM, carbons and unacknowledged avatar
+PEP. Inspect authoritative recipient journals and both archives before any
+reauthentication; then resume all resources and verify replay and counters.
+Uses the latency fixture's fsync/fdatasync interposer, retaining real syncs.
+"""
+import argparse
+import asyncio
+import hashlib
+import json
+import os
+from pathlib import Path
+import shutil
+import signal
+import socket
+import ssl
+import subprocess
+import sys
+import tempfile
+import time
+
+from muc_sm_load_network import SM
+from muc_restart_network import stop
+from sm_restart_network import durable_slot
+from slow_storage_latency_network import (
+    CARBONS, PUBSUB, MeasuredClient, check_shim_scope, counts, iq, sync_rows,
+)
+
+
+def start(binary, config, port, log, env):
+    server = subprocess.Popen([str(binary), 'serve', str(config), '--experimental'],
+                              env=env, 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=.1):
+                return server
+        except OSError:
+            time.sleep(.05)
+    server.kill()
+    server.wait()
+    raise AssertionError('server startup timeout')
+
+
+def committed_archive(path):
+    """Reconstruct native MAM bytes using its authoritative checksummed redo."""
+    native = path.read_bytes()
+    data = path.with_suffix('.rredo').read_bytes()
+    assert len(data) >= 96 and data[:8] == b'XARRED01', 'invalid MAM redo header'
+    assert hashlib.sha256(data[:64]).digest() == data[64:96], 'invalid MAM redo header checksum'
+    baseline = int.from_bytes(data[8:16], 'little')
+    assert baseline <= len(native), 'missing MAM native baseline'
+    committed, offset = native[:baseline], 96
+    while offset < len(data):
+        if len(data)-offset < 72:
+            break
+        size = int.from_bytes(data[offset:offset+8], 'little')
+        assert hashlib.sha256(data[offset:offset+8]).digest() == data[offset+8:offset+40], 'invalid MAM redo length checksum'
+        assert 0 < size <= 16*1024*1024, 'invalid MAM redo frame size'
+        end = offset+72+size
+        if end > len(data):
+            break
+        payload = data[offset+72:end]
+        assert hashlib.sha256(payload).digest() == data[offset+40:offset+72], 'invalid MAM redo frame checksum'
+        committed += payload
+        offset = end
+    # Sender handled acknowledgement must not depend on a native suffix that
+    # lacks a complete, durable redo frame.
+    assert native == committed, 'native MAM suffix is not committed in redo'
+    return committed.decode()
+
+
+def assert_committed(root, expected, identifiers, avatar):
+    snapshots = [durable_slot(root, old['token']) for old in expected]
+    for index, (record, old) in enumerate(zip(snapshots, expected)):
+        assert record['jid'] == old['jid'], (index, record)
+        assert record['handled'] == old['handled'], (index, record['handled'], old)
+        assert record['acked'] == old['acked'], (index, record['acked'], old)
+        assert record['sent']-record['acked'] == len(record['pending']), (index, record)
+        if index in (1, 2, 3):
+            for identifier in identifiers:
+                assert sum(identifier in xml for xml in record['pending']) == 1, (index, identifier, record)
+        if index in (0, 2):
+            assert any(avatar in xml and 'event' in xml and 'crash-avatar' in xml
+                       for xml in record['pending']), ('PEP missing before reauthentication', index, record)
+    paths = sorted((root/'data'/'localhost'/'archive').glob('*.list'))
+    assert len(paths) == 2, paths
+    for path in paths:
+        rows = [row for row in committed_archive(path).splitlines() if row.startswith('item(')]
+        assert len(rows) == len(identifiers), (path.name, len(rows), identifiers)
+        for identifier in identifiers:
+            assert sum(identifier in row for row in rows) == 1, ('acknowledged body missing/duplicated in MAM', path.name, identifier)
+    return snapshots
+
+
+async def exercise(root, binary, config, port, log, env, sync_log, args):
+    context = ssl.create_default_context(cafile=str(root/'cert.pem'))
+    clients, server = [], None
+    avatar = 'urn:xmpp:avatar:metadata'
+    try:
+        server = await asyncio.to_thread(start, binary, config, port, log, env)
+        for number in range(4):
+            client = await MeasuredClient.connect(number, port, context, 2)
+            clients.append(client)
+            client.auto_ack = False
+            await client.enable()
+            await iq(client, f'carbons-{number}', f"<enable xmlns='{CARBONS}'/>")
+            await client.stanza('<presence/>')
+            await client.fence(f'ready-{number}')
+        alice, bob, alice_other, _ = clients
+        await iq(alice, 'avatar-initial', f"<pubsub xmlns='{PUBSUB}'><publish node='{avatar}'><item id='initial'><metadata xmlns='{avatar}'/></item></publish></pubsub>")
+        for client in (alice, alice_other):
+            await iq(client, f'subscribe-{client.number}', f"<pubsub xmlns='{PUBSUB}'><subscribe node='{avatar}' jid='{client.jid}'/></pubsub>")
+        # Ack every setup stanza, then fence acknowledgement persistence. No
+        # resource sends another ACK until after the crash replay assertions.
+        for client in clients:
+            await client.fence(f'setup-{client.number}')
+            await client.ack()
+            await client.handled()
+        before = len(sync_rows(sync_log))
+        await iq(alice, 'avatar-pending', f"<pubsub xmlns='{PUBSUB}'><publish node='{avatar}'><item id='crash-avatar'><metadata xmlns='{avatar}'><info id='fixture-avatar' bytes='1' type='image/png'/></metadata></item></publish></pubsub>")
+        identifiers = [f'crash-body-{number}' for number in range(args.messages)]
+        for identifier in identifiers:
+            await alice.stanza(f"<message to='{bob.jid}' type='chat' id='{identifier}'><body>{identifier}</body></message>")
+            await alice.handled()
+        expected = [{'token': c.token, 'jid': c.jid, 'acked': c.acked, 'handled': c.inputs}
+                    for c in clients]
+        # No receiver fence/delivery wait between sender's final <a/> and kill:
+        # the acknowledged sender operation itself must make the data durable.
+        server.send_signal(signal.SIGKILL)
+        assert await asyncio.to_thread(server.wait, 10) == -signal.SIGKILL
+        snapshots = assert_committed(root, expected, identifiers, avatar)
+        rows = sync_rows(sync_log)[before:]
+        summary = counts(rows)
+        assert rows and all(row['end_ns']-row['begin_ns'] >= args.delay_ms*1000000 for row in rows)
+        assert {'sm', 'mam', 'pep'}.issubset({row['store'] for row in rows}), summary
+        assert any(row['kind'] == 'directory' for row in rows), summary
+        await asyncio.gather(*(c.close_raw() for c in clients), return_exceptions=True)
+        clients.clear()
+        server = await asyncio.to_thread(start, binary, config, port, log, env)
+        # Startup must recover the same queues before a single SASL login.
+        recovered = assert_committed(root, expected, identifiers, avatar)
+        for old, new in zip(snapshots, recovered):
+            assert old['pending'] == new['pending']
+        for number, old in enumerate(expected):
+            client = await MeasuredClient.connect(number, port, context, 2, bind=False)
+            clients.append(client)
+            client.auto_ack = False
+            client.received = client.acked = old['acked']
+            client.inputs, client.token = old['handled'], old['token']
+            await client.send(f"<resume xmlns='{SM}' previd='{client.token}' h='{client.acked}'/>")
+            result = await client.control('resumed')
+            assert result.get('previd') == client.token
+            assert int(result.get('h')) == old['handled'], (number, result.attrib, old)
+            kind = {1: 'direct', 2: 'sent', 3: 'received'}.get(number)
+            if kind:
+                for identifier in identifiers:
+                    await client.wait(lambda c=client,i=identifier,k=kind: c.arrivals[(i,k)], 'durable replay '+identifier)
+                    assert len(client.arrivals[(identifier,kind)]) == 1, (number, identifier, client.arrivals)
+            await client.fence(f'after-resume-{number}')
+            # The fence follows every replay, so the exact output counter also
+            # detects missing/extra PEP, carbon or IQ replay, beyond body counts.
+            assert client.received == snapshots[number]['sent']+1, (number, client.received, snapshots[number]['sent'])
+            await client.ack()
+            await client.handled()
+            record = durable_slot(root, client.token)
+            assert record['acked'] == client.received and not record['pending'], (number, record)
+            assert not client.errors, client.errors
+        # Native handoff is explicitly offline: release the server's data/store
+        # locks before asking the CLI to durably export and remove journals.
+        await asyncio.to_thread(stop, server)
+        server = None
+        final_slots = {c.token: durable_slot(root, c.token) for c in clients}
+        final_archives = {p: committed_archive(p) for p in
+                          (root/'data'/'localhost'/'archive').glob('*.list')}
+        checkpoint = [str(binary), 'storage-checkpoint', str(config)]
+        await asyncio.to_thread(subprocess.run, checkpoint, env=env, check=True, capture_output=True, text=True)
+        assert not list((root/'data').rglob('*.wal')), 'SM WAL survived offline checkpoint'
+        assert not list((root/'data').rglob('*.rredo')), 'MAM redo survived offline checkpoint'
+        exported = {}
+        for token, authoritative in final_slots.items():
+            paths = list((root/'data').rglob(token+'.json'))
+            assert len(paths) == 1, ('checkpoint lost SM export', token)
+            exported[token] = json.loads(paths[0].read_text())
+            previous = dict(authoritative)
+            current = dict(exported[token])
+            previous.pop('deadline_ms')
+            current.pop('deadline_ms')
+            assert current == previous, ('checkpoint changed authoritative SM fields', token, current, previous)
+        assert all(p.read_text() == content for p, content in final_archives.items()), 'checkpoint changed committed MAM records'
+        await asyncio.to_thread(subprocess.run, checkpoint, env=env, check=True, capture_output=True, text=True)
+        assert not list((root/'data').rglob('*.wal')) and not list((root/'data').rglob('*.rredo'))
+        for token, previous in exported.items():
+            paths = list((root/'data').rglob(token+'.json'))
+            assert len(paths) == 1 and json.loads(paths[0].read_text()) == previous, 'second checkpoint changed SM export'
+        assert all(p.read_text() == content for p, content in final_archives.items()), 'second checkpoint changed MAM records'
+        return {'status': 'pass', 'delay_ms_per_sync': args.delay_ms,
+                'acknowledged_messages': len(identifiers), 'mam_records_per_account': [len(identifiers)]*2,
+                'resumed_resources': len(clients), 'replayed_personal_deliveries': 3*len(identifiers),
+                'offline_checkpoint': {'exported_sessions': len(exported), 'native_archives': len(final_archives),
+                                       'journals_removed': True, 'second_invocation_idempotent': True},
+                'pending_before_reauthentication': [len(record['pending']) for record in snapshots],
+                'sync_between_publication_and_sigkill': summary}
+    finally:
+        if server is not None and server.poll() is None:
+            failing = sys.exc_info()[0] is not None
+            try:
+                await asyncio.to_thread(stop, server)
+            except BaseException:
+                if not failing:
+                    raise
+        await asyncio.gather(*(c.close_raw() for c in clients), return_exceptions=True)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--binary', default='target/release/rust-xmpp')
+    parser.add_argument('--delay-ms', type=int, default=100)
+    parser.add_argument('--messages', type=int, default=3)
+    parser.add_argument('--output', default='/tmp/slow-storage-crash.json')
+    args = parser.parse_args()
+    assert 0 <= args.delay_ms <= 10000 and args.messages > 0
+    with tempfile.TemporaryDirectory(prefix='xmpp-slow-storage-crash-') as directory:
+        root = Path(directory)
+        binary = root/'rust-xmpp'
+        shutil.copy2(Path(args.binary).resolve(), binary)
+        shim = root/'slow_sync.so'
+        subprocess.run(['cc', '-shared', '-fPIC', '-O2', '-Wall', '-Wextra', '-Werror',
+                        str(Path(__file__).parent/'fixtures'/'slow_sync.c'), '-o', str(shim), '-ldl'], check=True)
+        check_shim_scope(root, shim)
+        subprocess.run(['openssl', 'req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:P-256',
+                        '-nodes', '-days', '1', '-subj', '/CN=localhost', '-addext', 'subjectAltName=DNS:localhost',
+                        '-keyout', str(root/'key.pem'), '-out', str(root/'cert.pem')], check=True, capture_output=True)
+        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': [],
+            'smacks_hibernation_time': 600,
+            'modules_enabled': ['tls', 'saslauth', 'smacks', 'csi_simple', 'ping', 'mam', 'pep', 'carbons'],
+            '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([str(binary), 'create-user', str(config), f'load-{number}', '--password-stdin'],
+                           input='load-test-password\n', text=True, check=True, capture_output=True)
+        sync_log = root/'sync.jsonl'
+        env = dict(os.environ, LD_PRELOAD=str(shim), XMPP_SLOW_SYNC_ROOT=str(root/'data'),
+                   XMPP_SLOW_SYNC_LOG=str(sync_log), XMPP_SLOW_SYNC_MS=str(args.delay_ms))
+        with (root/'server.log').open('w+') as log:
+            try:
+                report = asyncio.run(exercise(root, binary, config, port, log, env, sync_log, args))
+            except BaseException as error:
+                # Preserve the first failing run's evidence outside the
+                # TemporaryDirectory before cleanup; never replace its error
+                # with a later shutdown or diagnostics failure.
+                try:
+                    log.flush()
+                    output = Path(args.output)
+                    output.parent.mkdir(parents=True, exist_ok=True)
+                    evidence = Path(tempfile.mkdtemp(prefix=output.stem+'-failure-', dir=output.parent))
+                    shutil.copy2(root/'server.log', evidence/'server.log')
+                    if sync_log.exists():
+                        shutil.copy2(sync_log, evidence/'sync.jsonl')
+                    Path(args.output).write_text(json.dumps({'status': 'fail',
+                        'failure': repr(error), 'delay_ms_per_sync': args.delay_ms,
+                        'binary_sha256': hashlib.sha256(binary.read_bytes()).hexdigest(),
+                        'evidence': str(evidence)}, indent=2)+'\n')
+                except Exception as diagnostic_error:
+                    print(f'Could not preserve failure evidence: {diagnostic_error}', file=sys.stderr)
+                log.seek(0)
+                print(log.read(), file=sys.stderr)
+                raise
+        report['binary_sha256'] = hashlib.sha256(binary.read_bytes()).hexdigest()
+        report['scope'] = 'loopback STARTTLS; real delayed fsync/fdatasync; sender SM acknowledgement then SIGKILL; authoritative queues and MAM checked before reauthentication; at-least-once replay'
+        result = json.dumps(report, indent=2)+'\n'
+        if args.output:
+            Path(args.output).write_text(result)
+        print(result, end='')
+
+
+if __name__ == '__main__':
+    main()

tests/slow_storage_latency_network.py

@@ -0,0 +1,299 @@
+#!/usr/bin/env python3
+"""Fixture-only slow durable IO: two accounts, four TLS/SM resources, MAM/PEP.
+
+Builds only a tiny C LD_PRELOAD interposer; uses an isolated copy of --binary.
+Every intercepted fsync/fdatasync still executes. Directory syncs under the
+fixture data root are delayed too and counted separately. No mount changes.
+Timing includes durable work, with client ACKs withheld during each sample.
+"""
+import argparse
+import asyncio
+import collections
+import hashlib
+import json
+import os
+import pathlib
+import re
+import shutil
+import socket
+import ssl
+import subprocess
+import sys
+import tempfile
+import time
+import xml.etree.ElementTree as ET
+
+from muc_sm_load_network import Client, CLIENT, SM
+from muc_restart_network import stop
+from message_latency_network import slot
+
+CARBONS = 'urn:xmpp:carbons:2'
+PUBSUB = 'http://jabber.org/protocol/pubsub'
+
+
+PHASE_FIELDS = ('pre_routing_ms', 'sender_mam_ms', 'recipient_mam_ms',
+    'delivery_ms', 'carbons_ms', 'offline_push_ms', 'handled_ms')
+SLOW_FIELDS = {'duration_ms', 'route', 'target_count', 'carbon_count', *PHASE_FIELDS}
+
+
+def slow_events(path, delay, rounds, allow_legacy=False):
+    events = []
+    for line in path.read_text().splitlines():
+        if 'event="message-processing-slow"' not in line:
+            continue
+        fields = dict(re.findall(r'(?:^|\s)"([^"\s]+)"="([^"\n]*)"', line))
+        legacy = allow_legacy and set(fields) == {'duration_ms', 'route'}
+        assert legacy or set(fields) == SLOW_FIELDS, ('unexpected slow-event fields', sorted(fields))
+        assert not any(value in line for value in
+            ('slow fixture ', 'body-settled-', 'body-avatar_pending-', '<body')), 'payload leaked into slow diagnostics'
+        event = {key: value if key == 'route' else int(value) for key,value in fields.items()}
+        assert event['duration_ms'] >= 250 and event['route'] == 'local', event
+        if not legacy:
+            assert all(event[key] >= 0 for key in PHASE_FIELDS), event
+            assert sum(event[key] for key in PHASE_FIELDS) <= event['duration_ms']+len(PHASE_FIELDS), event
+            assert event['target_count'] == 1 and event['carbon_count'] == 2, event
+        events.append(event)
+    # At >=100ms the retained durable commits alone exceed the 250ms threshold.
+    # Smaller optional delays may legitimately produce no warning.
+    if delay >= 100:
+        assert len(events) == 2*rounds, ('missing/extra slow diagnostics',len(events),2*rounds)
+    return events
+
+
+class WireReader:
+    def __init__(self, reader, owner):
+        self.reader, self.owner = reader, owner
+        self.parser = ET.XMLPullParser(events=('start', 'end'))
+        self.parser.feed("<root xmlns='jabber:client'>")
+        self.root = next(self.parser.read_events())[1]
+        self.depth = 0
+
+    def __getattr__(self, name):
+        return getattr(self.reader, name)
+
+    async def read(self, size):
+        data = await self.reader.read(size)
+        self.parser.feed(data)
+        for event, node in self.parser.read_events():
+            self.depth += 1 if event == 'start' else -1
+            if event != 'end' or self.depth:
+                continue
+            now = time.monotonic_ns()
+            if node.tag == f'{{{CLIENT}}}message':
+                for message in node.iter(f'{{{CLIENT}}}message'):
+                    if message.get('id'):
+                        kind = 'direct'
+                        for carbon in ('sent', 'received'):
+                            if node.find(f'{{{CARBONS}}}{carbon}') is not None:
+                                kind = carbon
+                        self.owner.arrivals[(message.get('id'), kind)].append(now)
+            if node.tag == f'{{{SM}}}a':
+                self.owner.ack_times[int(node.get('h'))] = now
+            self.root.remove(node)
+        return data
+
+
+class MeasuredClient(Client):
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.arrivals = collections.defaultdict(list)
+        self.ack_times = {}
+        self.reader = WireReader(self.reader, self)
+
+
+async def iq(client, identifier, payload):
+    await client.stanza(f"<iq type='set' id='{identifier}'>{payload}</iq>")
+    result = await client.wait(lambda: client.iqs.pop(identifier, None), identifier)
+    assert result.get('type') == 'result', ET.tostring(result)
+
+
+def sync_rows(path):
+    return [json.loads(row) for row in path.read_text().splitlines()] if path.exists() else []
+
+
+def counts(rows):
+    result = collections.Counter(f"{row['store']}:{row['kind']}:{row['op']}" for row in rows)
+    assert all(row['result'] == 0 for row in rows), 'real sync failed'
+    return {'total': len(rows), 'by_store_kind_op': dict(sorted(result.items()))}
+
+
+async def exercise(root, port, args, log):
+    context = ssl.create_default_context(cafile=str(root/'cert.pem'))
+    clients = []
+    report = []
+    try:
+        for number in range(4):
+            client = await MeasuredClient.connect(number, port, context, 2)
+            clients.append(client)
+            client.auto_ack = False
+            await client.enable()
+            await iq(client, f'carbons-{number}', f"<enable xmlns='{CARBONS}'/>")
+            await client.stanza('<presence/>')
+            await client.fence(f'ready-{number}')
+        alice, bob, alice_other, bob_other = clients
+        avatar = 'urn:xmpp:avatar:metadata'
+        # Self-account subscriptions make avatar events reach sibling resources.
+        await iq(alice, 'avatar-initial', f"<pubsub xmlns='{PUBSUB}'><publish node='{avatar}'><item id='initial'><metadata xmlns='{avatar}'/></item></publish></pubsub>")
+        for client in (alice, alice_other):
+            await iq(client, f'subscribe-{client.number}', f"<pubsub xmlns='{PUBSUB}'><subscribe node='{avatar}' jid='{client.jid}'/></pubsub>")
+        for phase in ('settled', 'avatar_pending'):
+            for turn in range(args.rounds):
+                # Fence all previous ACK work before taking operation counts.
+                for client in clients:
+                    await client.ack()
+                    await client.fence(f'before-{phase}-{turn}-{client.number}')
+                    await client.handled()
+                before = len(sync_rows(log))
+                start_ns = time.monotonic_ns()
+                if phase == 'avatar_pending':
+                    # Publication remains ahead of the body on this stream; no
+                    # client ACK is sent for its IQ result or PEP notifications.
+                    await alice.stanza(f"<iq type='set' id='publish-{turn}'><pubsub xmlns='{PUBSUB}'><publish node='{avatar}'><item id='sample-{turn}'><metadata xmlns='{avatar}'><info id='fixture-avatar' bytes='1' type='image/png'/></metadata></item></publish></pubsub></iq>")
+                identifier = f'body-{phase}-{turn}'
+                begun = time.monotonic_ns()
+                await alice.stanza(f"<message to='{bob.jid}' type='chat' id='{identifier}'><body>slow fixture {identifier}</body></message>")
+                await alice.handled()
+                events = []
+                for client, kind in ((bob, 'direct'), (alice_other, 'sent'), (bob_other, 'received')):
+                    await client.wait(lambda c=client,k=kind: c.arrivals[(identifier,k)], identifier+' '+kind)
+                    arrivals = client.arrivals[(identifier, kind)]
+                    assert len(arrivals) == 1, (identifier, kind, arrivals)
+                    events.append((kind, (arrivals[0]-begun)/1e6))
+                end_ns = time.monotonic_ns()
+                rows = sync_rows(log)[before:]
+                assert rows, 'interposer recorded no data-root sync operations'
+                assert all(row['end_ns']-row['begin_ns'] >= args.current_delay_ms*1000000
+                    for row in rows), 'configured sync delay was not applied'
+                snapshots = [slot(root,c) for c in clients]
+                for index in (1,2,3):
+                    assert sum(identifier in xml for xml in snapshots[index]['pending']) == 1, (identifier,index)
+                if phase == 'avatar_pending':
+                    assert all(any(avatar in xml and 'event' in xml for xml in snapshots[index]['pending'])
+                        for index in (0,2)), 'avatar PEP notifications were not durably pending'
+                report.append({'phase': phase, 'round': turn,
+                    'delivery_ms': dict(events),
+                    'sender_sm_handled_ms': (alice.ack_times[alice.inputs]-begun)/1e6,
+                    'publication_plus_delivery_ms': (end_ns-start_ns)/1e6,
+                    'sync': counts(rows),
+                    'pending_per_resource': [len(snapshot['pending']) for snapshot in snapshots]})
+        assert all(not client.errors for client in clients), [c.errors for c in clients]
+        # Both ends archive each personal body exactly once, despite carbons.
+        archive_counts = [sum(line.startswith('item(') for line in path.open())
+            for path in sorted((root/'data'/'localhost'/'archive').glob('*.list'))]
+        expected = args.history_records + 2*args.rounds
+        assert archive_counts == [expected, expected], (archive_counts, expected)
+        return {'samples': report, 'mam_records_per_account': archive_counts,
+            'status': 'pass'}
+    finally:
+        await asyncio.gather(*(c.close_raw() for c in clients), return_exceptions=True)
+
+
+def check_shim_scope(work, shim):
+    """Prove that both sync APIs preserve success and exclude non-data files."""
+    root = work/'shim-scope-data'
+    root.mkdir()
+    log = work/'shim-scope.jsonl'
+    code = """import os,pathlib,sys
+root,outside=map(pathlib.Path,sys.argv[1:])
+for path in (root/'inside',outside):
+    with path.open('wb') as stream:
+        stream.write(b'fixture')
+        stream.flush()
+        os.fsync(stream.fileno())
+        os.fdatasync(stream.fileno())
+fd=os.open(root,os.O_RDONLY)
+os.fsync(fd)
+os.close(fd)
+"""
+    env = dict(os.environ, LD_PRELOAD=str(shim), XMPP_SLOW_SYNC_ROOT=str(root),
+        XMPP_SLOW_SYNC_LOG=str(log), XMPP_SLOW_SYNC_MS='0')
+    subprocess.run([sys.executable,'-c',code,str(root),str(work/'outside')],
+        env=env,check=True,capture_output=True)
+    rows = sync_rows(log)
+    assert len(rows) == 3, rows
+    assert collections.Counter((r['op'],r['kind']) for r in rows) == {
+        ('fsync','file'):1, ('fdatasync','file'):1, ('fsync','directory'):1}
+    assert all(row['result'] == 0 for row in rows)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--binary', default='target/release/rust-xmpp')
+    parser.add_argument('--delay-ms', type=int, nargs='+', default=[0, 100])
+    parser.add_argument('--rounds', type=int, default=3)
+    parser.add_argument('--history-records', type=int, default=100)
+    parser.add_argument('--output')
+    parser.add_argument('--allow-legacy-phase-logs', action='store_true',
+        help='baseline binaries only: permit old duration/route diagnostics without phase metrics')
+    args = parser.parse_args()
+    assert args.rounds > 0 and args.history_records >= 0
+    assert all(0 <= delay <= 10000 for delay in args.delay_ms)
+    with tempfile.TemporaryDirectory(prefix='xmpp-slow-storage-') as directory:
+        work = pathlib.Path(directory)
+        binary = work/'rust-xmpp'
+        shutil.copy2(pathlib.Path(args.binary).resolve(), binary)
+        shim = work/'slow_sync.so'
+        subprocess.run(['cc', '-shared', '-fPIC', '-O2', '-Wall', '-Wextra', '-Werror',
+            str(pathlib.Path(__file__).parent/'fixtures'/'slow_sync.c'), '-o', str(shim), '-ldl'], check=True)
+        check_shim_scope(work, shim)
+        report = {'binary_sha256': hashlib.sha256(binary.read_bytes()).hexdigest(),
+            'scope': 'loopback STARTTLS; four SM resources/two accounts; MAM + pending avatar metadata PEP; fixture-only fsync/fdatasync including data directories; real sync retained; operation windows include background SM heartbeat',
+            'cases': []}
+        for delay in args.delay_ms:
+            root = work/f'delay-{delay}'
+            root.mkdir()
+            subprocess.run(['openssl', 'req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:P-256',
+                '-nodes', '-days', '1', '-subj', '/CN=localhost', '-addext', 'subjectAltName=DNS:localhost',
+                '-keyout', str(root/'key.pem'), '-out', str(root/'cert.pem')], check=True, capture_output=True)
+            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','csi_simple','ping','mam','pep','carbons'],
+                '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([str(binary),'create-user',str(config),f'load-{number}','--password-stdin'],
+                    input='load-test-password\n',text=True,check=True,capture_output=True)
+                archive = root/'data'/'localhost'/'archive'/f'load%2d{number}.list'
+                archive.parent.mkdir(parents=True,exist_ok=True)
+                with archive.open('w') as stream:
+                    for index in range(args.history_records):
+                        stream.write("item({name='message';attr={xmlns='jabber:client';type='chat'};"
+                            f"key='seed-{index}';when={time.time()};with='load-{1-number}@localhost';"
+                            "{name='body';attr={};'history'}});\n")
+            sync_log = root/'sync.jsonl'
+            env = dict(os.environ, LD_PRELOAD=str(shim), XMPP_SLOW_SYNC_ROOT=str(root/'data'),
+                XMPP_SLOW_SYNC_LOG=str(sync_log), XMPP_SLOW_SYNC_MS=str(delay))
+            with (root/'server.log').open('w+') as log:
+                server = subprocess.Popen([str(binary),'serve',str(config),'--experimental'],env=env,stdout=log,stderr=log)
+                try:
+                    deadline = time.monotonic()+60
+                    while True:
+                        assert server.poll() is None, log.read()
+                        try:
+                            with socket.create_connection(('127.0.0.1',port),timeout=.1):
+                                break
+                        except OSError:
+                            assert time.monotonic()<deadline, 'server startup timeout'
+                            time.sleep(.05)
+                    args.current_delay_ms = delay
+                    case = asyncio.run(exercise(root,port,args,sync_log))
+                    case['delay_ms_per_sync'] = delay
+                    events = slow_events(root/'server.log',delay,args.rounds,args.allow_legacy_phase_logs)
+                    case['server_message_processing_slow_events'] = events
+                    case['server_message_processing_slow_ms'] = [event['duration_ms'] for event in events]
+                    report['cases'].append(case)
+                finally:
+                    stop(server)
+            if args.output:
+                pathlib.Path(args.output).write_text(json.dumps(report,indent=2)+'\n')
+            print(f'slow storage: delay={delay}ms pass',flush=True)
+        print(json.dumps(report,indent=2))
+
+
+if __name__ == '__main__':
+    main()

tests/sm_journal.rs

@@ -0,0 +1,322 @@
+//! Redo recovery is independent of the best-effort readable JSON export.
+use rust_xmpp::sessions::Hub;
+use std::{fs, io::Write, path::Path, time::Duration};
+use tokio::sync::mpsc;
+
+fn committed(dir: &Path) -> (String, Vec<u8>) {
+    let hub = Hub::persistent(dir).unwrap();
+    let (tx, _rx) = mpsc::channel(8);
+    let session = hub.bind("a@x/r", "c", tx, Duration::from_secs(60)).unwrap();
+    let mut state = session.lock().unwrap();
+    state.enable(true).unwrap();
+    state.transmit("durable".into()).unwrap();
+    let token = state.resume_id.clone().unwrap();
+    let json = fs::read(dir.join(format!("{token}.json"))).unwrap();
+    drop(state);
+    drop(hub);
+    (token, json)
+}
+
+#[test]
+fn redo_repairs_missing_or_torn_export_before_json_validation() {
+    for torn in [false, true] {
+        let dir = tempfile::tempdir().unwrap();
+        let (token, _) = committed(dir.path());
+        let json = dir.path().join(format!("{token}.json"));
+        if torn {
+            fs::write(&json, b"{torn").unwrap();
+        } else {
+            fs::remove_file(&json).unwrap();
+        }
+        let hub = Hub::persistent(dir.path()).unwrap();
+        assert_eq!(
+            hub.get("a@x/r")
+                .unwrap()
+                .lock()
+                .unwrap()
+                .pending
+                .front()
+                .map(String::as_str),
+            Some("durable")
+        );
+        let export: serde_json::Value = serde_json::from_slice(&fs::read(json).unwrap()).unwrap();
+        assert_eq!(export["pending"][0], "durable");
+    }
+}
+
+#[test]
+fn partial_tail_is_discarded_and_subsequent_commits_survive_restart() {
+    let dir = tempfile::tempdir().unwrap();
+    let (token, _) = committed(dir.path());
+    let wal = dir.path().join(format!("{token}.wal"));
+    let good_len = fs::metadata(&wal).unwrap().len();
+    fs::OpenOptions::new()
+        .append(true)
+        .open(&wal)
+        .unwrap()
+        .write_all(&[42, 0, 0])
+        .unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    assert_eq!(fs::metadata(&wal).unwrap().len(), good_len);
+    hub.get("a@x/r")
+        .unwrap()
+        .lock()
+        .unwrap()
+        .mark_handled()
+        .unwrap();
+    drop(hub);
+    let hub = Hub::persistent(dir.path()).unwrap();
+    assert_eq!(hub.get("a@x/r").unwrap().lock().unwrap().handled, 1);
+}
+
+#[test]
+fn complete_corrupt_frame_fails_closed_even_with_valid_export() {
+    let dir = tempfile::tempdir().unwrap();
+    let (token, _) = committed(dir.path());
+    let wal = dir.path().join(format!("{token}.wal"));
+    let mut bytes = fs::read(&wal).unwrap();
+    *bytes.last_mut().unwrap() ^= 1;
+    fs::write(wal, bytes).unwrap();
+    assert!(Hub::persistent(dir.path()).is_err());
+}
+
+#[test]
+fn durable_tombstone_overrides_resurrected_stale_json() {
+    let dir = tempfile::tempdir().unwrap();
+    let (token, old) = committed(dir.path());
+    let hub = Hub::persistent(dir.path()).unwrap();
+    hub.kick_owner("a@x");
+    drop(hub);
+    fs::write(dir.path().join(format!("{token}.json")), old).unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    assert!(hub.get("a@x/r").is_none());
+}
+
+#[test]
+fn legacy_json_migrates_and_baseline_survives_partial_first_append() {
+    let dir = tempfile::tempdir().unwrap();
+    let (token, _) = committed(dir.path());
+    let wal = dir.path().join(format!("{token}.wal"));
+    fs::remove_file(&wal).unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    hub.get("a@x/r")
+        .unwrap()
+        .lock()
+        .unwrap()
+        .mark_handled()
+        .unwrap();
+    drop(hub);
+    let bytes = fs::read(&wal).unwrap();
+    let baseline_len = u32::from_le_bytes(bytes[8..12].try_into().unwrap()) as usize;
+    fs::write(&wal, &bytes[..8 + 40 + baseline_len + 12]).unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    let session = hub.get("a@x/r").unwrap();
+    let state = session.lock().unwrap();
+    assert_eq!(state.handled, 0);
+    assert_eq!(state.pending.front().map(String::as_str), Some("durable"));
+}
+
+#[test]
+fn rotation_preserves_latest_snapshot_and_bounds_redo_size() {
+    let dir = tempfile::tempdir().unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    let (tx, _rx) = mpsc::channel(8);
+    let session = hub.bind("a@x/r", "c", tx, Duration::from_secs(60)).unwrap();
+    let token;
+    {
+        let mut state = session.lock().unwrap();
+        state.enable(true).unwrap();
+        token = state.resume_id.clone().unwrap();
+        state.transmit("x".repeat(1024 * 1024)).unwrap();
+        for _ in 0..35 {
+            state.mark_handled().unwrap();
+        }
+    }
+    let wal = dir.path().join(format!("{token}.wal"));
+    assert!(fs::metadata(wal).unwrap().len() < 18 * 1024 * 1024);
+    drop(hub);
+    let hub = Hub::persistent(dir.path()).unwrap();
+    let session = hub.get("a@x/r").unwrap();
+    let state = session.lock().unwrap();
+    assert_eq!(state.handled, 35);
+    assert_eq!(state.pending.front().unwrap().len(), 1024 * 1024);
+}
+
+#[test]
+fn incomplete_baseline_or_magic_never_falls_back_to_json() {
+    for keep in [0, 4, 8, 12, 48] {
+        let dir = tempfile::tempdir().unwrap();
+        let (token, _) = committed(dir.path());
+        let wal = dir.path().join(format!("{token}.wal"));
+        let bytes = fs::read(&wal).unwrap();
+        fs::write(wal, &bytes[..keep]).unwrap();
+        assert!(
+            Hub::persistent(dir.path()).is_err(),
+            "truncated baseline at {keep}"
+        );
+    }
+}
+
+#[test]
+#[cfg(unix)]
+fn journal_symlink_is_rejected_without_following_target() {
+    use std::os::unix::fs::symlink;
+    let dir = tempfile::tempdir().unwrap();
+    let (token, _) = committed(dir.path());
+    let wal = dir.path().join(format!("{token}.wal"));
+    let other = tempfile::NamedTempFile::new().unwrap();
+    fs::remove_file(&wal).unwrap();
+    symlink(other.path(), wal).unwrap();
+    assert!(Hub::persistent(dir.path()).is_err());
+}
+
+#[test]
+fn export_failure_after_commit_advances_memory_and_fences_later_writes() {
+    let dir = tempfile::tempdir().unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    let (tx, _rx) = mpsc::channel(8);
+    let session = hub.bind("a@x/r", "c", tx, Duration::from_secs(60)).unwrap();
+    session.lock().unwrap().enable(true).unwrap();
+    let token = session.lock().unwrap().resume_id.clone().unwrap();
+    let obstruction = dir.path().join(format!("{token}.tmp"));
+    fs::create_dir(&obstruction).unwrap();
+    {
+        let mut state = session.lock().unwrap();
+        state.mark_handled().unwrap();
+        assert_eq!(state.handled, 1);
+        assert!(state.mark_handled().is_err());
+        assert_eq!(state.handled, 1);
+    }
+    drop(hub);
+    fs::remove_dir(obstruction).unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    assert_eq!(hub.get("a@x/r").unwrap().lock().unwrap().handled, 1);
+}
+
+#[test]
+fn retired_journals_are_collected_in_bounded_batches() {
+    let dir = tempfile::tempdir().unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    for index in 0..64 {
+        let (tx, _rx) = mpsc::channel(1);
+        let connection = format!("c{index}");
+        let session = hub
+            .bind("a@x/r", &connection, tx, Duration::from_secs(60))
+            .unwrap();
+        session.lock().unwrap().enable(true).unwrap();
+        hub.disconnect(&connection, true);
+    }
+    assert_eq!(
+        fs::read_dir(dir.path())
+            .unwrap()
+            .filter(|entry| {
+                entry
+                    .as_ref()
+                    .unwrap()
+                    .path()
+                    .extension()
+                    .is_some_and(|ext| ext == "wal")
+            })
+            .count(),
+        0
+    );
+    drop(hub);
+    assert!(Hub::persistent(dir.path()).unwrap().get("a@x/r").is_none());
+}
+
+#[test]
+fn offline_checkpoint_repairs_exports_removes_redo_and_allows_legacy_restart() {
+    let dir = tempfile::tempdir().unwrap();
+    let (token, _) = committed(dir.path());
+    let json = dir.path().join(format!("{token}.json"));
+    fs::write(&json, b"{torn").unwrap();
+    assert_eq!(
+        rust_xmpp::session_store::checkpoint_exports(dir.path()).unwrap(),
+        1
+    );
+    assert!(!dir.path().join(format!("{token}.wal")).exists());
+    let export: serde_json::Value = serde_json::from_slice(&fs::read(json).unwrap()).unwrap();
+    assert_eq!(export["pending"][0], "durable");
+    let hub = Hub::persistent(dir.path()).unwrap();
+    let session = hub.get("a@x/r").unwrap();
+    assert_eq!(
+        session.lock().unwrap().pending.front().map(String::as_str),
+        Some("durable")
+    );
+    session.lock().unwrap().mark_handled().unwrap();
+    assert!(dir.path().join(format!("{token}.wal")).exists());
+}
+
+#[test]
+fn offline_checkpoint_refuses_live_store_and_preserves_redo() {
+    let dir = tempfile::tempdir().unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    let (tx, _rx) = mpsc::channel(8);
+    let session = hub.bind("a@x/r", "c", tx, Duration::from_secs(60)).unwrap();
+    session.lock().unwrap().enable(true).unwrap();
+    let token = session.lock().unwrap().resume_id.clone().unwrap();
+    assert!(rust_xmpp::session_store::checkpoint_exports(dir.path()).is_err());
+    assert!(dir.path().join(format!("{token}.wal")).exists());
+}
+
+#[test]
+fn offline_checkpoint_durably_applies_expiration_and_tombstones() {
+    let dir = tempfile::tempdir().unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    let (tx, _rx) = mpsc::channel(8);
+    let session = hub.bind("a@x/r", "c", tx, Duration::from_secs(60)).unwrap();
+    let token;
+    {
+        let mut state = session.lock().unwrap();
+        state.enable(true).unwrap();
+        token = state.resume_id.clone().unwrap();
+        state.expires = Some(std::time::Instant::now());
+        state.checkpoint().unwrap();
+    }
+    drop(hub);
+    assert_eq!(
+        rust_xmpp::session_store::checkpoint_exports(dir.path()).unwrap(),
+        0
+    );
+    assert!(!dir.path().join(format!("{token}.json")).exists());
+    assert!(!dir.path().join(format!("{token}.wal")).exists());
+    assert!(Hub::persistent(dir.path()).unwrap().get("a@x/r").is_none());
+}
+
+#[test]
+fn superseded_frames_still_require_valid_checksums_json_and_baseline_flags() {
+    use sha2::{Digest, Sha256};
+    for damage in ["checksum", "json", "baseline"] {
+        let dir = tempfile::tempdir().unwrap();
+        let (token, _) = committed(dir.path());
+        let wal = dir.path().join(format!("{token}.wal"));
+        let mut bytes = fs::read(&wal).unwrap();
+        // Damage the baseline, leaving both later committed snapshots intact.
+        let mut len = u32::from_le_bytes(bytes[8..12].try_into().unwrap());
+        let mut payload = 48..48 + len as usize;
+        match damage {
+            "checksum" => bytes[payload.end - 1] ^= 1,
+            "json" => bytes[payload.end - 1] = b'!',
+            "baseline" => {
+                let start = bytes[payload.clone()]
+                    .windows(4)
+                    .position(|part| part == b"true")
+                    .unwrap()
+                    + payload.start;
+                // Preserve valid JSON, but make the first frame a continuation.
+                bytes.splice(start..start + 4, b"false".iter().copied());
+                len += 1;
+                payload.end += 1;
+                bytes[8..12].copy_from_slice(&len.to_le_bytes());
+                bytes[12..16].copy_from_slice(&(!len).to_le_bytes());
+            }
+            _ => unreachable!(),
+        }
+        if damage != "checksum" {
+            let checksum = Sha256::digest(&bytes[payload]);
+            bytes[16..48].copy_from_slice(&checksum);
+        }
+        fs::write(wal, bytes).unwrap();
+        assert!(Hub::persistent(dir.path()).is_err(), "{damage}");
+    }
+}

tests/sm_restart_network.py

@@ -5,6 +5,7 @@ Usage: python3 tests/sm_restart_network.py BINARY
 Pending server stanzas are at-least-once: a received but unacknowledged stanza
 must replay. This test does not assert exactly-once client stanza processing.
 """
+import hashlib
 import json
 from pathlib import Path
 import signal
@@ -23,6 +24,36 @@ SM = 'urn:xmpp:sm:3'
 ROOM = 'durable@rooms.localhost'
 
 
+def durable_slot(root, token):
+    """Read the authoritative checksummed WAL, never a potentially stale export."""
+    paths = list((root/'data').rglob(token + '.wal'))
+    assert len(paths) == 1, ('missing or duplicate durable SM journal', paths)
+    data = paths[0].read_bytes()
+    assert data[:8] == b'XMPPSM01', 'invalid durable SM journal header'
+    offset, frames, record = 8, 0, None
+    while offset < len(data):
+        # An abrupt kill can leave an incomplete final append. Recovery uses
+        # the preceding complete frame; an incomplete baseline is never valid.
+        if len(data)-offset < 8:
+            break
+        size = int.from_bytes(data[offset:offset+4], 'little')
+        inverse = int.from_bytes(data[offset+4:offset+8], 'little')
+        assert size ^ inverse == 0xffffffff and size <= 8*1024*1024+1024, 'invalid durable SM frame length'
+        end = offset+40+size
+        if end > len(data):
+            break
+        payload = data[offset+40:end]
+        assert hashlib.sha256(payload).digest() == data[offset+8:offset+40], 'invalid durable SM frame checksum'
+        frame = json.loads(payload)
+        assert frame['baseline'] == (frames == 0), 'invalid durable SM baseline sequence'
+        record = frame['snapshot']
+        frames += 1
+        offset = end
+    assert frames and record is not None, 'missing committed live SM snapshot'
+    assert record['resume_id'] == token, 'durable SM token mismatch'
+    return record
+
+
 def enable(client):
     client.send(f"<enable xmlns='{SM}' resume='true'/>")
     response = client.take(lambda e: e.tag == f'{{{SM}}}enabled')
@@ -115,7 +146,7 @@ def run(binary):
                 client = Client(port, cert, user, 'secret', resource)
                 clients.append(client)
                 return client
-            def restart(how):
+            def restart(how, before_start=None):
                 nonlocal server
                 if how == 'kill':
                     server.send_signal(signal.SIGKILL)
@@ -125,6 +156,8 @@ def run(binary):
                 for client in clients:
                     client.close()
                 clients.clear()
+                if before_start:
+                    before_start()
                 server = start(binary, config, port, log)
             try:
                 server = start(binary, config, port, log)
@@ -139,7 +172,15 @@ def run(binary):
                 alice.send("<message to='bob@localhost'><body>incoming before crash</body></message>")
                 bob.body('incoming before crash')
                 assert handled(alice) == 3
-                restart('kill')
+                def corrupt_export():
+                    record = durable_slot(root, token)
+                    assert record['handled'] == 3
+                    exports = list((root/'data').rglob(token + '.json'))
+                    assert len(exports) == 1
+                    exports[0].write_text('{torn JSON export')
+                # The fsynced journal must repair a torn readable export and
+                # retain the latest handled counter and pending IQ.
+                restart('kill', corrupt_export)
                 for user, ident, h in [('bob', token, 1), ('alice', 'unknown-token', 1), ('alice', token, 99)]:
                     resume(port, cert, ident, h, clients, user=user, success=False)
                 alice, result = resume(port, cert, token, 1, clients)
@@ -203,9 +244,7 @@ def run(binary):
                 assert handled(alice) == 5
                 # Inspect the committed slot before any Bob authentication.
                 # A history replay on rejoining is not offline SM delivery.
-                slots = list((root/'data').rglob(bob_token + '.json'))
-                assert len(slots) == 1, slots
-                record = json.loads(slots[0].read_text())
+                record = durable_slot(root, bob_token)
                 assert record['jid'] == 'bob@localhost/room', record
                 queued = [ET.fromstring(xml) for xml in record['pending']]
                 assert any(packet.get('from') == ROOM + '/Alice'
@@ -256,7 +295,7 @@ def run(binary):
                 if sys.exc_info()[0] is not None:
                     log.seek(0)
                     print(log.read(), file=sys.stderr)
-    print('PASS: TLS legacy/SASL2 durable SM after SIGKILL/SIGTERM, counters, ack pruning, at-least-once pending replay, owner/token/h rejection, transport/resource conflict, MUC activation/offline fanout, grace expiry and account deletion/recreation revocation')
+    print('PASS: TLS legacy/SASL2 durable SM after SIGKILL/SIGTERM, WAL repair of torn JSON export, counters, ack pruning, at-least-once pending replay, owner/token/h rejection, transport/resource conflict, MUC activation/offline fanout, grace expiry and account deletion/recreation revocation')
 
 
 if __name__ == '__main__':

tests/storage_accounts.rs

@@ -353,7 +353,7 @@ fn purge_account_exact_files_preserves_others_and_extras() {
     let root = t.path();
     for store in ["accounts", "roster", "pep_aabbcc", "custom_store"] {
         for user in ["alice", "bob"] {
-            for ext in ["dat", "list", "lidx", "rappend"] {
+            for ext in ["dat", "list", "lidx", "rappend", "rredo"] {
                 let p = storage::path(root, "example.test", Some(user), store).with_extension(ext);
                 std::fs::create_dir_all(p.parent().unwrap()).unwrap();
                 std::fs::write(&p, b"fixture").unwrap();
@@ -367,7 +367,7 @@ fn purge_account_exact_files_preserves_others_and_extras() {
     std::fs::write(&hostwide, b"keep").unwrap();
     a::remove_account(root, "example.test", "alice").unwrap();
     for store in ["accounts", "roster", "pep_aabbcc", "custom_store"] {
-        for ext in ["dat", "list", "lidx", "rappend"] {
+        for ext in ["dat", "list", "lidx", "rappend", "rredo"] {
             assert!(
                 !storage::path(root, "example.test", Some("alice"), store)
                     .with_extension(ext)