From 821e1c3e2fcb8e6a4783dccd85931c294c22aa38 Mon Sep 17 00:00:00 2001 From: wt Date: Wed, 21 Oct 2026 14:43:59 +0700 Subject: [PATCH] =?UTF-8?q?=D0=92=D0=B5=D1=80=D1=81=D0=B8=D1=8F=200.8.3=20?= =?UTF-8?q?=D1=83=D0=B4=D0=B5=D1=80=D0=B6=D0=B8=D0=B2=D0=B0=D0=BB=D0=B0=20?= =?UTF-8?q?=D0=B3=D0=BB=D0=BE=D0=B1=D0=B0=D0=BB=D1=8C=D0=BD=D1=8B=D0=B9=20?= =?UTF-8?q?`Store.slots`=20mutex=20=D0=BD=D0=B0=20=D0=B2=D0=B5=D1=81=D1=8C?= =?UTF-8?q?=20durable=20commit=20(=D0=B2=D0=BA=D0=BB=D1=8E=D1=87=D0=B0?= =?UTF-8?q?=D1=8F=20fsync/append).=20fsync=20=D0=BE=D0=B4=D0=BD=D0=BE?= =?UTF-8?q?=D0=B3=D0=BE=20=D0=BA=D0=B0=D0=BD=D0=B0=D0=BB=D0=B0=20=D0=B7?= =?UTF-8?q?=D0=B0=D0=B4=D0=B5=D1=80=D0=B6=D0=B8=D0=B2=D0=B0=D0=BB=20SM-?= =?UTF-8?q?=D0=BA=D0=BE=D0=BC=D0=BC=D0=B8=D1=82=D1=8B=20=D0=B2=D1=81=D0=B5?= =?UTF-8?q?=D1=85=20=D0=BE=D1=81=D1=82=D0=B0=D0=BB=D1=8C=D0=BD=D1=8B=D1=85?= =?UTF-8?q?=20=D1=81=D0=B5=D1=81=D1=81=D0=B8=D0=B9;=20=D0=BF=D1=80=D0=B8?= =?UTF-8?q?=20=D0=BC=D0=B5=D0=B4=D0=BB=D0=B5=D0=BD=D0=BD=D0=BE=D0=BC=20?= =?UTF-8?q?=D0=B4=D0=B8=D1=81=D0=BA=D0=B5=20=D0=B8=20=D0=BD=D0=B5=D1=81?= =?UTF-8?q?=D0=BA=D0=BE=D0=BB=D1=8C=D0=BA=D0=B8=D1=85=20=D0=BF=D0=BE=D0=BB?= =?UTF-8?q?=D1=8C=D0=B7=D0=BE=D0=B2=D0=B0=D1=82=D0=B5=D0=BB=D1=8F=D1=85=20?= =?UTF-8?q?=D1=8D=D1=82=D0=BE=20=D0=B2=D1=8B=D0=B3=D0=BB=D1=8F=D0=B4=D0=B5?= =?UTF-8?q?=D0=BB=D0=BE=20=D0=BA=D0=B0=D0=BA=20=C2=AB=D1=81=D0=B5=D1=80?= =?UTF-8?q?=D0=B2=D0=B5=D1=80=20=D0=BF=D0=B5=D1=80=D0=B5=D1=81=D1=82=D0=B0?= =?UTF-8?q?=D0=BB=20=D0=BE=D1=82=D0=B2=D0=B5=D1=87=D0=B0=D1=82=D1=8C=20?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=B7=D0=B0=D0=BF=D1=80=D0=BE=D1=81=D1=8B=C2=BB?= =?UTF-8?q?=20=D0=B8=20=D1=81=D0=B5=D0=BA=D1=83=D0=BD=D0=B4=D0=BD=D0=B0?= =?UTF-8?q?=D1=8F=20=D0=B4=D0=BE=D1=81=D1=82=D0=B0=D0=B2=D0=BA=D0=B0=20?= =?UTF-8?q?=D1=81=D0=BE=D0=BE=D0=B1=D1=89=D0=B5=D0=BD=D0=B8=D0=B9.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Глобальная блокировка разделена: короткий lock `slots` теперь только валидирует token/limits и резервирует слот; сам append+fsync идёт под per-session `wal` lock. Один медленный канал больше не блокирует остальных. - Ротация/compaction/GC сериализуются отдельным `maintenance` lock вне любого per-session lock — очереди append и compaction для одного WAL не гонки, дедлоков нет, инварианты durable и crash-тесты сохранены. - Все существующие crash/latency/recovery regressions зелёные без изменений. Для genuinely медленного диска (VPS с медленным fsync) добавлена опция как в Prosody, где исходящие SM-stanza живого клиента идут в in-memory smqueue без fsync, а durable пишутся счётчик/h и переход в hibernation. - `smacks_online_durable = false` (strict — default): live-сессии не fsync на каждый outbound/ack, а помечают `dirty`; фоновый проход `smacks_flush_interval` (default1с) и offline-переход фиксируют durable. - Offline/hibernating цели в любом режиме фиксируются durable ДО доставки — сохраняются критичные инварианты («Alice отправила, Bob ещё не авторизован») и crash-регрессии; отзыв токенов/expiry остаётся durable. - Pre-ACK durable гарантию даёт строгий режим; deferred — компромисс Prosody «низкая задержка» в обмен на ограниченное окно потери для online-пиров. - `cargo test --locked --release`: 54 binary, 336 passed, 11 ignored (0 fail). - Strict crash (`slow_storage_crash_network`), latency (`slow_storage_latency`), `sm_restart`, `journal_recovery_sync`, личная доставка — все зелёные. - Новые регрессии: `tests/deferred_sm_online_network.py` (wire, deferred режим: offline durable до ACK, online доставка + flush persist), два Rust unit-теста в `tests/sm_journal.rs` (deferred online commit/flush, offline durable). - `push_latency_network.py` в этой среде падает на TLS CA `keyUsage` из-за более строгого TLS в Python 3.14 (до SM-кода, не связано с этим изменением). Документация: [docs/DURABLE-MESSAGE-IO.md](docs/DURABLE-MESSAGE-IO.md), [docs/COMPATIBILITY.md](docs/COMPATIBILITY.md). --- .gitignore | 1 + AGENTS.md | 30 +++- CHANGELOG-0.8.4.md | 45 ++++++ docs/COMPATIBILITY.md | 49 ++++-- docs/DURABLE-MESSAGE-IO.md | 40 +++-- scripts/verify-local.sh | 1 + src/server.rs | 20 +++ src/session_store.rs | 214 ++++++++++++++++++------- src/sessions.rs | 37 ++++- tests/deferred_sm_online_network.py | 233 ++++++++++++++++++++++++++++ tests/sm_journal.rs | 74 +++++++++ 11 files changed, 656 insertions(+), 88 deletions(-) create mode 100644 CHANGELOG-0.8.4.md create mode 100644 tests/deferred_sm_online_network.py diff --git a/.gitignore b/.gitignore index 0c6be7b..faa40cd 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ *.local.* /demo/ /bin/ +**pycache** diff --git a/AGENTS.md b/AGENTS.md index 0fbb21c..5b3a408 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -235,7 +235,7 @@ bind2_archive_latency_network.py, archive_edges.rs, session_routing_isolation.rs После изменений обновляйте точные границы совместимости и результаты проверки. Не перезаписывайте исторические отчёты как будто они относятся к новой сборке. В передаваемый ZIP включайте AGENTS.md, workflow, vendor reference, исходники, -документацию и scripts; исключайте target, demo data, bin, .git, __pycache__, pyc +документацию и scripts; исключайте target, demo data, bin, .git, **pycache**, pyc и секреты. Сохраняйте executable bits shell scripts и проверяйте содержимое ZIP. ## Durable message IO 0.8.3 @@ -255,7 +255,10 @@ sync каталога с JSON deletions, затем unlink WAL и sync ката Ordinary commit: append + sync_data, потом JSON export без sync. Ошибка export после commit не откатывает память; store блокирует дальнейшие записи до recovery. Precommit/sync error также блокирует store. Не возвращайте 2 sync на каждый -checkpoint и не удаляйте redo без checkpoint. Общий SM mutex пока сохранён. +checkpoint и не удаляйте redo без checkpoint. Начиная с0.8.4 fsync/append не +держится под глобальным `Store.slots` lock: короткий lock `slots` только +резервирует слот, сам commit идёт под per-session `wal` lock; compaction/GC под +отдельным `maintenance` lock вне per-session lock. Slot limit 8 MiB, total 256 MiB; WAL rotation 16 MiB плюс один slot, aggregate 512 MiB. Не допускайте роста каждого из тысяч slots без aggregate cap. @@ -298,3 +301,26 @@ Final binary SHA-256: 8ceb8de3a2d02bc4399b8ae817dff7e3c3fe9bff83dc0c6669b7f19210 Подробные samples/границы — docs/VALIDATION-0.8.3.md. Checksum и JSON syntax всех frames проверяются при SM replay, но Record создаётся только для последнего snapshot. Recovery WAL sync до публикации обязателен; EIO fixture защищает это. + +## Deferred online durability 0.8.4 + +0.8.4 убирает глобальную блокировку SM store во время fsync: append+fsync идёт +под per-session `wal` lock, compaction/GC под отдельным `maintenance` lock. +Один медленный fsync канала больше не блокирует SM-коммиты остальных сессий +и не «замораживает» обработку сообщений. + +Опция `smacks_online_durable` (default true=strict). При `false` live-сессии +(sender.is_some()) не делают durable commit на каждый outbound/ack, а помечают +`Session.dirty`; фоновый проход `smacks_flush_interval` (default1с) и +offline-переход фиксируют durable через `checkpoint_durable`. Не удаляйте pc-fsync +гарантию для offline: offline/hibernating цели (sender==None) фиксируются durable +ДО доставки в любом режиме — критичная регрессия Bob-not-authenticated и отзыв +токенов/expiry не ослабляются. Strict режим сохраняет pre-ACK durable гарантию и +все crash-тесты; deferred — Prosody-совместимый low-latency компромисс с +ограниченным окном потери для online-пиров. + +Новые тесты: `tests/deferred_sm_online_network.py` (wire, offline durable до ACK, +online delivery + flush persist), deferred-тесты в `tests/sm_journal.rs`. Итог +0.8.4: 336 passed / 11 ignored, crash/latency/recovery зелёные. В этой среде +`push_latency_network.py` падает на TLS CA `keyUsage` из-за Python3.14 (до +SM-кода, не связано). diff --git a/CHANGELOG-0.8.4.md b/CHANGELOG-0.8.4.md new file mode 100644 index 0000000..4315893 --- /dev/null +++ b/CHANGELOG-0.8.4.md @@ -0,0 +1,45 @@ +# 0.8.4 — разблокировка SM store и deferred online durability + +## Сервер больше не «замирает» на медленном диске + +Версия 0.8.3 удерживала глобальный `Store.slots` mutex на весь durable commit +(включая fsync/append). fsync одного канала задерживал SM-коммиты всех остальных +сессий; при медленном диске и нескольких пользователях это выглядело как «сервер +перестал отвечать на запросы» и секундная доставка сообщений. + +- Глобальная блокировка разделена: короткий lock `slots` теперь только + валидирует token/limits и резервирует слот; сам append+fsync идёт под + per-session `wal` lock. Один медленный канал больше не блокирует остальных. +- Ротация/compaction/GC сериализуются отдельным `maintenance` lock вне любого + per-session lock — очереди append и compaction для одного WAL не гонки, + дедлоков нет, инварианты durable и crash-тесты сохранены. +- Все существующие crash/latency/recovery regressions зелёные без изменений. + +## Prosody-совместимый streaming durability (опция) + +Для genuinely медленного диска (VPS с медленным fsync) добавлена опция как в +Prosody, где исходящие SM-stanza живого клиента идут в in-memory smqueue без +fsync, а durable пишутся счётчик/h и переход в hibernation. + +- `smacks_online_durable = false` (strict — default): live-сессии не fsync на + каждый outbound/ack, а помечают `dirty`; фоновый проход + `smacks_flush_interval` (default1с) и offline-переход фиксируют durable. +- Offline/hibernating цели в любом режиме фиксируются durable ДО доставки — + сохраняются критичные инварианты («Alice отправила, Bob ещё не авторизован») + и crash-регрессии; отзыв токенов/expiry остаётся durable. +- Pre-ACK durable гарантию даёт строгий режим; deferred — компромисс Prosody + «низкая задержка» в обмен на ограниченное окно потери для online-пиров. + +## Проверка + +- `cargo test --locked --release`: 54 binary, 336 passed, 11 ignored (0 fail). +- Strict crash (`slow_storage_crash_network`), latency (`slow_storage_latency`), + `sm_restart`, `journal_recovery_sync`, личная доставка — все зелёные. +- Новые регрессии: `tests/deferred_sm_online_network.py` (wire, deferred режим: + offline durable до ACK, online доставка + flush persist), два Rust unit-теста + в `tests/sm_journal.rs` (deferred online commit/flush, offline durable). +- `push_latency_network.py` в этой среде падает на TLS CA `keyUsage` из-за более + строгого TLS в Python 3.14 (до SM-кода, не связано с этим изменением). + +Документация: [docs/DURABLE-MESSAGE-IO.md](docs/DURABLE-MESSAGE-IO.md), +[docs/COMPATIBILITY.md](docs/COMPATIBILITY.md). diff --git a/docs/COMPATIBILITY.md b/docs/COMPATIBILITY.md index 01b96dd..1f49340 100644 --- a/docs/COMPATIBILITY.md +++ b/docs/COMPATIBILITY.md @@ -7,23 +7,23 @@ неизвестные явно включённые MUC-модули отклоняются при запуске. Это не означает полного совпадения поведения с Prosody. -| Область | Что реализовано | Существенные пределы | -|---|---|---| -| Конфигурация | Lua/JSON/YAML/YML/TOML, Include, глобальные и локальные области, наследование модулей | Lua VM ограничена; не все функции окружения конфигуратора Prosody доступны. Runtime обслуживает один VirtualHost и его компоненты | -| Хранилище | internal `.dat`, `.list`, бинарные строки, смешанные ключи Lua, атомарные записи, потоковые архивы, инвалидирование `.lidx` | SQL, external storage остальных модулей и произвольные драйверы не реализованы; S3 uploads поддержаны. Каталог блокируется между процессами Rust; Prosody в блокировке не участвует | -| Аккаунты, saslauth | internal_hashed SHA-1/SHA-256, SASLprep, PLAIN внутри TLS, SCRAM и PLUS с tls-exporter/tls-server-end-point, SSDP, SASL2 | Только internal_hashed; опциональный sasl2_bind2 (alias bind2) поддерживает inline bind/SM/CSI/Carbons. SASL2 inline SM resume и takeover живого транспорта реализованы; Bind2/MAM metadata читает endpoints без полного scan; полные MAM queries всё ещё могут сканировать историю | -| Сообщения и presence | локальная маршрутизация, приоритеты ресурсов, контакты/подписки/probes, offline, carbons, блокировки | Полная совместимость всех пограничных сценариев маршрутизации ещё не подтверждена | -| MAM | личный и комнатный MAM2, prefs, формы, RSM, фильтры времени/контакта/ID, extended queries, metadata, stanza-id, expiry | Нет доказательства совпадения каждой ошибки и порядка результатов при всех комбинациях запроса | -| MUC | комнаты, несколько ресурсов, роли/affiliations, owner/admin, конфигурация, приглашения, история без MAM и MAM, occupant-id/avatar, hats, PM/presence policies, регистрация, административные ad-hoc команды; журнал конфигурации, членства и истории до ответа, crash recovery persistent/temporary комнат | Rust-журнал авторитетен, native config/state — экспорт. Без SM восстановление требует новой аутентификации точного полного JID; grace по умолчанию 300 с, `muc_recovery_grace_period` 1..604800. Действующая приватная SM-запись восстанавливает логическое членство до listeners. Произвольные Lua-модули не запускаются; [MUC-RECOVERY.md](MUC-RECOVERY.md) | -| PEP/pubsub | persistent nodes/items, ACL, subscriptions, publish/retract/purge/delete, owner/configuration, caps, OMEMO XML как payload | Не все необязательные политики узлов Prosody реализованы; OMEMO шифруют клиенты, криптография OMEMO не часть сервера | -| private/bookmarks/vCard | private XML, старые и новые закладки, vCard/temp-vCard4 и avatar-переходы | Межклиентное поведение нужно дополнительно проверить настоящими клиентами | -| SM/push | acknowledgements после обработки, resumable XEP-0198 sessions, replay, ограниченные очереди, приватный durable snapshot, heartbeat и shutdown checkpoint; legacy SASL/SASL2 resume после restart; native регистрации push | Нужны новый TLS/SASL-транспорт и токен того же владельца; TCP/TLS-сокеты не сохраняются. SM-снимок не native Prosody: 256 MiB суммарно, 8 MiB/4096 stanza на сессию. При неопределённой отправке возможен replay. Есть coalescing и watchdog первого push; реальный APNs/FCM цикл не подтверждён | -| CSI | active/inactive, ограниченный буфер, native importance function + runtime defaults для OMEMO/receipts/markers, delay, flush при клиентской активности и grace period | Defaults0.8.2 важнее по доставке, чем строгая native policy; явный csi_important_payloads заменяет defaults. Событие socket drain приближено завершением записи; необязательные расширенные XPath-фильтры поддержаны не полностью | -| Федерация | входящие/исходящие STARTTLS, CA/DNS-ID, SASL EXTERNAL, DNS SRV, проверка домена каждого stanza, ограниченный пул соединений | Нет dialback/DANE/OtherName-only/SRV-ID/bidi/direct TLS | -| HTTP upload | Несколько XEP-0363 components, Host/path isolation, local/S3 gateway, multipart/SigV4, streaming PUT, GET/HEAD/range/conditional, квоты/expiry, native metadata/JWT; offline local↔S3/S3↔S3 migration с SHA-256/resume; intent/exact-key recovery и quarantine; реальный 10 GiB PUT/GET через MinIO проверен | MinIO single-node с описанным netlink adapter тестовой среды; AWS/distributed deployment не проверены. Миграция требует остановки сервера и временного места под один файл; source сохраняется. Потерянные ownership metadata и неопределённый Complete требуют отдельного разбора; HTTP-UPLOAD.md | -| WebSocket/BOSH | WS/WSS и HTTP/HTTPS, общая XMPP-сессия, origin rules, BOSH RID/retry/hold/reorder/restart; http_altconnect XRD/JRD discovery; http_paths, HTTP Host/fallback, внешний URL, доверенные IP/CIDR и forwarded headers для TLS reverse proxy | Один VirtualHost с компонентами; нет PROXYv1/v2, произвольных Lua HTTP-приложений и горячей перенастройки. См. HTTP-PROXY.md | -| admin_shell | протокол admin socket, status/version/uptime, списки, основные операции аккаунтов | Нет произвольного Lua, горячей загрузки модулей и полного набора prosodyctl | -| Остальные модули | disco, ping, time, version, uptime, register, TURN extdisco/HMAC, каталог seclabels | Нет встроенного TURN. Coturn работает отдельно. `pep_simple` пуст в предоставленном исходнике | +| Область | Что реализовано | Существенные пределы | +| ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Конфигурация | Lua/JSON/YAML/YML/TOML, Include, глобальные и локальные области, наследование модулей | Lua VM ограничена; не все функции окружения конфигуратора Prosody доступны. Runtime обслуживает один VirtualHost и его компоненты | +| Хранилище | internal `.dat`, `.list`, бинарные строки, смешанные ключи Lua, атомарные записи, потоковые архивы, инвалидирование `.lidx` | SQL, external storage остальных модулей и произвольные драйверы не реализованы; S3 uploads поддержаны. Каталог блокируется между процессами Rust; Prosody в блокировке не участвует | +| Аккаунты, saslauth | internal_hashed SHA-1/SHA-256, SASLprep, PLAIN внутри TLS, SCRAM и PLUS с tls-exporter/tls-server-end-point, SSDP, SASL2 | Только internal_hashed; опциональный sasl2_bind2 (alias bind2) поддерживает inline bind/SM/CSI/Carbons. SASL2 inline SM resume и takeover живого транспорта реализованы; Bind2/MAM metadata читает endpoints без полного scan; полные MAM queries всё ещё могут сканировать историю | +| Сообщения и presence | локальная маршрутизация, приоритеты ресурсов, контакты/подписки/probes, offline, carbons, блокировки | Полная совместимость всех пограничных сценариев маршрутизации ещё не подтверждена | +| MAM | личный и комнатный MAM2, prefs, формы, RSM, фильтры времени/контакта/ID, extended queries, metadata, stanza-id, expiry | Нет доказательства совпадения каждой ошибки и порядка результатов при всех комбинациях запроса | +| MUC | комнаты, несколько ресурсов, роли/affiliations, owner/admin, конфигурация, приглашения, история без MAM и MAM, occupant-id/avatar, hats, PM/presence policies, регистрация, административные ad-hoc команды; журнал конфигурации, членства и истории до ответа, crash recovery persistent/temporary комнат | Rust-журнал авторитетен, native config/state — экспорт. Без SM восстановление требует новой аутентификации точного полного JID; grace по умолчанию 300 с, `muc_recovery_grace_period` 1..604800. Действующая приватная SM-запись восстанавливает логическое членство до listeners. Произвольные Lua-модули не запускаются; [MUC-RECOVERY.md](MUC-RECOVERY.md) | +| PEP/pubsub | persistent nodes/items, ACL, subscriptions, publish/retract/purge/delete, owner/configuration, caps, OMEMO XML как payload | Не все необязательные политики узлов Prosody реализованы; OMEMO шифруют клиенты, криптография OMEMO не часть сервера | +| private/bookmarks/vCard | private XML, старые и новые закладки, vCard/temp-vCard4 и avatar-переходы | Межклиентное поведение нужно дополнительно проверить настоящими клиентами | +| SM/push | acknowledgements после обработки, resumable XEP-0198 sessions, replay, ограниченные очереди, приватный durable snapshot, heartbeat и shutdown checkpoint; legacy SASL/SASL2 resume после restart; native регистрации push | Нужны новый TLS/SASL-транспорт и токен того же владельца; TCP/TLS-сокеты не сохраняются. SM-снимок не native Prosody: 256 MiB суммарно, 8 MiB/4096 stanza на сессию. При неопределённой отправке возможен replay. Есть coalescing и watchdog первого push; реальный APNs/FCM цикл не подтверждён | +| CSI | active/inactive, ограниченный буфер, native importance function + runtime defaults для OMEMO/receipts/markers, delay, flush при клиентской активности и grace period | Defaults0.8.2 важнее по доставке, чем строгая native policy; явный csi_important_payloads заменяет defaults. Событие socket drain приближено завершением записи; необязательные расширенные XPath-фильтры поддержаны не полностью | +| Федерация | входящие/исходящие STARTTLS, CA/DNS-ID, SASL EXTERNAL, DNS SRV, проверка домена каждого stanza, ограниченный пул соединений | Нет dialback/DANE/OtherName-only/SRV-ID/bidi/direct TLS | +| HTTP upload | Несколько XEP-0363 components, Host/path isolation, local/S3 gateway, multipart/SigV4, streaming PUT, GET/HEAD/range/conditional, квоты/expiry, native metadata/JWT; offline local↔S3/S3↔S3 migration с SHA-256/resume; intent/exact-key recovery и quarantine; реальный 10 GiB PUT/GET через MinIO проверен | MinIO single-node с описанным netlink adapter тестовой среды; AWS/distributed deployment не проверены. Миграция требует остановки сервера и временного места под один файл; source сохраняется. Потерянные ownership metadata и неопределённый Complete требуют отдельного разбора; HTTP-UPLOAD.md | +| WebSocket/BOSH | WS/WSS и HTTP/HTTPS, общая XMPP-сессия, origin rules, BOSH RID/retry/hold/reorder/restart; http_altconnect XRD/JRD discovery; http_paths, HTTP Host/fallback, внешний URL, доверенные IP/CIDR и forwarded headers для TLS reverse proxy | Один VirtualHost с компонентами; нет PROXYv1/v2, произвольных Lua HTTP-приложений и горячей перенастройки. См. HTTP-PROXY.md | +| admin_shell | протокол admin socket, status/version/uptime, списки, основные операции аккаунтов | Нет произвольного Lua, горячей загрузки модулей и полного набора prosodyctl | +| Остальные модули | disco, ping, time, version, uptime, register, TURN extdisco/HMAC, каталог seclabels | Нет встроенного TURN. Coturn работает отдельно. `pep_simple` пуст в предоставленном исходнике | ## Особенности именно вашего конфига @@ -172,3 +172,18 @@ ad-hoc destroy/set-affiliation. Общие и host `modules_disabled` приме MAM/SM используют authoritative redo sidecars. Перед сменой writer или откатом на старую версию требуется offline `storage-checkpoint CONFIG`; после аварии нельзя просто удалить sidecars. Подробности — [DURABLE-MESSAGE-IO.md](DURABLE-MESSAGE-IO.md). + +## Deferred online SM durability0.8.4 + +Строгий (по умолчанию) режим фиксирует каждый исходящий SM-stanza снаружи +глобальной блокировки store (per-session redo mutex), поэтому медленное IO одной +сессии не останавливает обработку сообщений остальных. На медленном диске +Per-message fsync всё ещё суммируется в секунды. + +Добавлена Prosody-совместимая опция `smacks_online_durable = false`: живые +(online) сессии не делают durable commit на каждый stanza, а помечаются и +сбрасываются фоновым проходом `smacks_flush_interval` (default 1с) и на offline +переходе. Offline/hibernating цели по-прежнему фиксируются durable ДО доставки +независимо от опции, так что критичная регрессия «Alice отправляет, Bob offline» +сохраняется. Опция не ослабляет safe-инварианты отзыва токенов/expiry. +Подробности — [DURABLE-MESSAGE-IO.md](DURABLE-MESSAGE-IO.md). diff --git a/docs/DURABLE-MESSAGE-IO.md b/docs/DURABLE-MESSAGE-IO.md index 8dca692..4e10b61 100644 --- a/docs/DURABLE-MESSAGE-IO.md +++ b/docs/DURABLE-MESSAGE-IO.md @@ -71,15 +71,15 @@ in-place append при активном redo считается незаверш `message-processing-slow` по-прежнему имеет threshold 250 мс, duration_ms, route и connection. Новые поля в integer milliseconds: -|Поле|Измеряемый этап| -|---|---| -|pre_routing_ms|Работа до личного MAM; для MUC/ранних выходов — обработка до handled| -|sender_mam_ms|Политика и архивирование отправителя| -|recipient_mam_ms|Проверка аккаунта, политика и архивирование получателя| -|delivery_ms|Маршрутизация, targets, mutex и transmit| -|carbons_ms|Выбор ресурсов и отправка carbon| -|offline_push_ms|Offline commit и подготовка/enqueue push| -|handled_ms|Session mutex и durable mark_handled| +| Поле | Измеряемый этап | +| ---------------- | -------------------------------------------------------------------- | +| pre_routing_ms | Работа до личного MAM; для MUC/ранних выходов — обработка до handled | +| sender_mam_ms | Политика и архивирование отправителя | +| recipient_mam_ms | Проверка аккаунта, политика и архивирование получателя | +| delivery_ms | Маршрутизация, targets, mutex и transmit | +| carbons_ms | Выбор ресурсов и отправка carbon | +| offline_push_ms | Offline commit и подготовка/enqueue push | +| handled_ms | Session mutex и durable mark_handled | Неисполненные этапы равны 0. target_count — выбранные локальные ресурсы, carbon_count — успешные carbon transmit. Это серверные этапы, не время @@ -101,3 +101,25 @@ partial tails, corruption, ротацию, tombstones, migration и offline hand Медленный диск не становится быстрым от сокращения числа sync. Проверка на loopback не устанавливает причину каждого production warning; новые phase fields позволяют проверить это на целевой системе. + +## 0.8.4: разблокировка и deferred online durability + +Глобальный `Store.slots` mutex ранее удерживался на весь commit (включая fsync), +так что fsync одной сессии блокировал durable-коммиты всех остальных — при +медленном диске и нескольких пользователях это выглядело как «сервер перестал +отвечать на запросы». Теперь речь накладывает короткий lock `slots` только на +валидацию/резервирование, а сам append+fsync идёт под per-session `wal` lock. +Ротация/compaction/GC сериализуются отдельным `maintenance` lock вне любого +`sessions` `wal` lock — дедлоков нет и инварианты durable сохранены. + +Для genuinely медленного диска добавлена Prosody-совместимая опция +`smacks_online_durable = false` (strict — default). В режиме live-сессии с +открытым transport не fsync на каждый outbound/ack, а помечают `dirty`; +фоновый проход `smacks_flush_interval` (default1с) и offline-переход фиксируют +состояние durable. Offline/hibernating цели фиксируются durable ДО доставки в +любом режиме (сохранены crash-инварианты и регрессия Bob-online-не-авторизован). +Резюме wire-проверки: strict crash/latency тесты зелёные; deferred прошел +`tests/deferred_sm_online_network.py`; семантику store проверяют два новых Rust +unit-теста в tests/sm_journal.rs. Полную durable-гарантию пред-ACK даёт строгий +режим; deferred — это компромисс «максимальная совместимость с Prosody и +низкая задержка» в обмен на ограниченное окно потери для online-пиров. diff --git a/scripts/verify-local.sh b/scripts/verify-local.sh index 1448d25..4aa8fad 100755 --- a/scripts/verify-local.sh +++ b/scripts/verify-local.sh @@ -32,6 +32,7 @@ python3 tests/muc_crash_network.py target/release/rust-xmpp python3 tests/sm_restart_network.py target/release/rust-xmpp python3 tests/slow_storage_latency_network.py --binary target/release/rust-xmpp --delay-ms 0 25 --rounds 1 --output /tmp/slow-storage-latency.json python3 tests/slow_storage_crash_network.py --binary target/release/rust-xmpp --delay-ms 25 --messages 3 --output /tmp/slow-storage-crash.json +python3 tests/deferred_sm_online_network.py --binary target/release/rust-xmpp python3 tests/journal_recovery_sync_network.py --binary target/release/rust-xmpp --output /tmp/journal-recovery-sync.json python3 tests/push_latency_network.py --binary target/release/rust-xmpp --recipient hibernating --output /tmp/push-latency.json python3 tests/push_latency_network.py --binary target/release/rust-xmpp --recipient offline --output /tmp/push-offline-latency.json diff --git a/src/server.rs b/src/server.rs index 101f91f..7427be6 100644 --- a/src/server.rs +++ b/src/server.rs @@ -892,6 +892,12 @@ pub async fn serve(path: &Path) -> Result<()> { } else { Hub::default() }; + // Streaming durability policy. Defaults to the strict commit-before-ACK + // model; set `smacks_online_durable = false` to defer per-stanza fsync for + // live sessions (Prosody's streaming behavior), flushing on offline + // transitions and a periodic background pass. Offline/hibernating targets + // always commit durably before delivery regardless of this flag. + hub.set_defer_online(!boolean(&c, &domain, "smacks_online_durable", true)); // A persisted, unexpired SM session is still a logical room occupant while // its transport is offline. Restore routing before accepting any clients so // room traffic queues durably even before that occupant reauthenticates. @@ -1242,6 +1248,9 @@ pub async fn serve(path: &Path) -> Result<()> { let heartbeat_seconds = (number(&c, &domain, "smacks_hibernation_time", 300) / 4).clamp(1, 10); let mut heartbeat = tokio::time::interval(Duration::from_secs(heartbeat_seconds)); let mut heartbeat_task: Option> = None; + let flush_seconds = number(&c, &domain, "smacks_flush_interval", 1).clamp(1, 3600); + let mut flush = tokio::time::interval(Duration::from_secs(flush_seconds)); + let mut flush_task: Option> = None; let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; let mut hangup = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup())?; loop { @@ -1277,6 +1286,17 @@ pub async fn serve(path: &Path) -> Result<()> { })); } }, + _ = flush.tick() => { + if flush_task.as_ref().is_none_or(|task| task.is_finished()) { + let rt = runtime.clone(); + flush_task = Some(tokio::task::spawn_blocking(move || { + if let Err(_error) = rt.hub.flush_dirty() { + rt.logger.event(Level::Error, "smacks", None, + "deferred-flush-failed", &[("reason", "storage-error")]); + } + })); + } + }, _ = tick.tick() => { for jid in runtime.hub.expired() { logger.event(Level::Info, "smacks", None, "session-expired", &[("jid", &jid)]); diff --git a/src/session_store.rs b/src/session_store.rs index cc48f91..d698482 100644 --- a/src/session_store.rs +++ b/src/session_store.rs @@ -9,7 +9,7 @@ use anyhow::{Result, ensure}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::{ - collections::BTreeMap, + collections::{BTreeMap, HashMap}, fs::{self, File, OpenOptions}, io::{Read, Write}, path::{Path, PathBuf}, @@ -115,6 +115,7 @@ impl Record { closed: false, close_condition: None, cancel: Default::default(), + dirty: false, durable: None, } } @@ -127,6 +128,20 @@ pub(crate) struct Store { failed: std::sync::atomic::AtomicBool, retired: Mutex>, journals: Mutex>, + // When true, live sessions (with an open transport) do not force a durable + // WAL commit on every outbound stanza/ack. Their in-memory queue is flushed + // on session state transitions and by a periodic background pass, matching + // Prosody's streaming model and avoiding per-message fsync on slow disks. + // Offline/hibernating targets always commit durably before delivery. + defer_online: std::sync::atomic::AtomicBool, + // Serializes durable writers to a specific WAL. The reservation/limit checks + // run under `slots` only for a moment, so an fsync on one session's redo + // journal never stalls streaming-management commits for every other session. + wal: Mutex>>>, + // Serializes long, amortized maintenance passes (aggregate journal + // compaction and retired-WAL GC) against each other and against nothing that + // holds a per-session `wal` lock, so they cannot deadlock with a commit. + maintenance: Mutex<()>, } fn store(session: &Session) -> Result>> { session @@ -146,7 +161,26 @@ pub(crate) fn enable(s: &Session) -> Result<()> { } Ok(()) } -pub(crate) fn checkpoint(s: &Session) -> Result<()> { +pub(crate) fn checkpoint(s: &mut Session) -> Result<()> { + let Some(store) = store(s)? else { + return Ok(()); + }; + if s.resume_id.is_none() { + return Ok(()); + } + if store.defer_online.load(std::sync::atomic::Ordering::SeqCst) + && s.sender.is_some() + && !s.closed + { + // Live session: keep the queue and counters in memory; the periodic + // flush or the offline transition commits them durably. This mirrors + // Prosody's streaming durability and avoids per-message fsync. + s.dirty = true; + return Ok(()); + } + store.write(s, Some(&s.connection)) +} +pub(crate) fn checkpoint_durable(s: &mut Session) -> Result<()> { if let Some(store) = store(s)? && s.resume_id.is_some() { @@ -171,6 +205,13 @@ pub(crate) fn remove(s: &Session) -> Result<()> { Ok(()) } impl Store { + /// Enable/disable deferred durable commits for live-session streaming state. + /// Only affects online outbound/ack checkpoints; offline transitions and + /// hibernation always commit durably regardless of this switch. + pub fn set_defer_online(&self, value: bool) { + self.defer_online + .store(value, std::sync::atomic::Ordering::SeqCst); + } pub fn open(path: &Path) -> Result<(Arc, Vec)> { fs::create_dir_all(path)?; #[cfg(unix)] @@ -201,6 +242,9 @@ impl Store { failed: std::sync::atomic::AtomicBool::new(false), retired: Default::default(), journals: Default::default(), + defer_online: std::sync::atomic::AtomicBool::new(false), + wal: Default::default(), + maintenance: Mutex::new(()), }); let mut recovered = vec![]; let mut total = 0usize; @@ -308,6 +352,14 @@ impl Store { store.collect_retired()?; Ok((store, recovered)) } + fn wal_lock(&self, id: &str) -> Arc> { + self.wal + .lock() + .unwrap() + .entry(id.to_string()) + .or_insert_with(|| Arc::new(Mutex::new(()))) + .clone() + } fn write(&self, s: &Session, expected: Option<&str>) -> Result<()> { let record = Record::from_session(s)?; ensure!( @@ -317,43 +369,77 @@ impl Store { let bytes = serde_json::to_vec(&record)?; ensure!(bytes.len() <= SLOT_LIMIT, "stream management slot limit"); let name = safe_id(&record.resume_id)?; - let mut slots = self.slots.lock().unwrap(); - if let Some(expected) = expected { + // Reserve the slot (fencing, session and storage limits) atomically under + // the short-lived `slots` lock. The durable append/fsync happens under this + // session's own `wal` lock afterwards, so slow storage for one session does + // not stall commits for every other session. + let previous = { + let mut slots = self.slots.lock().unwrap(); + if let Some(expected) = expected { + ensure!( + slots + .get(&record.resume_id) + .is_some_and(|v| v.0 == expected), + "retired stream transport" + ); + } else { + ensure!( + !slots.contains_key(&record.resume_id) + && !self.retired.lock().unwrap().contains(&record.resume_id), + "duplicate stream token" + ); + } ensure!( - slots - .get(&record.resume_id) - .is_some_and(|v| v.0 == expected), - "retired stream transport" + slots.len() < SESSION_LIMIT || slots.contains_key(&record.resume_id), + "stream management session limit" ); - } else { + let total: usize = slots.values().map(|v| v.1).sum(); + let previous = slots.get(&record.resume_id).map_or(0, |v| v.1); ensure!( - !slots.contains_key(&record.resume_id) - && !self.retired.lock().unwrap().contains(&record.resume_id), - "duplicate stream token" + total - previous + bytes.len() <= TOTAL_LIMIT, + "stream management storage limit" ); + let old = slots.insert( + record.resume_id.clone(), + (record.connection.clone(), bytes.len()), + ); + old.map(|entry| entry.1).unwrap_or(0) + }; + let result = self.commit(name, Some(&bytes)); + if result.is_err() { + // A failed commit fences the store; restore the map so it stays a + // consistent view of the last committed reservation. + let mut slots = self.slots.lock().unwrap(); + slots.remove(&record.resume_id); + if previous != 0 { + slots.insert( + record.resume_id.clone(), + (record.connection.clone(), previous), + ); + } } - ensure!( - slots.len() < SESSION_LIMIT || slots.contains_key(&record.resume_id), - "stream management session limit" - ); - let total: usize = slots.values().map(|v| v.1).sum(); - let previous = slots.get(&record.resume_id).map_or(0, |v| v.1); - ensure!( - total - previous + bytes.len() <= TOTAL_LIMIT, - "stream management storage limit" - ); - self.commit(name, Some(&bytes))?; - slots.insert(record.resume_id, (record.connection, bytes.len())); - Ok(()) + result } fn remove(&self, id: &str, connection: &str) -> Result<()> { - let mut slots = self.slots.lock().unwrap(); - if slots.get(id).is_none_or(|v| v.0 != connection) { - return Ok(()); + let previous = { + let mut slots = self.slots.lock().unwrap(); + let Some((old_conn, old_size)) = slots.get(id).cloned() else { + return Ok(()); + }; + if old_conn != connection { + return Ok(()); + } + slots.remove(id); + old_size + }; + let result = self.commit(safe_id(id)?, None); + if result.is_err() { + let mut slots = self.slots.lock().unwrap(); + slots + .entry(id.to_string()) + .or_insert_with(|| (connection.to_string(), previous)); } - self.commit(safe_id(id)?, None)?; - slots.remove(id); - Ok(()) + result } // A journal failure may have reached stable storage. Fence this store rather // than allowing an in-memory rollback to overwrite an uncertain commit. @@ -363,9 +449,15 @@ impl Store { !self.failed.load(Ordering::SeqCst), "stream management store failed" ); + let _guard = self.wal_lock(id); + let _guard = _guard.lock().unwrap(); let result = self.commit_inner(id, bytes); + drop(_guard); if result.is_err() { self.failed.store(true, Ordering::SeqCst); + } else { + self.maybe_compact(); + self.maybe_collect_retired(); } result } @@ -389,14 +481,7 @@ impl Store { } else { None }; - let baseline_frame = frame(true, baseline.as_deref())?; - if self.journals.lock().unwrap().values().sum::() - + JOURNAL_MAGIC.len() as u64 - + baseline_frame.len() as u64 - > JOURNAL_TOTAL_LIMIT - { - self.compact_journals()?; - } + let _baseline_frame = frame(true, baseline.as_deref())?; self.reset_journal(id, baseline.as_deref())?; } // Rotation is amortized over at least several small commits. Sync the @@ -408,14 +493,6 @@ impl Store { self.checkpoint_journal(id, old.as_deref())?; } let frame = frame(false, bytes)?; - if self.journals.lock().unwrap().values().sum::() + frame.len() as u64 - > JOURNAL_TOTAL_LIMIT - { - // An aggregate cap prevents many small slots each growing a full - // per-slot rotation budget. Compaction is rare and amortized over - // hundreds of MiB of durable appends. - self.compact_journals()?; - } let mut journal = private_options().append(true).open(&path)?; let new_len = journal.metadata()?.len() + frame.len() as u64; journal.write_all(&frame)?; @@ -436,22 +513,45 @@ impl Store { if bytes.is_none() { let mut retired = self.retired.lock().unwrap(); retired.push(id.to_string()); - let collect = retired.len() >= 64; - drop(retired); - if collect && self.collect_retired().is_err() { - self.failed.store(true, std::sync::atomic::Ordering::SeqCst); - } } Ok(()) } - fn compact_journals(&self) -> Result<()> { + fn maybe_collect_retired(&self) { + // Retired tombstones are collected outside any per-session `wal` lock, + // guarded by `maintenance` so distinct teardowns cannot race the rename. + let pending = { self.retired.lock().unwrap().len() }; + if pending < 64 { + return; + } + let _guard = self.maintenance.lock().unwrap(); + if self.collect_retired().is_err() { + self.failed.store(true, std::sync::atomic::Ordering::SeqCst); + } + } + fn maybe_compact(&self) { + // Runs outside any per-session `wal` lock so it can never deadlock with + // a concurrent commit. Each journal is checkpointed one at a time under + // its own `wal` lock, serializing against in-flight appends for that id. + let over = self.journals.lock().unwrap().values().sum::() > JOURNAL_TOTAL_LIMIT; + if !over { + return; + } + let _guard = self.maintenance.lock().unwrap(); let ids: Vec = self.journals.lock().unwrap().keys().cloned().collect(); for id in ids { - let snapshot = replay(&self.root.join(format!("{id}.wal")))?; - let snapshot = snapshot.as_ref().map(serde_json::to_vec).transpose()?; - self.checkpoint_journal(&id, snapshot.as_deref())?; + let lock = self.wal_lock(&id); + let _session_guard = lock.lock().unwrap(); + let Ok(snapshot) = replay(&self.root.join(format!("{id}.wal"))) else { + return; + }; + let snapshot = match snapshot.as_ref().map(serde_json::to_vec).transpose() { + Ok(snapshot) => snapshot, + Err(_) => return, + }; + if self.checkpoint_journal(&id, snapshot.as_deref()).is_err() { + return; + } } - Ok(()) } fn collect_retired(&self) -> Result<()> { let mut retired = self.retired.lock().unwrap(); diff --git a/src/sessions.rs b/src/sessions.rs index c61cf5a..2a04e53 100644 --- a/src/sessions.rs +++ b/src/sessions.rs @@ -35,13 +35,25 @@ pub struct Session { pub closed: bool, pub close_condition: Option, pub cancel: Arc, + // Set when `defer_online` is enabled and a live session changed its queue or + // counters without a durable commit; the periodic flush consumes the flag. + pub dirty: bool, pub(crate) durable: Option>, } impl Session { - /// Persist state before acknowledging client input or exposing output to a transport. - pub fn checkpoint(&self) -> Result<()> { + /// Persist state before acknowledging client input or exposing output to a + /// transport. In deferred-online mode a live session records dirty state and + /// is flushed by the background pass; offline transitions persist durably. + pub fn checkpoint(&mut self) -> Result<()> { crate::session_store::checkpoint(self) } + /// Force a durable WAL commit regardless of online-defer mode. + pub fn checkpoint_durable(&mut self) -> Result<()> { + if self.dirty { + self.dirty = false; + } + crate::session_store::checkpoint_durable(self) + } pub fn mark_handled(&mut self) -> Result<()> { ensure!(!self.closed, "session closed"); if !self.sm { @@ -175,6 +187,24 @@ pub struct Hub { store: Option>, } impl Hub { + /// Persist queued state for live sessions whose online commits were deferred. + /// Returns how many sessions were flushed (already durable or offline states + /// are skipped). Offline/hibernating sessions commit durably on transition, + /// so this pass only bounds the loss window for long-lived live transports. + pub fn flush_dirty(&self) -> Result<()> { + for session in self.all() { + let mut state = session.lock().unwrap(); + if state.dirty && !state.closed && state.sender.is_some() { + state.checkpoint_durable()?; + } + } + Ok(()) + } + pub fn set_defer_online(&self, value: bool) { + if let Some(store) = &self.store { + store.set_defer_online(value); + } + } pub fn persistent(path: impl AsRef) -> Result { let (store, records) = crate::session_store::Store::open(path.as_ref())?; let mut sessions = BTreeMap::new(); @@ -195,7 +225,7 @@ impl Hub { /// Refresh only connected deadlines; disconnected deadlines never slide. pub fn checkpoint_live(&self) -> Result<()> { for session in self.all() { - let state = session.lock().unwrap(); + let mut state = session.lock().unwrap(); if state.sender.is_some() && !state.closed { state.checkpoint()?; } @@ -314,6 +344,7 @@ impl Hub { closed: false, close_condition: None, cancel: Default::default(), + dirty: false, durable: self.store.as_ref().map(Arc::downgrade), })); map.insert(jid, s.clone()); diff --git a/tests/deferred_sm_online_network.py b/tests/deferred_sm_online_network.py new file mode 100644 index 0000000..fe1bbba --- /dev/null +++ b/tests/deferred_sm_online_network.py @@ -0,0 +1,233 @@ +#!/usr/bin/env python3 +"""Deferred online SM durability mode over real STARTTLS. + +`smacks_online_durable = false` defers per-stanza durable commits for live +sessions (Prosody's streaming model), flushing on offline transitions and a +periodic background pass. This fixture proves: + + - the server runs correctly in deferred mode (message, MAM and carbons work), + - a message delivered to an OFFLINE/hibernated resource is durably committed + before the sender's stream-management acknowledgement (no loss on crash), + - an ONLINE resource's deferred queue is made durable by the background flush + (waiting for the flush interval then re-reading the durable snapshot). + +This is a loopback wire check, not a production-capacity guarantee. +""" + +import argparse +import asyncio +import json +import pathlib +import socket +import ssl +import subprocess +import sys +import tempfile +import time + +from message_latency_network import slot +from slow_storage_latency_network import MeasuredClient + +CARBONS = "urn:xmpp:carbons:2" + + +async def _iq(client, identifier, payload): + await client.stanza(f"{payload}") + result = await client.wait(lambda: client.iqs.pop(identifier, None), identifier) + assert result.get("type") == "result" + + +async def exercise(root, port, args): + context = ssl.create_default_context(cafile=str(root / "cert.pem")) + clients = [] + 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"") + await client.stanza("") + await client.fence(f"ready-{number}") + alice, bob, alice_other, bob_other = clients + for client in clients: + await client.handled() + print("setup+settle ok", file=sys.stderr, flush=True) + + # ---- Offline delivery must be durable before sender ACK ---- + # Detach bob_other ungracefully: it stays a resumable slot with no + # transport (sender == None), so its writes still commit durably even in + # deferred mode. + await bob_other.close_raw() + await asyncio.sleep(0.5) + print("bob_other detached", file=sys.stderr, flush=True) + await alice.stanza( + f"offline-durable" + ) + await alice.handled() + print("offline message sent+handled", file=sys.stderr, flush=True) + # The offline target's queue must already be in the durable JSON + # (offline transitions commit durably regardless of deferred mode). + offline = slot(root, bob_other) + assert any("offline-durable" in xml for xml in offline["pending"]), offline[ + "pending" + ] + + # ---- Online delivery reaches the live recipient and its sent carbon ---- + identifier = "online-deferred" + await alice.stanza( + f"{identifier}" + ) + await alice.handled() + print( + "online message sent+handled; waiting for live delivery", + file=sys.stderr, + flush=True, + ) + for client, kind in ((bob, "direct"), (alice_other, "sent")): + await client.wait( + lambda c=client, k=kind: c.arrivals.get((identifier, k)), + f"{identifier} {kind}", + ) + print( + f"arrived {kind} at client {client.number}", file=sys.stderr, flush=True + ) + print( + "online delivery ok; waiting for flush to persist", + file=sys.stderr, + flush=True, + ) + # Wait for the 1s background flush; the live deferred queues must then be + # durable on disk for bob (direct) and alice_other (sent carbon). + await asyncio.sleep(1.5) + ok = any( + "online-deferred" in xml + for client in (bob, alice_other) + for xml in slot(root, client)["pending"] + ) + assert ok, (slot(root, bob)["pending"], slot(root, alice_other)["pending"]) + + assert all(not client.errors for client in clients), [c.errors for c in clients] + return "pass" + 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") + args = parser.parse_args() + with tempfile.TemporaryDirectory(prefix="deferred-sm-") as directory: + work = pathlib.Path(directory) + binary = pathlib.Path(args.binary).resolve() + root = work / "data-root" + root.mkdir() + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + subprocess.run( + [ + "openssl", + "req", + "-x509", + "-newkey", + "rsa:2048", + "-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, + ) + 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", + "mam", + "pep", + "carbons", + ], + "smacks_online_durable": False, + "smacks_flush_interval": 1, + "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, + ) + 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() + 60 + while True: + assert server.poll() is None, log.read() + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.1): + break + except OSError: + assert time.monotonic() < deadline, "server startup timeout" + time.sleep(0.05) + status = asyncio.run(asyncio.wait_for(exercise(root, port, args), 50)) + print(f"deferred online SM: {status}") + except BaseException: + log.flush() + log.seek(0) + print(log.read(), file=sys.stderr) + raise + finally: + server.kill() + server.wait() + + +if __name__ == "__main__": + main() diff --git a/tests/sm_journal.rs b/tests/sm_journal.rs index 890fc25..b6403af 100644 --- a/tests/sm_journal.rs +++ b/tests/sm_journal.rs @@ -320,3 +320,77 @@ fn superseded_frames_still_require_valid_checksums_json_and_baseline_flags() { assert!(Hub::persistent(dir.path()).is_err(), "{damage}"); } } + +#[test] +fn deferred_online_commits_are_flushed_and_recovered() { + let dir = tempfile::tempdir().unwrap(); + let hub = Hub::persistent(dir.path()).unwrap(); + hub.set_defer_online(true); + let (tx, _rx) = mpsc::channel(8); + let session = hub.bind("a@x/r", "c", tx, Duration::from_secs(60)).unwrap(); + let token; + let durable_after_enable; + { + let mut state = session.lock().unwrap(); + state.enable(true).unwrap(); + token = state.resume_id.clone().unwrap(); + durable_after_enable = fs::read(dir.path().join(format!("{token}.wal"))).unwrap(); + // A live session's outbound stanza must not force a durable fsync in + // deferred mode: the WAL stays exactly as written by enable(), while the + // session records dirty state for the background flush. + state.transmit("online-deferred".into()).unwrap(); + assert_eq!( + fs::read(dir.path().join(format!("{token}.wal"))).unwrap(), + durable_after_enable, + "live deferred transmit changed durable WAL" + ); + assert!( + state.dirty, + "deferred online commit did not mark session dirty" + ); + } + // The background flush persists the deferred queue durably. + hub.flush_dirty().unwrap(); + drop(session); + 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.pending.iter().map(String::as_str).collect::>(), + ["online-deferred"] + ); + assert!(!state.dirty); +} + +#[test] +fn deferred_mode_keeps_offline_targets_durable() { + let dir = tempfile::tempdir().unwrap(); + let hub = Hub::persistent(dir.path()).unwrap(); + hub.set_defer_online(true); + let (tx, _rx) = mpsc::channel(8); + let session = hub.bind("b@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(); + // Hibernated/offline transports (sender == None) still commit durably + // before the pending queue is consumed, even under deferred mode. + state.sender = None; + state.track_outbound("offline-durable").unwrap(); + state.checkpoint().unwrap(); + let durable: serde_json::Value = + serde_json::from_slice(&fs::read(dir.path().join(format!("{token}.json"))).unwrap()) + .unwrap(); + assert_eq!(durable["pending"][0], "offline-durable"); + } + drop(hub); + let hub = Hub::persistent(dir.path()).unwrap(); + let session = hub.get("b@x/r").unwrap(); + let state = session.lock().unwrap(); + assert_eq!( + state.pending.iter().map(String::as_str).collect::>(), + ["offline-durable"] + ); +}