- Исправлена задержка доставки MUC с растущим архивом: после каждого

сообщения
  больше не выполняется полный scan/rewrite архива для expiry. Очистка
  остаётся
  в maintenance и перед MAM-запросами; append и crash checkpoint до
  fanout сохранены.
- Неизменённые native config/persistent index не перезаписываются при
  каждом
  сообщении. Удалённые файлы восстанавливаются, ошибки экспорта
  повторяются.
- CSI больше не сохраняет SM snapshot при отсутствии изменения durable
  очереди.
  Изменённая очередь сохраняется до отправки; при ошибке записи очередь
  откатывается.
- Личные сообщения, IQ и входящая федерация к аккаунтам больше не
  захватывают
  общий MUC mutex, ожидая несвязанную работу с комнатными архивами.
- Добавлены регрессии native exports/archive scanning/CSI rollback и
  реальный
  TLS latency check в verify-local.sh, автоматически включённый и в CI.

Измерения и границы подтверждения:
[docs/MESSAGE-LATENCY.md](docs/MESSAGE-LATENCY.md).
Результаты итогового прогона:
[docs/VALIDATION-0.8.1.md](docs/VALIDATION-0.8.1.md).

- Личные сообщения и SM ACK больше не ждут сетевую отправку push.
  Уведомления
  отправляются через очередь1024 с максимум8параллельных маршрутов;
  worker
  и его задачи отменяются при завершении serve. Ошибки очереди/маршрута
  логируются.
- Offline сообщение сохраняется до публикации push, чтобы проснувшийся
  клиент
  мог сразу получить его. Подтверждена очередь при первом TLS stream
  push gateway.
- По умолчанию CSI считает важными bodyless OMEMO legacy/2, delivery
  receipts
  и chat markers. Явный список `csi_important_payloads` заменяет
  расширенные defaults.
  Низкоприоритетный headline по-прежнему буферизуется. Policy
  обновляется и при SM resume.
- Выбор адресатов и carbons фильтрует владельцев до захвата session
  mutex;
  зависшая сессия чужого аккаунта не задерживает этот путь. Duplicate SM
  ACK
  без изменения durable counters/queue не переписывает snapshot.
- Bind2 и MAM metadata IQ читают только первый/последний native record
  вместо
  сканирования всей истории. Индекс не требуется; recovery/лимиты записи
  сохранены.
- Добавлен `message-processing-slow`: обработка message дольше250мс,
  поля
  duration_ms/route, без payload/токенов. Это время обработки сервером,
  не peer receipt.
- Новые сетевые регрессии включены в verify-local.sh/GitHub CI.

Причины, изменение CSI policy, измерения и ограничения:
[docs/PERSONAL-MESSAGE-LATENCY.md](docs/PERSONAL-MESSAGE-LATENCY.md).

tests/archive_edges.rs

@@ -0,0 +1,133 @@
+use rust_xmpp::{archive, storage};
+use std::{
+    fs::{self, File},
+    io::{Seek, SeekFrom, Write},
+    path::{Path, PathBuf},
+};
+
+const MAX_RECORD: usize = 16 * 1024 * 1024;
+
+fn path(root: &Path) -> PathBuf {
+    let p = storage::path(root, "example.org", Some("alice"), "archive").with_extension("list");
+    fs::create_dir_all(p.parent().unwrap()).unwrap();
+    p
+}
+fn record(id: &str) -> String {
+    format!(
+        "item({{name='message';attr={{xmlns='jabber:client'}};key='{id}';when=42;with='bob@example.org';{{name='body';attr={{}};'hello'}}}});\n"
+    )
+}
+fn edges(root: &Path) -> anyhow::Result<Option<(archive::Entry, archive::Entry)>> {
+    archive::edges_store(root, "example.org", Some("alice"), "archive")
+}
+
+#[test]
+fn metadata_skips_large_malformed_middle_without_creating_cache() {
+    let t = tempfile::tempdir().unwrap();
+    let p = path(t.path());
+    let mut file = File::create(&p).unwrap();
+    file.write_all(record("first").as_bytes()).unwrap();
+    file.write_all(b"item(not valid Lua in the middle\n")
+        .unwrap();
+    // A sparse >128 MiB history makes accidental full scans fail the record
+    // bound; only the endpoints are part of this metadata operation.
+    file.seek(SeekFrom::Start(129 * 1024 * 1024)).unwrap();
+    file.write_all(b"\n").unwrap();
+    file.write_all(record("last").as_bytes()).unwrap();
+    drop(file);
+    let (first, last) = edges(t.path()).unwrap().unwrap();
+    assert_eq!((first.id.as_str(), last.id.as_str()), ("first", "last"));
+    assert_eq!(first.when, 42.0);
+    assert_eq!(last.peer, "bob@example.org");
+    assert!(!p.with_extension("lidx").exists());
+    assert!(archive::load(t.path(), "example.org", "alice").is_err());
+
+    // An opaque native index is irrelevant and must remain untouched.
+    fs::write(p.with_extension("lidx"), b"stale binary index").unwrap();
+    assert_eq!(edges(t.path()).unwrap().unwrap().1.id, "last");
+    assert_eq!(
+        fs::read(p.with_extension("lidx")).unwrap(),
+        b"stale binary index"
+    );
+}
+
+#[test]
+fn multiline_endpoints_match_native_scanner_with_blank_lines() {
+    let t = tempfile::tempdir().unwrap();
+    let p = path(t.path());
+    let first = record("first").replace(";", ";\n");
+    let last = format!(
+        "item({{\nname='message';attr={{}};key='last';when=43;\n{{name='body';attr={{}};[=[{}]=]}}\n}});\n \t\n",
+        "long multiline text\n".repeat(1800)
+    );
+    fs::write(&p, format!(" \t\n\n{first}\n{}{last}", record("middle"))).unwrap();
+    let rows = archive::load(t.path(), "example.org", "alice").unwrap();
+    let (a, b) = edges(t.path()).unwrap().unwrap();
+    assert_eq!(a.id, rows.first().unwrap().id);
+    assert_eq!(b.id, rows.last().unwrap().id);
+    assert_eq!(b.xml, rows.last().unwrap().xml);
+}
+
+#[test]
+fn missing_empty_whitespace_and_single_record() {
+    let t = tempfile::tempdir().unwrap();
+    assert!(edges(t.path()).unwrap().is_none());
+    let p = path(t.path());
+    for contents in ["", " \t\r\n\n"] {
+        fs::write(&p, contents).unwrap();
+        assert!(edges(t.path()).unwrap().is_none());
+    }
+    // EOF need not have a newline.
+    fs::write(&p, record("only").trim_end()).unwrap();
+    let (first, last) = edges(t.path()).unwrap().unwrap();
+    assert_eq!(first.id, "only");
+    assert_eq!(last.id, first.id);
+    assert_eq!(last.xml, first.xml);
+}
+
+#[test]
+fn tail_padding_respects_record_bound() {
+    let t = tempfile::tempdir().unwrap();
+    let p = path(t.path());
+    let mut data = record("first").into_bytes();
+    let last = record("last");
+    data.extend_from_slice(last.as_bytes());
+    data.resize(data.len() + MAX_RECORD - last.len(), b' ');
+    fs::write(&p, &data).unwrap();
+    assert_eq!(edges(t.path()).unwrap().unwrap().1.id, "last");
+    data.push(b' ');
+    fs::write(&p, data).unwrap();
+    assert!(edges(t.path()).unwrap_err().to_string().contains("16 MiB"));
+}
+
+#[test]
+fn oversized_prefix_and_malformed_endpoints_fail() {
+    let t = tempfile::tempdir().unwrap();
+    let p = path(t.path());
+    let mut data = vec![b' '; MAX_RECORD + 1];
+    data.extend_from_slice(b"\n");
+    data.extend_from_slice(record("last").as_bytes());
+    fs::write(&p, data).unwrap();
+    assert!(edges(t.path()).unwrap_err().to_string().contains("16 MiB"));
+    fs::write(&p, format!("item(bad first);\n{}", record("last"))).unwrap();
+    assert!(edges(t.path()).is_err());
+    fs::write(&p, format!("{}item(bad last);\n", record("first"))).unwrap();
+    assert!(edges(t.path()).is_err());
+}
+
+#[test]
+fn endpoint_read_recovers_partial_append_before_seeking() {
+    let t = tempfile::tempdir().unwrap();
+    let p = path(t.path());
+    let committed = format!("{}{}", record("first"), record("last"));
+    fs::write(&p, format!("{committed}item({{broken partial")).unwrap();
+    fs::write(
+        p.with_extension("rappend"),
+        (committed.len() as u64).to_le_bytes(),
+    )
+    .unwrap();
+    let (first, last) = edges(t.path()).unwrap().unwrap();
+    assert_eq!((first.id.as_str(), last.id.as_str()), ("first", "last"));
+    assert_eq!(fs::read(&p).unwrap(), committed.as_bytes());
+    assert!(!p.with_extension("rappend").exists());
+}

