diff --git a/.gitignore b/.gitignore index faa40cd..0c6be7b 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,3 @@ *.local.* /demo/ /bin/ -**pycache** diff --git a/AGENTS.md b/AGENTS.md index 5b3a408..0fbb21c 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,10 +255,7 @@ 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. Начиная с0.8.4 fsync/append не -держится под глобальным `Store.slots` lock: короткий lock `slots` только -резервирует слот, сам commit идёт под per-session `wal` lock; compaction/GC под -отдельным `maintenance` lock вне per-session lock. +checkpoint и не удаляйте redo без checkpoint. Общий SM mutex пока сохранён. Slot limit 8 MiB, total 256 MiB; WAL rotation 16 MiB плюс один slot, aggregate 512 MiB. Не допускайте роста каждого из тысяч slots без aggregate cap. @@ -301,26 +298,3 @@ 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 deleted file mode 100644 index 4315893..0000000 --- a/CHANGELOG-0.8.4.md +++ /dev/null @@ -1,45 +0,0 @@ -# 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 1f49340..01b96dd 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,18 +172,3 @@ 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 4e10b61..8dca692 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,25 +101,3 @@ 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 4aa8fad..1448d25 100755 --- a/scripts/verify-local.sh +++ b/scripts/verify-local.sh @@ -32,7 +32,6 @@ 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 7427be6..101f91f 100644 --- a/src/server.rs +++ b/src/server.rs @@ -892,12 +892,6 @@ 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. @@ -1248,9 +1242,6 @@ 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 { @@ -1286,17 +1277,6 @@ 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 d698482..cc48f91 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, HashMap}, + collections::BTreeMap, fs::{self, File, OpenOptions}, io::{Read, Write}, path::{Path, PathBuf}, @@ -115,7 +115,6 @@ impl Record { closed: false, close_condition: None, cancel: Default::default(), - dirty: false, durable: None, } } @@ -128,20 +127,6 @@ 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 @@ -161,26 +146,7 @@ pub(crate) fn enable(s: &Session) -> Result<()> { } Ok(()) } -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<()> { +pub(crate) fn checkpoint(s: &Session) -> Result<()> { if let Some(store) = store(s)? && s.resume_id.is_some() { @@ -205,13 +171,6 @@ 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)] @@ -242,9 +201,6 @@ 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; @@ -352,14 +308,6 @@ 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!( @@ -369,77 +317,43 @@ 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)?; - // 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" - ); - } + let mut slots = self.slots.lock().unwrap(); + if let Some(expected) = expected { ensure!( - slots.len() < SESSION_LIMIT || slots.contains_key(&record.resume_id), - "stream management session limit" + slots + .get(&record.resume_id) + .is_some_and(|v| v.0 == expected), + "retired stream transport" ); - let total: usize = slots.values().map(|v| v.1).sum(); - let previous = slots.get(&record.resume_id).map_or(0, |v| v.1); + } else { ensure!( - total - previous + bytes.len() <= TOTAL_LIMIT, - "stream management storage limit" + !slots.contains_key(&record.resume_id) + && !self.retired.lock().unwrap().contains(&record.resume_id), + "duplicate stream token" ); - 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), - ); - } } - result + 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(()) } fn remove(&self, id: &str, connection: &str) -> Result<()> { - 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)); + let mut slots = self.slots.lock().unwrap(); + if slots.get(id).is_none_or(|v| v.0 != connection) { + return Ok(()); } - result + self.commit(safe_id(id)?, None)?; + slots.remove(id); + Ok(()) } // A journal failure may have reached stable storage. Fence this store rather // than allowing an in-memory rollback to overwrite an uncertain commit. @@ -449,15 +363,9 @@ 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 } @@ -481,7 +389,14 @@ impl Store { } else { None }; - let _baseline_frame = frame(true, baseline.as_deref())?; + 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()?; + } self.reset_journal(id, baseline.as_deref())?; } // Rotation is amortized over at least several small commits. Sync the @@ -493,6 +408,14 @@ 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)?; @@ -513,45 +436,22 @@ 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 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(); + fn compact_journals(&self) -> Result<()> { let ids: Vec = self.journals.lock().unwrap().keys().cloned().collect(); for id in ids { - 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; - } + 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())?; } + Ok(()) } fn collect_retired(&self) -> Result<()> { let mut retired = self.retired.lock().unwrap(); diff --git a/src/sessions.rs b/src/sessions.rs index 2a04e53..c61cf5a 100644 --- a/src/sessions.rs +++ b/src/sessions.rs @@ -35,25 +35,13 @@ 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. 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<()> { + /// Persist state before acknowledging client input or exposing output to a transport. + pub fn checkpoint(&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 { @@ -187,24 +175,6 @@ 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(); @@ -225,7 +195,7 @@ impl Hub { /// Refresh only connected deadlines; disconnected deadlines never slide. pub fn checkpoint_live(&self) -> Result<()> { for session in self.all() { - let mut state = session.lock().unwrap(); + let state = session.lock().unwrap(); if state.sender.is_some() && !state.closed { state.checkpoint()?; } @@ -344,7 +314,6 @@ 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 deleted file mode 100644 index fe1bbba..0000000 --- a/tests/deferred_sm_online_network.py +++ /dev/null @@ -1,233 +0,0 @@ -#!/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 b6403af..890fc25 100644 --- a/tests/sm_journal.rs +++ b/tests/sm_journal.rs @@ -320,77 +320,3 @@ 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"] - ); -}