tests/bind2_archive_latency_network.py

@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""Measure real TLS/SASL2 Bind2 reconnects with a large personal MAM history.
+
+Usage: python3 tests/bind2_archive_latency_network.py --binary BINARY \
+    --records 100000 --output REPORT.json
+
+Seeds valid native records before startup, then checks exact first/last metadata
+on every fresh connection. Latency measurements have no performance threshold.
+The executable is copied before use so concurrent rebuilds cannot replace it.
+"""
+import argparse
+import datetime
+import hashlib
+import json
+from pathlib import Path
+import shutil
+import socket
+import statistics
+import subprocess
+import tempfile
+import time
+import xml.etree.ElementTree as ET
+
+from sasl2_bind import BIND2, SASL2, Client
+
+MAM = 'urn:xmpp:mam:2'
+BASE_TIME = 1767225600
+
+
+def stamp(seconds):
+    return datetime.datetime.fromtimestamp(seconds, datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
+
+
+def distribution(values):
+    ordered = sorted(values)
+    return {'samples_ms': values, 'median_ms': statistics.median(values),
+            'p95_ms': ordered[min(len(ordered) - 1, int(len(ordered) * .95))],
+            'max_ms': ordered[-1]}
+
+
+def exercise(binary, root, records, samples):
+    with socket.socket() as probe:
+        probe.bind(('127.0.0.1', 0))
+        port = probe.getsockname()[1]
+    subprocess.run(['openssl', 'req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:P-256',
+                    '-nodes', '-days', '1', '-subj', '/CN=localhost', '-addext', 'subjectAltName=DNS:localhost',
+                    '-addext', 'basicConstraints=critical,CA:FALSE', '-keyout', str(root/'key.pem'),
+                    '-out', str(root/'cert.pem')], check=True, capture_output=True)
+    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', 'sasl2', 'sasl2_bind2', 'mam', 'carbons', 'csi_simple', 'smacks'],
+        'default_archive_policy': True, 'archive_expires_after': 'never'},
+        'hosts': {'localhost': {'options': {'authentication': 'internal_hashed', 'password_hash': 'SHA-256',
+            'ssl': {'key': str(root/'key.pem'), 'certificate': str(root/'cert.pem')}}}}}))
+    subprocess.run([str(binary), 'create-user', str(config), 'alice', '--password-stdin'],
+                   input='secret\n', text=True, check=True, capture_output=True)
+    archive = root/'data'/'localhost'/'archive'/'alice.list'
+    archive.parent.mkdir(parents=True, exist_ok=True)
+    with archive.open('w') as output:
+        for number in range(records):
+            output.write("item({name='message';attr={xmlns='jabber:client';type='chat';"
+                         f"from='bob@localhost';stamp='{stamp(BASE_TIME + number)}'" + "};"
+                         f"key='seed-{number}';when={BASE_TIME + number};with='bob@localhost';"
+                         "{name='body';attr={};'seed personal history'}});\n")
+    expected = {'start': {'id': 'seed-0', 'timestamp': stamp(BASE_TIME)},
+                'end': {'id': f'seed-{records - 1}', 'timestamp': stamp(BASE_TIME + records - 1)}}
+    archive_bytes = archive.stat().st_size
+    index_before_start = archive.with_suffix('.lidx').exists()
+    assert not index_before_start, 'seed unexpectedly has a native index'
+    auth_ms, connect_ms, metadata_ms = [], [], []
+    metadata_samples, iq_metadata_samples = [], []
+    with (root/'server.log').open('w+') as log:
+        server = subprocess.Popen([str(binary), 'serve', str(config), '--experimental'], stdout=log, stderr=log)
+        try:
+            deadline = time.monotonic() + 30
+            while True:
+                assert server.poll() is None, 'server stopped before readiness'
+                try:
+                    with socket.create_connection(('127.0.0.1', port), timeout=.1):
+                        break
+                except OSError:
+                    assert time.monotonic() < deadline, 'server did not start'
+                    time.sleep(.05)
+            for sample in range(samples):
+                # Each sample reconnects and authenticates the same account.
+                # Client verifies the advertised SASL2/Bind2 inline features.
+                connected = time.perf_counter()
+                client = Client(port, root/'cert.pem')
+                client.sock.settimeout(30)
+                try:
+                    begun = time.perf_counter()
+                    response = client.auth(f"<bind xmlns='{BIND2}'><tag>HistoryReconnect</tag></bind>")
+                    arrived = time.perf_counter()
+                    auth_ms.append((arrived - begun) * 1000)
+                    connect_ms.append((arrived - connected) * 1000)
+                    jid = response.find('{' + SASL2 + '}authorization-identifier')
+                    assert jid is not None and jid.text.startswith('alice@localhost/HistoryReconnect/'), 'missing bound JID'
+                    bound = response.find('{' + BIND2 + '}bound')
+                    assert bound is not None, 'missing Bind2 bound result'
+                    metadata = bound.find('{' + MAM + '}metadata')
+                    assert metadata is not None, 'missing MAM metadata'
+                    actual = {}
+                    for edge in ('start', 'end'):
+                        node = metadata.find('{' + MAM + '}' + edge)
+                        assert node is not None, f'missing MAM {edge}'
+                        actual[edge] = dict(node.attrib)
+                    assert actual == expected, (actual, expected)
+                    assert len(list(metadata)) == 2, 'unexpected MAM metadata children'
+                    metadata_samples.append(actual)
+                    # Session is immediately bound; the post-success features
+                    # must not ask this client to perform legacy resource bind.
+                    assert b'xmpp-bind' not in client.until(b'</stream:features>')
+                    query_id = f'metadata-{sample}'
+                    begun = time.perf_counter()
+                    client.send(f"<iq type='get' id='{query_id}'><metadata xmlns='{MAM}'/></iq>")
+                    iq = ET.fromstring(client.until(b'</iq>'))
+                    metadata_ms.append((time.perf_counter() - begun) * 1000)
+                    assert iq.get('type') == 'result' and iq.get('id') == query_id, 'metadata IQ failed'
+                    iq_metadata = iq.find('{' + MAM + '}metadata')
+                    assert iq_metadata is not None, 'metadata IQ payload missing'
+                    iq_actual = {}
+                    for edge in ('start', 'end'):
+                        node = iq_metadata.find('{' + MAM + '}' + edge)
+                        assert node is not None, f'missing metadata IQ {edge}'
+                        iq_actual[edge] = dict(node.attrib)
+                    assert iq_actual == expected, (iq_actual, expected)
+                    assert len(list(iq_metadata)) == 2, 'unexpected metadata IQ children'
+                    iq_metadata_samples.append(iq_actual)
+                finally:
+                    client.close()
+            assert archive.stat().st_size == archive_bytes, 'reconnect modified seeded history'
+        finally:
+            server.terminate()
+            try:
+                server.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                server.kill()
+                server.wait()
+    return {'status': 'pass', 'seeded_personal_archive_records': records, 'archive_bytes': archive_bytes,
+            'archive_expires_after': 'never', 'index_present_before_start': index_before_start,
+            'reconnect_samples': samples, 'expected_metadata': expected, 'observed_metadata': metadata_samples,
+            'mam_metadata_iq': distribution(metadata_ms), 'observed_iq_metadata': iq_metadata_samples,
+            'sasl2_bind2_success': distribution(auth_ms), 'tcp_starttls_to_bind2_success': distribution(connect_ms)}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--binary', required=True)
+    parser.add_argument('--records', type=int, default=100000)
+    parser.add_argument('--samples', type=int, default=10)
+    parser.add_argument('--output')
+    args = parser.parse_args()
+    assert args.records > 0 and args.samples > 0
+    source = Path(args.binary).resolve()
+    with tempfile.TemporaryDirectory(prefix='xmpp-bind2-archive-latency-') as directory:
+        root = Path(directory)
+        binary = root/'rust-xmpp'
+        shutil.copy2(source, binary)
+        report = {'binary_source': str(source), 'binary_sha256': hashlib.sha256(binary.read_bytes()).hexdigest(),
+                  'scope': 'real loopback TCP/STARTTLS/SASL2 Bind2; isolated personal MAM fixture; no production latency claim',
+                  'timing': 'raw ordered samples; auth timer starts before authenticate and ends at parsed success; full timer includes TCP/STARTTLS/features; metadata IQ timer runs from send to parsed result',
+                  'result': exercise(binary, root, args.records, args.samples)}
+    text = json.dumps(report, indent=2) + '\n'
+    if args.output:
+        Path(args.output).write_text(text)
+    print(text, end='')
+
+
+if __name__ == '__main__':
+    main()

tests/message_latency_network.py

@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+"""Measure real STARTTLS direct/MUC delivery with durable SM and CSI.
+
+Usage: python3 tests/message_latency_network.py --binary target/release/rust-xmpp
+Reports loopback median/p95/max, not a production latency guarantee. There are
+no millisecond performance assertions: bounded delivery, exact counts, SM
+counters and durable CSI buffering/urgent flush are the regression assertions.
+The source executable is copied before testing so concurrent rebuilds are safe.
+"""
+import argparse
+import asyncio
+import collections
+import datetime
+import hashlib
+import json
+import pathlib
+import shutil
+import socket
+import ssl
+import statistics
+import subprocess
+import tempfile
+import time
+
+from muc_sm_load_network import Client, MUC, start
+from muc_restart_network import stop
+
+
+class TimedBodies(collections.Counter):
+    def __init__(self):
+        super().__init__()
+        self.arrivals = {}
+
+    def __setitem__(self, key, value):
+        self.arrivals.setdefault(key, time.perf_counter())
+        super().__setitem__(key, value)
+
+
+def distribution(values):
+    values = sorted(value * 1000 for value in values)
+    return {'samples': len(values), 'median_ms': statistics.median(values),
+            'p95_ms': values[min(len(values) - 1, int(len(values) * .95))],
+            'max_ms': values[-1], 'samples_ms': values}
+
+
+async def bounded(operation):
+    # A functional timeout, deliberately much wider than a performance budget.
+    return await asyncio.wait_for(operation, 20)
+
+
+async def body(client, text):
+    await bounded(client.wait(lambda: client.bodies[text] == 1, text))
+    return client.bodies.arrivals[text]
+
+
+def slot(root, client):
+    paths = list((root / 'data').rglob(client.token + '.json'))
+    assert len(paths) == 1, 'missing or duplicate durable SM record'
+    return json.loads(paths[0].read_text())
+
+
+async def queued(root, client, text):
+    # Observe persisted queue state instead of assuming a short socket timeout
+    # proves buffering. Bob sends no input/acks that could flush CSI here.
+    deadline = time.monotonic() + 20
+    while time.monotonic() < deadline:
+        record = slot(root, client)
+        if any(text in xml for xml in record['csi']['queue']):
+            assert client.bodies[text] == 0, 'buffered message leaked onto wire'
+            return
+        assert client.bodies[text] == 0, 'inactive CSI failed to hold headline'
+        await asyncio.sleep(.01)
+    raise AssertionError('headline did not enter durable CSI queue')
+
+
+async def csi_state(root, client, state):
+    # Alternate first so persisted state proves this exact control was handled,
+    # including resetting any temporary flushing window. A ping on the receiver
+    # would itself cause CSI activity and invalidate the buffering experiment.
+    for target in (state != 'inactive', state == 'inactive'):
+        await client.send(f"<{'inactive' if target else 'active'} xmlns='urn:xmpp:csi:0'/>")
+        deadline = time.monotonic() + 20
+        while slot(root, client)['csi']['inactive'] != target:
+            assert time.monotonic() < deadline, 'CSI state was not durably handled'
+            await asyncio.sleep(.001)
+
+
+async def exercise(root, port, rounds, archive, history_records):
+    context = ssl.create_default_context(cafile=str(root / 'cert.pem'))
+    clients = []
+    report = {'mam_enabled': archive, 'sm_enabled': True, 'rounds': rounds,
+              'seeded_muc_archive_records': history_records if archive else 0}
+    try:
+        for number in range(2):
+            client = await bounded(Client.connect(number, port, context, 2))
+            clients.append(client)
+            client.bodies = TimedBodies()
+            # Explicit acks between samples avoid racing inbound ack activity
+            # with inactive-CSI assertions. SM remains enabled throughout.
+            client.auto_ack = False
+            await bounded(client.enable())
+            await client.stanza('<presence/>')
+            await bounded(client.fence(f'connected-{number}'))
+            await client.ack()
+        alice, bob = clients
+        room = 'latency@rooms.localhost'
+        for client in clients:
+            await client.stanza(f"<presence to='{room}/Nick-{client.number}'><x xmlns='{MUC}'><history maxstanzas='0'/></x></presence>")
+            await bounded(client.wait(lambda c=client: c.presences.get(f'{room}/Nick-{c.number}'), 'MUC self presence'))
+            assert client.presences[f'{room}/Nick-{client.number}'].get('type') != 'error'
+        for client in clients:
+            await bounded(client.fence(f'joined-{client.number}'))
+            await client.ack()
+
+        for state in ('active', 'inactive'):
+            measurements = collections.defaultdict(list)
+            for turn in range(rounds):
+                # Reset CSI immediately before each sample: an important output
+                # opens a configured temporary flushing window for later output.
+                await csi_state(root, bob, state)
+                text = f'{state}-direct-{turn}'
+                begun = time.perf_counter()
+                await alice.stanza(f"<message to='{bob.jid}' type='chat'><body>{text}</body></message>")
+                arrived = await body(bob, text)
+                measurements['direct_peer'].append(arrived - begun)
+                reply = text + '-reply'
+                await bob.stanza(f"<message to='{alice.jid}' type='chat'><body>{reply}</body></message>")
+                measurements['direct_application_rtt'].append(await body(alice, reply) - begun)
+                await csi_state(root, bob, state)
+                text = f'{state}-muc-{turn}'
+                begun = time.perf_counter()
+                await alice.stanza(f"<message to='{room}' type='groupchat'><body>{text}</body></message>")
+                echo, peer = await asyncio.gather(body(alice, text), body(bob, text))
+                measurements['muc_sender_echo'].append(echo - begun)
+                measurements['muc_peer'].append(peer - begun)
+                for client in clients:
+                    await client.ack()
+            report[state] = {name: distribution(values) for name, values in measurements.items()}
+
+        # A low-priority headline is actually held; a normal chat body must
+        # release it in order even while the receiver remains CSI inactive.
+        await csi_state(root, bob, 'inactive')
+        held, urgent = 'queued-before-urgent', 'urgent-flush-body'
+        await alice.stanza(f"<message to='{bob.jid}' type='headline'><body>{held}</body></message>")
+        await bounded(alice.fence('headline-routed'))
+        await queued(root, bob, held)
+        begun = time.perf_counter()
+        await alice.stanza(f"<message to='{bob.jid}' type='chat'><body>{urgent}</body></message>")
+        held_at, urgent_at = await asyncio.gather(body(bob, held), body(bob, urgent))
+        assert held_at <= urgent_at, 'urgent flush reordered queued output'
+        report['urgent_flush'] = distribution([urgent_at - begun])
+        report['headline_durably_buffered'] = True
+
+        # Explicit activation must independently drain low-priority output.
+        await bob.ack()
+        await csi_state(root, bob, 'inactive')
+        held = 'queued-before-active'
+        await alice.stanza(f"<message to='{bob.jid}' type='headline'><body>{held}</body></message>")
+        await bounded(alice.fence('activation-headline-routed'))
+        await queued(root, bob, held)
+        await bob.send("<active xmlns='urn:xmpp:csi:0'/>")
+        await body(bob, held)
+        report['activation_drained_queue'] = True
+        for client in clients:
+            await bounded(client.fence(f'final-{client.number}'))
+            await client.ack()
+            await bounded(client.handled())
+            assert not client.errors, client.errors
+            assert all(count == 1 for count in client.bodies.values()), 'duplicate wire delivery'
+        archives = list((root / 'data' / 'localhost' / 'archive').glob('*.list'))
+        records = sum(sum(line.startswith('item(') for line in path.open()) for path in archives)
+        assert (records > 0) == archive, ('MAM toggle did not affect actual archive storage', records)
+        report['personal_archive_records'] = records
+        muc_archive = root / 'data' / 'rooms%2elocalhost' / 'muc_log' / 'latency.list'
+        muc_records = sum(line.startswith('item(') for line in muc_archive.open()) if muc_archive.exists() else 0
+        assert muc_records == (history_records + 2 * rounds if archive else 0), ('MUC archive seed/append mismatch', muc_records)
+        report['muc_archive_records'] = muc_records
+        report['wire_body_deliveries'] = sum(sum(client.bodies.values()) for client in clients)
+        assert report['wire_body_deliveries'] == rounds * 8 + 3
+        report['status'] = 'pass'
+        return report
+    finally:
+        await asyncio.gather(*(client.close_raw() for client in clients), return_exceptions=True)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--binary', default='target/release/rust-xmpp')
+    parser.add_argument('--rounds', type=int, default=20)
+    parser.add_argument('--history-records', type=int, nargs='+', default=[0],
+                        help='MUC native archive sizes to seed before starting each isolated server')
+    parser.add_argument('--output', help='optional JSON measurement report')
+    args = parser.parse_args()
+    assert args.rounds > 0
+    assert all(count >= 0 for count in args.history_records)
+    with tempfile.TemporaryDirectory(prefix='xmpp-message-latency-') as directory:
+        work = pathlib.Path(directory)
+        binary = work / 'rust-xmpp'
+        shutil.copy2(pathlib.Path(args.binary).resolve(), binary)
+        report = {'binary_sha256': hashlib.sha256(binary.read_bytes()).hexdigest(),
+                  'scope': 'loopback STARTTLS; isolated fixture; no production latency claim', 'cases': []}
+        cases = [(False, 0)] + [(True, count) for count in args.history_records]
+        for archive, history_records in cases:
+            root = work / (f'mam-on-{history_records}' if archive else 'mam-off')
+            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',
+                            '-addext', 'basicConstraints=critical,CA:FALSE', '-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]
+            modules = ['tls', 'saslauth', 'smacks', 'csi_simple', 'ping', 'disco']
+            if archive:
+                modules.append('mam')
+            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': modules, 'default_archive_policy': True, 'archive_expires_after': 'never',
+                'ssl': {'key': str(root/'key.pem'), 'certificate': str(root/'cert.pem')}},
+                'hosts': {'localhost': {'options': {'authentication': 'internal_hashed', 'password_hash': 'SHA-256'}},
+                          'rooms.localhost': {'component': 'muc', 'options': {'muc_room_locking': False,
+                              'modules_enabled': ['muc_mam'] if archive else [],
+                              'muc_log_expires_after': 604800}}}}))
+            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)
+            if archive and history_records:
+                # Native Prosody `.list` fixture, matching archive_mam.rs. All
+                # timestamps are recent, so retention scans must retain them.
+                # Seeding occurs before any server opens this isolated store.
+                archive_path = root / 'data' / 'rooms%2elocalhost' / 'muc_log' / 'latency.list'
+                archive_path.parent.mkdir(parents=True, exist_ok=True)
+                now = time.time()
+                stamp = datetime.datetime.fromtimestamp(now, datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
+                with archive_path.open('w') as output:
+                    for number in range(history_records):
+                        output.write("item({name='message';attr={xmlns='jabber:client';type='groupchat';"
+                                     f"from='latency@rooms.localhost/Seed';stamp='{stamp}'" + "};"
+                                     f"key='seed-{number}';when={now};with='message<groupchat';"
+                                     "{name='body';attr={};'seed history'}});\n")
+            with (root/'server.log').open('w+') as log:
+                server = start(str(binary), config, port, log)
+                try:
+                    report['cases'].append(asyncio.run(exercise(root, port, args.rounds, archive, history_records)))
+                finally:
+                    stop(server)
+        result = json.dumps(report, indent=2) + '\n'
+        if args.output:
+            pathlib.Path(args.output).write_text(result)
+        print(result, end='')
+
+
+if __name__ == '__main__':
+    main()

tests/muc_latency_storage.rs

@@ -0,0 +1,146 @@
+use rust_xmpp::{archive, muc::Muc, protocol::ModuleResult, storage};
+use std::{fs, path::Path, time::SystemTime};
+
+fn send(muc: &mut Muc, xml: &str) -> ModuleResult {
+    let doc = roxmltree::Document::parse(xml).unwrap();
+    muc.handle("alice@test/a", doc.root_element())
+        .unwrap()
+        .unwrap()
+}
+
+fn room(muc: &mut Muc, name: &str) {
+    send(
+        muc,
+        &format!(
+            "<presence to='{name}@rooms.test/Alice'><x xmlns='http://jabber.org/protocol/muc'/></presence>"
+        ),
+    );
+    send(
+        muc,
+        &format!(
+            "<iq type='set' to='{name}@rooms.test'><query xmlns='http://jabber.org/protocol/muc#owner'><x xmlns='jabber:x:data' type='submit'/></query></iq>"
+        ),
+    );
+}
+
+fn chat(muc: &mut Muc) -> ModuleResult {
+    send(
+        muc,
+        "<message to='r@rooms.test' type='groupchat'><body>Fresh</body></message>",
+    )
+}
+
+fn mark(path: &Path) -> SystemTime {
+    let marker = SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_000_000);
+    fs::File::options()
+        .write(true)
+        .open(path)
+        .unwrap()
+        .set_times(fs::FileTimes::new().set_modified(marker))
+        .unwrap();
+    fs::metadata(path).unwrap().modified().unwrap()
+}
+
+#[test]
+fn groupchat_skips_native_rewrites_and_archive_retention_scan() {
+    let temp = tempfile::tempdir().unwrap();
+    let mut muc = Muc::new(temp.path().into(), "rooms.test".into());
+    muc.set_defaults(serde_json::json!({"persistent":true,"archiving":true}));
+    muc.set_archive_retention(Some(60.0));
+    room(&mut muc, "r");
+    room(&mut muc, "other");
+
+    let config = storage::path(temp.path(), "rooms.test", Some("r"), "config");
+    let other = storage::path(temp.path(), "rooms.test", Some("other"), "config");
+    let index = storage::path(temp.path(), "rooms.test", None, "persistent");
+    let journal = storage::path(temp.path(), "rooms.test", None, "rust_muc_checkpoint");
+    let config_time = mark(&config);
+    let other_time = mark(&other);
+    let index_time = mark(&index);
+    let journal_time = mark(&journal);
+
+    // An unreadable historic record makes any accidental full archive scan fail.
+    // Appending a fresh record needs only the existing file length, never its XML.
+    let log = storage::path(temp.path(), "rooms.test", Some("r"), "muc_log").with_extension("list");
+    fs::create_dir_all(log.parent().unwrap()).unwrap();
+    let historic = b"item(this is intentionally invalid historic archive data);\n";
+    fs::write(&log, historic).unwrap();
+    let delivered = chat(&mut muc);
+    assert_eq!(delivered.deliveries.len(), 1);
+    let contents = fs::read(&log).unwrap();
+    assert!(contents.starts_with(historic));
+    assert!(contents.len() > historic.len());
+    assert!(archive::load_store(temp.path(), "rooms.test", Some("r"), "muc_log").is_err());
+    for (path, before) in [
+        (&config, config_time),
+        (&other, other_time),
+        (&index, index_time),
+    ] {
+        assert_eq!(fs::metadata(path).unwrap().modified().unwrap(), before);
+    }
+    assert_ne!(
+        fs::metadata(&journal).unwrap().modified().unwrap(),
+        journal_time
+    );
+    assert!(fs::read_to_string(journal).unwrap().contains("Fresh"));
+}
+
+#[test]
+fn native_exports_repair_missing_files_and_export_changed_config() {
+    let temp = tempfile::tempdir().unwrap();
+    let mut muc = Muc::new(temp.path().into(), "rooms.test".into());
+    muc.set_defaults(serde_json::json!({"persistent":true}));
+    room(&mut muc, "r");
+    room(&mut muc, "other");
+    let config = storage::path(temp.path(), "rooms.test", Some("r"), "config");
+    let other = storage::path(temp.path(), "rooms.test", Some("other"), "config");
+    let index = storage::path(temp.path(), "rooms.test", None, "persistent");
+    let original = fs::read(&config).unwrap();
+    fs::remove_file(&config).unwrap();
+    fs::remove_file(&index).unwrap();
+    assert_eq!(chat(&mut muc).deliveries.len(), 1);
+    assert_eq!(fs::read(&config).unwrap(), original);
+    let recovered = storage::load(temp.path(), "rooms.test", None, "persistent")
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        recovered.get("r@rooms.test"),
+        Some(&storage::LuaValue::Bool(true))
+    );
+    assert_eq!(
+        recovered.get("other@rooms.test"),
+        Some(&storage::LuaValue::Bool(true))
+    );
+    let other_time = mark(&other);
+    let index_time = mark(&index);
+    let result = send(
+        &mut muc,
+        "<iq type='set' to='r@rooms.test'><query xmlns='http://jabber.org/protocol/muc#owner'><x xmlns='jabber:x:data' type='submit'><field var='muc#roomconfig_roomname'><value>Updated</value></field></x></query></iq>",
+    );
+    assert!(result.replies[0].contains("result"));
+    assert!(fs::read_to_string(config).unwrap().contains("Updated"));
+    assert_eq!(fs::metadata(other).unwrap().modified().unwrap(), other_time);
+    assert_eq!(
+        fs::metadata(&index).unwrap().modified().unwrap(),
+        index_time
+    );
+
+    // Export failures are nonfatal after the authoritative room commit, and
+    // exports must be retried even if no configuration changed since failure.
+    fs::remove_file(&index).unwrap();
+    fs::create_dir(&index).unwrap();
+    assert_eq!(chat(&mut muc).deliveries.len(), 1);
+    fs::remove_dir(&index).unwrap();
+    assert_eq!(chat(&mut muc).deliveries.len(), 1);
+    let recovered = storage::load(temp.path(), "rooms.test", None, "persistent")
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        recovered.get("r@rooms.test"),
+        Some(&storage::LuaValue::Bool(true))
+    );
+    assert_eq!(
+        recovered.get("other@rooms.test"),
+        Some(&storage::LuaValue::Bool(true))
+    );
+}

tests/personal_message_latency_network.py

@@ -0,0 +1,270 @@
+#!/usr/bin/env python3
+"""Separate personal-chat wire delivery, handled SM ACK and client receipt timing.
+
+Native personal MAM/roster fixtures, real loopback STARTTLS, durable SM/CSI.
+A forced CSI dwell is reported explicitly and is not a server performance sample.
+The default exercises both the runtime defaults (urgent encrypted/receipt payloads)
+and an explicit empty importance list (native policy). Use --expect-held for the
+old binary; --csi-policy default limits a baseline run to its default policy.
+"""
+import argparse
+import asyncio
+import collections
+import hashlib
+import json
+import pathlib
+import shutil
+import socket
+import ssl
+import subprocess
+import tempfile
+import time
+import xml.etree.ElementTree as ET
+
+from message_latency_network import bounded, csi_state, distribution, slot
+from muc_sm_load_network import Client, CLIENT, SM, start
+from muc_restart_network import stop
+
+
+class ObservedReader:
+    """Observe complete wire stanzas while retaining the existing client's parser."""
+    def __init__(self, reader, client):
+        self.reader, self.client = reader, client
+        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):
+        chunk = await self.reader.read(size)
+        self.parser.feed(chunk)
+        for event, node in self.parser.read_events():
+            self.depth += 1 if event == 'start' else -1
+            if event == 'end' and self.depth == 0:
+                now = time.perf_counter()
+                if node.tag == f'{{{CLIENT}}}message' and node.get('id'):
+                    self.client.message_times[node.get('id')].append(now)
+                if node.tag == f'{{{SM}}}a':
+                    self.client.ack_times[int(node.get('h'))] = now
+                self.root.remove(node)
+        return chunk
+
+
+class MeasuredClient(Client):
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.message_times = collections.defaultdict(list)
+        self.ack_times = {}
+        self.reader = ObservedReader(self.reader, self)
+
+
+async def seen(client, identifier):
+    await bounded(client.wait(lambda: client.message_times[identifier], identifier))
+    assert len(client.message_times[identifier]) == 1
+    return client.message_times[identifier][0]
+
+
+async def send_measured(sender, receiver, identifier, payload, typ='chat'):
+    begun = time.perf_counter()
+    await sender.stanza(f"<message to='{receiver.jid}' type='{typ}' id='{identifier}'>{payload}</message>")
+    await bounded(sender.handled())
+    ack_at = sender.ack_times[sender.inputs]
+    return begun, ack_at
+
+
+async def exercise(root, port, args, policy):
+    clients = []
+    try:
+        context = ssl.create_default_context(cafile=str(root/'cert.pem'))
+        for n in range(2):
+            client = await bounded(MeasuredClient.connect(n, port, context, 2))
+            clients.append(client)
+            client.auto_ack = False
+            await bounded(client.enable())
+            await client.stanza('<presence/>')
+            await bounded(client.fence(f'connected-{n}'))
+            await client.ack()
+        alice, bob = clients
+        results = {}
+        for state in ('active', 'inactive'):
+            values = collections.defaultdict(list)
+            for turn in range(args.rounds):
+                await csi_state(root, alice, 'active')
+                await csi_state(root, bob, state)
+                identifier = f'{state}-body-{turn}'
+                begun, ack_at = await send_measured(alice, bob, identifier, '<body>personal test</body>')
+                delivery = await seen(bob, identifier)
+                receipt_id = identifier + '-receipt'
+                # Receipt generation belongs to the receiving client; the server
+                # does not synthesize XEP-0184 receipts on SM acknowledgement.
+                receipt_sent = time.perf_counter()
+                await bob.stanza(f"<message to='{alice.jid}' type='chat' id='{receipt_id}'><received xmlns='urn:xmpp:receipts' id='{identifier}'/></message>")
+                receipt = await seen(alice, receipt_id)
+                values['delivery'].append(delivery-begun)
+                values['sender_sm_handled'].append(ack_at-begun)
+                values['client_receipt_end_to_end'].append(receipt-begun)
+                values['receipt_return_leg'].append(receipt-receipt_sent)
+                for client in clients:
+                    await client.ack()
+            results[state] = {name: distribution(v) for name, v in values.items()}
+
+        payloads = {
+            'receipt_only': "<received xmlns='urn:xmpp:receipts' id='earlier-message'/>",
+            'omemo_legacy_only': "<encrypted xmlns='eu.siacs.conversations.axolotl'><header sid='1'><iv>AA==</iv></header><payload>AA==</payload></encrypted>",
+            'omemo_2_only': "<encrypted xmlns='urn:xmpp:omemo:2'><header sid='1'><iv>AA==</iv></header><payload>AA==</payload></encrypted>",
+            'omemo_with_eme': "<encrypted xmlns='eu.siacs.conversations.axolotl'><payload>AA==</payload></encrypted><encryption xmlns='urn:xmpp:eme:0' namespace='eu.siacs.conversations.axolotl'/>",
+        }
+        payloads.update({f'marker_{name}':
+            f"<{name} xmlns='urn:xmpp:chat-markers:0' id='earlier-message'/>"
+            for name in ('received', 'displayed', 'acknowledged')})
+        payloads['low_priority_headline'] = '<body>held low priority</body>'
+        for name, payload in payloads.items():
+            for state in ('active', 'inactive'):
+                await csi_state(root, alice, 'active')
+                await csi_state(root, bob, state)
+                identifier = name + '-' + state
+                begun, ack_at = await send_measured(alice, bob, identifier, payload,
+                    'headline' if name == 'low_priority_headline' else 'chat')
+                deadline = time.monotonic() + 20
+                while True:
+                    buffered = any(identifier in xml for xml in slot(root, bob)['csi']['queue'])
+                    if buffered or bob.message_times[identifier]:
+                        break
+                    assert time.monotonic() < deadline, identifier
+                    await asyncio.sleep(.001)
+                expect_held = state == 'inactive' and (
+                    name == 'low_priority_headline' or (
+                        name != 'omemo_with_eme' and (policy == 'native' or args.expect_held)))
+                assert buffered == expect_held, (policy, identifier, buffered, expect_held)
+                activation = None
+                if buffered:
+                    assert not bob.message_times[identifier]
+                    await asyncio.sleep(args.csi_dwell)
+                    assert not bob.message_times[identifier], 'CSI unexpectedly emitted held payload'
+                    activation = time.perf_counter()
+                    await bob.send("<active xmlns='urn:xmpp:csi:0'/>")
+                delivered = await seen(bob, identifier)
+                results[identifier] = {'delivery_ms': (delivered-begun)*1000,
+                    'sender_sm_handled_ms': (ack_at-begun)*1000,
+                    'durably_csi_buffered': buffered,
+                    'forced_dwell_seconds': args.csi_dwell if buffered else 0,
+                    'after_activation_ms': (delivered-activation)*1000 if activation else None}
+                for client in clients:
+                    await client.ack()
+
+        await csi_state(root, bob, 'active')
+        # Keep receiving/draining but deliberately withhold recipient ACKs.
+        for n in range(args.backlog):
+            await alice.stanza(f"<message to='{bob.jid}' type='chat' id='backlog-{n}'><body>unacked backlog</body></message>")
+            await seen(bob, f'backlog-{n}')
+        if args.backlog:
+            await seen(bob, f'backlog-{args.backlog-1}')
+        await bounded(alice.handled())
+        pending = len(slot(root, bob)['pending'])
+        begun, ack_at = await send_measured(alice, bob, 'after-backlog', '<body>after backlog</body>')
+        delivery = await seen(bob, 'after-backlog')
+        results['sm_backlog'] = {'seeded_messages': args.backlog, 'durable_pending_before_sample': pending,
+            'delivery_ms': (delivery-begun)*1000, 'sender_sm_handled_ms': (ack_at-begun)*1000}
+        for client in clients:
+            await client.ack()
+            assert not client.errors, client.errors
+        results['status'] = 'pass'
+        return results
+    finally:
+        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('--rounds', type=int, default=3)
+    parser.add_argument('--history-records', type=int, default=100000)
+    parser.add_argument('--roster-sizes', type=int, nargs='+', default=[0, 5000, 10000])
+    parser.add_argument('--backlog', type=int, default=500)
+    parser.add_argument('--csi-dwell', type=float, default=2)
+    parser.add_argument('--csi-policy', choices=('default', 'native', 'both'), default='both',
+                        help='native explicitly configures csi_important_payloads=[]; both verifies override')
+    parser.add_argument('--expect-held', action='store_true',
+                        help='baseline binaries: expect native holds even when option is absent')
+    parser.add_argument('--output')
+    args = parser.parse_args()
+    assert args.rounds > 0 and args.history_records >= 0
+    assert all(count >= 0 for count in args.roster_sizes)
+    assert 0 <= args.backlog < 4000 and args.csi_dwell >= 0
+    with tempfile.TemporaryDirectory(prefix='xmpp-personal-latency-') as directory:
+        work = pathlib.Path(directory)
+        binary = work/'rust-xmpp'
+        shutil.copy2(pathlib.Path(args.binary).resolve(), binary)
+        report = {'binary_sha256': hashlib.sha256(binary.read_bytes()).hexdigest(),
+            'scope': 'loopback STARTTLS; fixture only; forced CSI dwell explicitly separated', 'cases': []}
+        policies = ('default', 'native') if args.csi_policy == 'both' else (args.csi_policy,)
+        for roster_size, policy in ((n, p) for n in args.roster_sizes for p in policies):
+            root = work/f'roster-{roster_size}-{policy}'
+            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', 'roster'],
+                'default_archive_policy': 'roster' if roster_size else True, 'archive_expires_after': 'never',
+                'ssl': {'key': str(root/'key.pem'), 'certificate': str(root/'cert.pem')}},
+                'hosts': {'localhost': {'options': {'authentication': 'internal_hashed', 'password_hash': 'SHA-256'}}}}))
+            if policy == 'native':
+                configured = json.loads(config.read_text())
+                configured['global']['csi_important_payloads'] = []
+                config.write_text(json.dumps(configured))
+            for n in range(2):
+                subprocess.run([str(binary), 'create-user', str(config), f'load-{n}', '--password-stdin'],
+                    input='load-test-password\n', text=True, check=True, capture_output=True)
+                archive = root/'data'/'localhost'/'archive'/f'load%2d{n}.list'
+                archive.parent.mkdir(parents=True, exist_ok=True)
+                now = time.time()
+                with archive.open('w') as output:
+                    for i in range(args.history_records):
+                        output.write("item({name='message';attr={xmlns='jabber:client';type='chat'};"
+                            f"key='seed-{i}';when={now};with='load-{1-n}@localhost';"
+                            "{name='body';attr={};'native personal history'}});\n")
+                if roster_size:
+                    roster = root/'data'/'localhost'/'roster'/f'load%2d{n}.dat'
+                    roster.parent.mkdir(parents=True, exist_ok=True)
+                    with roster.open('w') as output:
+                        output.write("return { [false]={version=1};\n")
+                        output.write(f"['load-{1-n}@localhost']={{subscription='none'}};\n")
+                        for i in range(roster_size-1):
+                            output.write(f"['contact-{i}@example.invalid']={{subscription='none'}};\n")
+                        output.write('}\n')
+            with (root/'server.log').open('w+') as log:
+                server = start(str(binary), config, port, log)
+                try:
+                    results = asyncio.run(exercise(root, port, args, policy))
+                    counts = [sum(line.startswith('item(') for line in path.open())
+                        for path in sorted((root/'data'/'localhost'/'archive').glob('*.list'))]
+                    expected = args.history_records + 4*args.rounds + 10 + args.backlog + 1
+                    assert counts == [expected, expected], (counts, expected)
+                    results['final_personal_mam_records_per_user'] = counts
+                    results.update({'csi_policy': policy, 'baseline_expect_held': args.expect_held,
+                        'roster_contacts_per_user': roster_size,
+                        'seeded_personal_mam_records_per_user': args.history_records,
+                        'archive_policy': 'roster' if roster_size else 'always'})
+                    report['cases'].append(results)
+                    if args.output:
+                        pathlib.Path(args.output).write_text(json.dumps(report, indent=2)+'\n')
+                    print(f'personal latency: roster={roster_size} csi={policy} pass', flush=True)
+                finally:
+                    stop(server)
+        result = json.dumps(report, indent=2)+'\n'
+        if args.output:
+            pathlib.Path(args.output).write_text(result)
+        print(result, end='')
+
+
+if __name__ == '__main__':
+    main()

tests/push_latency_network.py

@@ -0,0 +1,296 @@
+#!/usr/bin/env python3
+"""Personal delivery must not await a stalled external XEP-0357 gateway.
+
+Real c2s and s2s STARTTLS; Bob hibernates or goes offline with registered push. The
+account domain is chat.localhost and the external push service is localhost,
+which is not a configured local host. Its standard s2s endpoint is a local
+fixture on 127.0.0.1:5269; the special-use localhost name avoids external DNS.
+After verified TLS it reads the server's stream but never sends features.
+--expect-blocked demonstrates the old bug in a bounded two-second observation;
+the default asserts both durable deliveries and Alice's handled SM ACK while
+the real gateway transport remains stalled. This is a local regression fixture,
+not an APNs/FCM or production-client validation.
+"""
+import argparse
+import asyncio
+import hashlib
+import json
+import os
+import pathlib
+import shutil
+import socket
+import ssl
+import subprocess
+import tempfile
+import time
+
+from message_latency_network import slot
+import muc_sm_load_network as network
+from muc_sm_load_network import SM, start
+
+PUSH = 'urn:xmpp:push:0'
+GATEWAY = 'localhost'
+DOMAIN = 'chat.localhost'
+network.STREAM = network.STREAM.replace("to='localhost'", f"to='{DOMAIN}'")
+
+
+class Client(network.Client):
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.jid = self.jid.replace('@localhost/', f'@{DOMAIN}/')
+
+    async def fence(self, identifier):
+        await self.stanza(f"<iq type='get' to='{DOMAIN}' id='{identifier}'><ping xmlns='urn:xmpp:ping'/></iq>")
+        result = await self.wait(lambda: self.iqs.pop(identifier, None), identifier)
+        assert result.get('type') == 'result'
+
+
+def openssl(root, *args):
+    subprocess.run(['openssl', *args], cwd=root, check=True, capture_output=True)
+
+
+def certificates(root):
+    openssl(root, 'req', '-x509', '-newkey', 'ec', '-pkeyopt',
+            'ec_paramgen_curve:P-256', '-nodes', '-days', '1', '-subj',
+            '/CN=Push latency fixture CA', '-addext',
+            'basicConstraints=critical,CA:TRUE', '-keyout', 'ca.key', '-out', 'ca.pem')
+    openssl(root, 'req', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:P-256',
+            '-nodes', '-subj', '/CN=localhost', '-keyout', 'key.pem', '-out', 'server.csr')
+    (root/'extensions').write_text('basicConstraints=critical,CA:FALSE\n'
+        'keyUsage=critical,digitalSignature\nextendedKeyUsage=serverAuth,clientAuth\n'
+        'subjectAltName=DNS:localhost,DNS:chat.localhost\n')
+    openssl(root, 'x509', '-req', '-in', 'server.csr', '-CA', 'ca.pem',
+            '-CAkey', 'ca.key', '-CAcreateserial', '-days', '1', '-extfile',
+            'extensions', '-out', 'cert.pem')
+
+
+async def poll(predicate, description, timeout=5):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return
+        await asyncio.sleep(.01)
+    raise AssertionError(description)
+
+
+class Blackhole:
+    def __init__(self, root):
+        self.context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        self.context.load_cert_chain(root/'cert.pem', root/'key.pem')
+        self.seen = asyncio.Event()
+        self.tasks = set()
+        self.writers = set()
+        self.errors = []
+        self.closed = False
+        self.snapshot = lambda: None
+        self.pending_at_stream = None
+
+    async def accept(self, reader, writer):
+        self.tasks.add(asyncio.current_task())
+        self.writers.add(writer)
+        try:
+            stream = await reader.readuntil(b'>')
+            if stream.startswith(b'<?xml'):
+                stream = await reader.readuntil(b'>')
+            assert b'stream:stream' in stream and GATEWAY.encode() in stream
+            header = ("<stream:stream xmlns='jabber:server' "
+                "xmlns:stream='http://etherx.jabber.org/streams' "
+                "from='localhost' to='chat.localhost' version='1.0'>")
+            writer.write((header + "<stream:features><starttls "
+                "xmlns='urn:ietf:params:xml:ns:xmpp-tls'><required/>"
+                "</starttls></stream:features>").encode())
+            await writer.drain()
+            assert b'starttls' in await reader.readuntil(b'>')
+            writer.write(b"<proceed xmlns='urn:ietf:params:xml:ns:xmpp-tls'/>")
+            await writer.drain()
+            await writer.start_tls(self.context)
+            stream = await reader.readuntil(b'>')
+            if stream.startswith(b'<?xml'):
+                stream = await reader.readuntil(b'>')
+            assert b'stream:stream' in stream and GATEWAY.encode() in stream
+            writer.write(header.encode())
+            await writer.drain()
+            self.pending_at_stream = self.snapshot()
+            self.seen.set()
+            # Keep transport open; features are never sent and no route completes.
+            await reader.read()
+            self.closed = True
+        except asyncio.CancelledError:
+            raise
+        except Exception as error:
+            self.errors.append(repr(error))
+        finally:
+            writer.close()
+            self.writers.discard(writer)
+            self.tasks.discard(asyncio.current_task())
+
+    async def close(self):
+        for task in list(self.tasks):
+            task.cancel()
+        await asyncio.gather(*self.tasks, return_exceptions=True)
+        for writer in list(self.writers):
+            writer.close()
+
+
+async def exercise(root, binary, config, port, args, log):
+    gateway = Blackhole(root)
+    try:
+        listener = await asyncio.start_server(gateway.accept, '127.0.0.1', 5269)
+    except OSError as error:
+        raise RuntimeError(f'push fixture needs an unused {GATEWAY}:5269: {error}') from error
+    clients = []
+    server = None
+    try:
+        # rustls-native-certs uses this explicit fixture trust bundle; identity
+        # verification still checks the remote JID's DNS SAN during real TLS.
+        previous = os.environ.get('SSL_CERT_FILE')
+        os.environ['SSL_CERT_FILE'] = str(root/'ca.pem')
+        try:
+            server = start(str(binary), config, port, log)
+        finally:
+            if previous is None:
+                os.environ.pop('SSL_CERT_FILE', None)
+            else:
+                os.environ['SSL_CERT_FILE'] = previous
+        context = ssl.create_default_context(cafile=str(root/'ca.pem'))
+        for number in range(2):
+            client = await asyncio.wait_for(Client.connect(number, port, context, 2), 10)
+            clients.append(client)
+            client.auto_ack = False
+            if number == 0 or args.recipient == 'hibernating':
+                await client.enable()
+            await client.stanza('<presence/>')
+            await client.fence(f'ready-{number}')
+            if client.token:
+                await client.handled()
+                await client.ack()
+        alice, bob = clients
+        await bob.stanza(f"<iq type='set' to='{bob.jid}' id='push-enable'>"
+            f"<enable xmlns='{PUSH}' jid='{GATEWAY}' node='phone'/></iq>")
+        response = await bob.wait(lambda: bob.iqs.get('push-enable'), 'push registration')
+        assert response.get('type') == 'result'
+        if bob.token:
+            await bob.handled()
+            await bob.ack()
+            await poll(lambda: bool(slot(root, bob).get('push_identifier')), 'missing durable push binding')
+        else:
+            registration = list((root/'data').rglob('cloud_notify/*.dat'))
+            assert len(registration) == 1 and 'phone' in registration[0].read_text(), 'missing native push registration'
+        await bob.close_raw()
+        mode = 'hibernating' if bob.token else 'removed'
+        await poll(lambda: any('session-disconnected' in line and mode in line
+                   and bob.jid in line for line in (root/'server.log').read_text().splitlines()),
+                   f'Bob did not enter {mode} state')
+        if not bob.token:
+            assert len(list((root/'data').rglob('rust_smacks/*.json'))) == 1, 'offline Bob unexpectedly has an SM slot'
+
+        def pending_state():
+            if bob.token:
+                pending = slot(root, bob)['pending']
+            else:
+                pending = [path.read_text() for path in (root/'data').rglob('offline/*.list')]
+            return {identifier: any(identifier in stanza for stanza in pending)
+                    for identifier in ('push-first', 'push-second')}
+
+        gateway.snapshot = pending_state
+        before = alice.inputs
+        began = time.perf_counter()
+        for identifier in ('push-first', 'push-second'):
+            await alice.stanza(f"<message to='{bob.jid}' type='chat' id='{identifier}'>"
+                f"<body>{identifier}</body></message>")
+        await alice.send(f"<r xmlns='{SM}'/>")
+        # Observe for the fixed interval even if a faulty implementation emits an
+        # early ACK. Durable pending state and a real blocked gateway are required.
+        await asyncio.sleep(args.bound)
+        present = pending_state()
+        acknowledgements = [int(node.get('h')) for node in alice.controls[f'{{{SM}}}a']]
+        seen_at_bound = gateway.seen.is_set()
+        assert not gateway.errors and not gateway.closed, (gateway.errors, gateway.closed)
+        if not args.expect_blocked or bob.token:
+            assert present['push-first'], 'first message was not durably delivered before push'
+        if args.expect_blocked:
+            assert not present['push-second'], 'old blocking behavior was not reproduced'
+            assert not acknowledgements, acknowledgements
+            assert slot(root, alice)['handled'] == before, 'first message marked handled before blocked push'
+        else:
+            assert present['push-second'], 'second message blocked behind external push'
+            assert acknowledgements == [alice.inputs], acknowledgements
+            assert slot(root, alice)['handled'] == alice.inputs
+        assert not alice.errors, alice.errors
+        assert gateway.seen.is_set(), f'no real post-TLS push attempt: {gateway.errors}'
+        assert not gateway.errors and not gateway.closed, (gateway.errors, gateway.closed)
+        if not args.expect_blocked:
+            assert gateway.pending_at_stream['push-first'], 'push started before first durable recipient delivery'
+        elif not bob.token:
+            assert not present['push-first'], 'old offline ordering bug was not reproduced'
+            assert not gateway.pending_at_stream['push-first']
+        return {'status': 'reproduced-blocking' if args.expect_blocked else 'pass',
+                'recipient_state': args.recipient,
+                'durable_pending_at_gateway_stream': gateway.pending_at_stream,
+                'observation_seconds': args.bound, 'elapsed_seconds': time.perf_counter()-began,
+                'gateway_post_tls_stream_observed': gateway.seen.is_set(),
+                'gateway_post_tls_stream_at_bound': seen_at_bound,
+                'blocked_stage': 'post-tls-features' if gateway.seen.is_set() else 'external-route-before-tcp',
+                'durable_pending': present,
+                'sender_sm_ack_h': acknowledgements, 'sender_inputs': alice.inputs,
+                'sender_handled_before': before}
+    finally:
+        # Kill before closing the blackhole: releasing it could complete the old
+        # awaited route and would invalidate this bounded reproducer.
+        if server is not None:
+            server.kill()
+            server.wait(timeout=5)
+        listener.close()
+        await listener.wait_closed()
+        await gateway.close()
+        await asyncio.gather(*(client.close_raw() for client in clients), return_exceptions=True)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--binary', default='target/release/rust-xmpp')
+    parser.add_argument('--expect-blocked', action='store_true')
+    parser.add_argument('--recipient', choices=('hibernating', 'offline'), default='hibernating')
+    parser.add_argument('--bound', type=float, default=2)
+    parser.add_argument('--output')
+    args = parser.parse_args()
+    assert 0 < args.bound < 10, 'observation must precede the 30-second federation timeout'
+    with tempfile.TemporaryDirectory(prefix='xmpp-push-latency-') as directory:
+        root = pathlib.Path(directory)
+        binary = root/'rust-xmpp'
+        shutil.copy2(pathlib.Path(args.binary).resolve(), binary)
+        certificates(root)
+        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', 'cloud_notify', 's2s'],
+            'ssl': {'key': str(root/'key.pem'), 'certificate': str(root/'cert.pem')}},
+            'hosts': {DOMAIN: {'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)
+        with (root/'server.log').open('w+') as log:
+            try:
+                report = asyncio.run(exercise(root, binary, config, port, args, log))
+            except BaseException:
+                log.seek(0)
+                print(log.read())
+                raise
+        report['binary_version'] = subprocess.run([str(binary), '--version'],
+            check=True, capture_output=True, text=True).stdout.strip()
+        report['binary_sha256'] = hashlib.sha256(binary.read_bytes()).hexdigest()
+        report['scope'] = 'loopback TLS fixture; stalled external XMPP push service; no APNs/FCM'
+        result = json.dumps(report, indent=2)+'\n'
+        if args.output:
+            pathlib.Path(args.output).write_text(result)
+        print(result, end='')
+
+
+if __name__ == '__main__':
+    main()

tests/session_routing_isolation.rs

@@ -0,0 +1,62 @@
+use rust_xmpp::sessions::Hub;
+use std::{sync::mpsc, thread, time::Duration};
+use tokio::sync::mpsc as async_mpsc;
+
+#[test]
+fn unrelated_busy_session_does_not_block_account_routing_or_carbons_selection() {
+    let hub = Hub::default();
+    let (a_tx, _a_rx) = async_mpsc::channel(8);
+    let (b_tx, mut b_rx) = async_mpsc::channel(8);
+    let a = hub
+        .bind("alice@x/r", "a", a_tx, Duration::from_secs(30))
+        .unwrap();
+    let b = hub
+        .bind("bob@x/r", "b", b_tx, Duration::from_secs(30))
+        .unwrap();
+    b.lock().unwrap().available = true;
+    let held = a.lock().unwrap();
+    let (done, result) = mpsc::channel();
+    let worker = thread::spawn(move || {
+        assert_eq!(hub.message_targets("bob@x").len(), 1);
+        assert_eq!(hub.for_owners(&["bob@x", "bob@x"]).len(), 1);
+        assert_eq!(hub.deliver("bob@x", "bare"), 1);
+        assert_eq!(hub.deliver("bob@x/r", "full"), 1);
+        done.send(()).unwrap();
+    });
+    let completed_while_locked = result.recv_timeout(Duration::from_secs(2)).is_ok();
+    drop(held);
+    worker.join().unwrap();
+    assert!(
+        completed_while_locked,
+        "routing waited on unrelated account's session lock"
+    );
+    assert_eq!(b_rx.try_recv().unwrap(), "bare");
+    assert_eq!(b_rx.try_recv().unwrap(), "full");
+}
+
+#[test]
+fn duplicate_acknowledgement_does_not_rewrite_durable_slot() {
+    let dir = tempfile::tempdir().unwrap();
+    let hub = Hub::persistent(dir.path()).unwrap();
+    let (tx, _rx) = async_mpsc::channel(8);
+    let session = hub
+        .bind("bob@x/r", "b", tx, Duration::from_secs(30))
+        .unwrap();
+    let mut state = session.lock().unwrap();
+    state.enable(true).unwrap();
+    let path = dir
+        .path()
+        .join(format!("{}.json", state.resume_id.as_deref().unwrap()));
+    let marker = std::time::UNIX_EPOCH + Duration::from_secs(1_000_000);
+    std::fs::File::options()
+        .write(true)
+        .open(&path)
+        .unwrap()
+        .set_times(std::fs::FileTimes::new().set_modified(marker))
+        .unwrap();
+    state.ack_requested = true;
+    state.acknowledge(0).unwrap();
+    assert!(!state.ack_requested);
+    assert_eq!(std::fs::metadata(path).unwrap().modified().unwrap(), marker);
+    assert!(state.acknowledge(1).is_err());
+}