diff --git a/.github/workflows/signal-durability-nightly.yml b/.github/workflows/signal-durability-nightly.yml new file mode 100644 index 000000000..7b1b55182 --- /dev/null +++ b/.github/workflows/signal-durability-nightly.yml @@ -0,0 +1,82 @@ +name: Signal Durability Nightly + +on: + schedule: + - cron: '17 3 * * *' + workflow_dispatch: + inputs: + seed: + description: First deterministic seed (decimal or 0x-prefixed) + required: false + default: '0x51A6DA7AB1E50001' + seeds: + description: Number of seeds + required: false + default: '128' + steps: + description: Actions per seed + required: false + default: '256' + +permissions: + contents: read + +concurrency: + group: signal-durability-nightly + cancel-in-progress: false + +env: + CARGO_TERM_COLOR: always + PROTOC_VERSION: '3.25.3' + SCCACHE_GHA_ENABLED: 'true' + RUSTC_WRAPPER: sccache + CARGO_INCREMENTAL: '0' + +jobs: + durability: + name: Signal durability chaos + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 + with: + persist-credentials: false + - name: Reclaim unused Android SDK space + run: | + df -h / + sudo rm -rf --one-file-system /usr/local/lib/android + df -h / + - uses: dtolnay/rust-toolchain@2c7215f132e9ebf062739d9130488b56d53c060c + with: + toolchain: nightly-2026-06-16 + - name: Install protoc + uses: taiki-e/install-action@ed67fa35ac944f3a9b33f12c4dd43b6f31a47e20 + with: + tool: protoc@${{ env.PROTOC_VERSION }} + - name: Setup sccache + uses: mozilla-actions/sccache-action@9e7fa8a12102821edf02ca5dbea1acd0f89a2696 + - name: Cache Rust build (registry + target) + uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 + with: + cache-targets: 'true' + - name: Run deterministic DM and group matrix + env: + SIGNAL_CHAOS_SEED: ${{ inputs.seed || '0x51A6DA7AB1E50001' }} + SIGNAL_CHAOS_SEEDS: ${{ inputs.seeds || '128' }} + SIGNAL_CHAOS_STEPS: ${{ inputs.steps || '256' }} + run: >- + cargo test -p wacore --lib signal_durability_chaos_nightly -- + --ignored --nocapture + - name: Run abrupt SQLite restart test + run: >- + cargo test -p whatsapp-rust --test signal_durability_sqlite + signal_durability_sqlite_process_restart -- + --ignored --exact --nocapture + - name: Preserve failed SQLite fixture + if: failure() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 + with: + name: signal-durability-sqlite-${{ github.run_id }} + path: /tmp/whatsapp-rust-signal-durability-*.db* + if-no-files-found: ignore + retention-days: 7 diff --git a/AGENTS.md b/AGENTS.md index 52eac20f1..b17d5c72d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -44,5 +44,6 @@ Read these when working on the relevant area: - `agent_docs/debugging.md` — evcxr REPL, binary protocol debugging - `agent_docs/binary_size_ci.md` — size-tracking CI: metrics, budgets, baseline semantics - `agent_docs/observability.md` — per-session stats (I/O, memory report, TaskInstrument/CPU), design rules +- `agent_docs/signal_durability.md` — Signal counter leases, pre-wire gates, crash recovery, review checklist When adding comments to the code, dont be so verbose, also only explain why, not what diff --git a/agent_docs/signal_durability.md b/agent_docs/signal_durability.md new file mode 100644 index 000000000..9d9136a51 --- /dev/null +++ b/agent_docs/signal_durability.md @@ -0,0 +1,134 @@ +# Signal durability + +This document is the checklist for code that reads, mutates, persists, or sends +Signal state. The security property is simple: an outbound message key and IV +must never reach the wire twice, including after cancellation, storage failure, +reconnect, or process death. + +## State and leases + +DM sessions and group sender keys use the same durability scheme: + +| State | Counter | Cache gate | +| --- | --- | --- | +| `SessionRecord` | `reserved_sender_chain_index` | `reservation_pending` | +| `SenderKeyRecord` | `reserved_iteration` | `wire_gate_pending` | + +The reservation is an exclusive upper bound. A send below it is covered by a +previously persisted lease and can use write-behind. A send at the bound raises +it by `SENDER_CHAIN_RESERVATION_BATCH` and must wait for a successful durable +flush before its ciphertext is published. + +The cache takes ownership of transient record gates. A failed write, a checked +out record skipped by a flush, or a tombstone whose delete failed must remain +gated. Only the backend operation that persisted that address may release it. +Decrypt-side advances are dirty but not pre-wire gated: they can be derived +forward again after a crash. + +Stored records carry the cache incarnation that wrote them: + +- A reload in the same live cache is exact. Eviction and `clear_after_flush()` + must not burn the unused part of a lease. +- A new process or lossy cache reset has a new incarnation. It fast-forwards to + the stored reservation because any counter below it may already have been + published. +- Stores that bypass `SignalStoreCache` cannot claim an exact reload. They must + use the incarnation-aware record format or conservatively recover as a new + incarnation. + +The current batch is 64 while the peer forward-jump limit is 2000. That makes a +single crash gap at most 3.2% of the receiver limit and amortizes a monotonic +sender chain to one synchronous reservation write per 64 messages. Tune this +only with sender-chain run-length and restart data; transport stanza counts do +not expose Signal iterations. Keep the batch well below `MAX_FORWARD_JUMPS` and +run the recovery matrix after any change. + +Real crash/send cycles can accumulate burned ranges for a receiver that misses +every intervening message. Crossing its forward-jump bound is recoverable via +the retry/SKDM path, but clean cache reloads must never contribute to that gap. + +## Publication boundary + +All ciphertext APIs follow this order: + +1. Load the record for mutation under its per-address or per-sender-key lock. +2. Derive the message key, advance the chain, and return the record to the + cache even if the future can be cancelled. +3. If the advance raised a lease, let the cache adopt its transient gate. +4. Call `persist_signal_state_pre_wire()` after every recipient has been + encrypted and before handing the stanza to the transport. +5. Abort the send if that flush fails. + +The predicate is intentionally global. A pending lease for one address can +force another send to flush, but no call site can accidentally omit an address +whose ciphertext is already part of the stanza. + +Do not replace the batch-safe flush with a raw cache flush. During offline +drain, inbound rows must become durable before their ratchet advances; otherwise +a crash can turn redelivery into an acknowledged duplicate and lose the event. + +## Cancellation, deletion, and teardown + +`SessionCheckout` owns the only mutable copy while a session operation is in +flight. Its drop path returns the advanced record synchronously or queues the +restore if the cache lock is contended. A checkout token and recovery generation +prevent a stale owner from overwriting a delete, newer owner, or lossy reset. + +Deletion is durable state. A session or sender-key tombstone must retain any +existing pre-wire gate until the backend delete succeeds. A consumed one-time +prekey is deleted only after its promoted session is durable, so a crash cannot +lose both recovery inputs. + +Clean reconnect teardown flushes and then calls `clear_after_flush()`. Dirty +state from a failed final flush stays resident for the next attempt. `clear()` +is lossy: it changes the incarnation and is only valid when the corresponding +uncommitted inbound work is also dropped so the server can redeliver it. + +## Review checklist + +For any new or changed ciphertext path, verify: + +- The chain mutation is returned to the cache across every error and + cancellation edge. +- A newly raised lease reaches durable storage before any ciphertext derived + from it reaches the transport. +- A failed flush aborts publication and leaves both dirty state and gate intact. +- Covered sends avoid synchronous storage without skipping eventual + write-behind. +- Clean eviction/reload preserves the exact counter; crash reload burns to the + exclusive reservation ceiling. +- Deletes cannot be undone by a stale checkout or in-flight sender-key writer. +- Lock ordering matches existing session, sender-key, inbound-drain, and cache + ordering; no backend await is added under an unrelated device lock. +- Tests use fictitious JIDs and never log key material, plaintext, or production + identifiers. + +## Verification + +Focused unit tests live beside `SignalStoreCache`, record serialization, and +the libsignal ciphers. The deterministic state machine combines DM and group +sends, failed writes, cancellation, checkout/flush overlap, tombstones, clean +reloads, lossy clears, crash recovery, out-of-order group delivery, receiver +state loss, and retry redistribution. + +```bash +# Small matrix in normal CI +cargo test -p wacore signal_durability_chaos_smoke + +# Nightly-sized local run +SIGNAL_CHAOS_SEEDS=128 SIGNAL_CHAOS_STEPS=256 \ + cargo test -p wacore --lib signal_durability_chaos_nightly -- --ignored --nocapture + +# Replay the seed printed by a failure +SIGNAL_CHAOS_SEED=0x... SIGNAL_CHAOS_SEEDS=1 SIGNAL_CHAOS_STEPS=256 \ + cargo test -p wacore --lib signal_durability_chaos_nightly -- --ignored --nocapture + +# Real SQLite database across a SIGKILL and restart on Unix +cargo test -p whatsapp-rust --test signal_durability_sqlite \ + signal_durability_sqlite_process_restart -- --ignored --exact --nocapture +``` + +The scheduled workflow runs the large state-machine matrix and the SQLite +subprocess test. A state-machine failure reports its replay seed, step, and +action; a failed SQLite job retains its synthetic database as a short-lived CI +artifact. diff --git a/tests/signal_durability_sqlite.rs b/tests/signal_durability_sqlite.rs new file mode 100644 index 000000000..a7f494e68 --- /dev/null +++ b/tests/signal_durability_sqlite.rs @@ -0,0 +1,255 @@ +#![cfg(feature = "sqlite-storage")] + +use std::path::Path; +use std::process::Command; + +use rand::SeedableRng; +use wacore::libsignal::protocol::consts::SENDER_CHAIN_RESERVATION_BATCH; +use wacore::libsignal::protocol::{ + ChainKey, IdentityKey, KeyPair, RootKey, SenderKeyRecord, SenderKeyStore, SessionRecord, + SessionState, create_sender_key_distribution_message, group_encrypt, +}; +use wacore::libsignal::store::sender_key_name::SenderKeyName; +use whatsapp_rust::store::SqliteStore; +use whatsapp_rust::store::signal_cache::SignalStoreCache; + +const CHILD_MARKER: &str = "SIGNAL_DURABILITY_CRASH_CHILD"; +const DATABASE_ENV: &str = "SIGNAL_DURABILITY_DATABASE"; +#[cfg(not(unix))] +const CHILD_EXIT_CODE: i32 = 91; +const FIXTURE_SEED: u64 = 0x51A6_5A17_EC4A_5E01; + +type DmFingerprint = ([u8; 32], [u8; 16]); + +struct CachedSenderKeyStore<'a> { + cache: &'a SignalStoreCache, + backend: &'a SqliteStore, +} + +#[async_trait::async_trait] +impl SenderKeyStore for CachedSenderKeyStore<'_> { + async fn store_sender_key( + &mut self, + name: &SenderKeyName, + record: SenderKeyRecord, + ) -> wacore::libsignal::protocol::error::Result<()> { + self.cache.put_sender_key(name, record).await; + Ok(()) + } + + async fn load_sender_key( + &self, + name: &SenderKeyName, + ) -> wacore::libsignal::protocol::error::Result> { + Ok(self + .cache + .get_sender_key(name, self.backend) + .await + .map_err(|error| { + wacore::libsignal::protocol::SignalProtocolError::BackendError( + "SQLite durability test", + error.into(), + ) + })? + .map(|record| (*record).clone())) + } +} + +fn dm_address() -> wacore::libsignal::protocol::ProtocolAddress { + wacore::libsignal::protocol::ProtocolAddress::new("15550008001".to_string(), 1.into()) +} + +fn group_name() -> SenderKeyName { + SenderKeyName::from_parts("120363000000080001@g.us", "15550008002@s.whatsapp.net:0") +} + +fn fresh_session(rng: &mut rand::rngs::StdRng) -> SessionRecord { + let local = IdentityKey::new(KeyPair::generate(rng).public_key); + let remote = IdentityKey::new(KeyPair::generate(rng).public_key); + let base_key = KeyPair::generate(rng).public_key; + let mut state = SessionState::new(3, &local, &remote, &RootKey::new([7; 32]), &base_key); + state.set_sender_chain(&KeyPair::generate(rng), &ChainKey::new([11; 32], 0)); + SessionRecord::new(state) +} + +fn spend_dm(record: &mut SessionRecord) -> (u32, DmFingerprint) { + let chain = record + .session_state() + .expect("session state") + .get_sender_chain_key() + .expect("sender chain"); + let keys = chain.message_keys().generate_keys(); + let fingerprint = (*keys.cipher_key(), *keys.iv()); + let next = chain.next_chain_key().expect("next chain key"); + record + .session_state_mut() + .expect("session state") + .set_sender_chain_key(&next) + .expect("sender chain update"); + if chain.index() >= record.reserved_sender_chain_index() { + record.reserve_sender_chain_counters(chain.index()); + } + (chain.index(), fingerprint) +} + +async fn crash_child(database: &str) -> ! { + let store = SqliteStore::new(database).await.expect("SQLite store"); + let cache = SignalStoreCache::new(); + let address = dm_address(); + let name = group_name(); + let mut rng = rand::rngs::StdRng::seed_from_u64(FIXTURE_SEED); + + let mut dm = fresh_session(&mut rng); + assert_eq!(spend_dm(&mut dm).0, 0); + cache.put_session(&address, dm).await; + + let mut sender = CachedSenderKeyStore { + cache: &cache, + backend: &store, + }; + create_sender_key_distribution_message(&name, &mut sender, &mut rng) + .await + .expect("sender-key setup"); + let first = group_encrypt(&mut sender, &name, b"first", &mut rng) + .await + .expect("first group encrypt"); + assert_eq!(first.iteration(), 0); + + cache.flush(&store).await.expect("durable lease flush"); + assert!(!cache.needs_pre_wire_flush().await); + + let mut dm = cache + .get_session(&address, &store) + .await + .expect("session load") + .expect("session"); + for expected in 1..=5 { + assert_eq!(spend_dm(&mut dm).0, expected); + } + cache.put_session(&address, dm).await; + for expected in 1..=5 { + let message = group_encrypt(&mut sender, &name, b"unflushed", &mut rng) + .await + .expect("group encrypt"); + assert_eq!(message.iteration(), expected); + } + + crash_now() +} + +#[cfg(unix)] +fn crash_now() -> ! { + // SIGKILL prevents process-exit hooks from making the fixture cleaner than a real crash. + let result = unsafe { libc::raise(libc::SIGKILL) }; + panic!("SIGKILL failed with {result}") +} + +#[cfg(not(unix))] +fn crash_now() -> ! { + std::process::exit(CHILD_EXIT_CODE) +} + +async fn verify_recovery(database: &str) { + let store = SqliteStore::new(database) + .await + .expect("reopen SQLite store"); + let cache = SignalStoreCache::new(); + let address = dm_address(); + let name = group_name(); + let mut rng = rand::rngs::StdRng::seed_from_u64(FIXTURE_SEED ^ 0xFFFF); + + let mut child_record = fresh_session(&mut rand::rngs::StdRng::seed_from_u64(FIXTURE_SEED)); + let mut child_fingerprints = Vec::with_capacity(6); + for _ in 0..6 { + child_fingerprints.push(spend_dm(&mut child_record).1); + } + + let mut recovered = cache + .get_session(&address, &store) + .await + .expect("recovery load") + .expect("durable session"); + let (counter, fingerprint) = spend_dm(&mut recovered); + assert_eq!(counter, SENDER_CHAIN_RESERVATION_BATCH); + assert!(!child_fingerprints.contains(&fingerprint)); + cache.put_session(&address, recovered).await; + + let mut sender = CachedSenderKeyStore { + cache: &cache, + backend: &store, + }; + let recovered_group = group_encrypt(&mut sender, &name, b"recovered", &mut rng) + .await + .expect("group recovery encrypt"); + assert_eq!(recovered_group.iteration(), SENDER_CHAIN_RESERVATION_BATCH); + assert!(cache.needs_pre_wire_flush().await); + cache.flush(&store).await.expect("recovery lease flush"); + assert!(!cache.needs_pre_wire_flush().await); + + cache.clear_after_flush().await; + let mut exact = cache + .get_session(&address, &store) + .await + .expect("clean reload") + .expect("durable session"); + assert_eq!(spend_dm(&mut exact).0, SENDER_CHAIN_RESERVATION_BATCH + 1); + cache.put_session(&address, exact).await; + let exact_group = group_encrypt(&mut sender, &name, b"exact", &mut rng) + .await + .expect("clean group reload"); + assert_eq!(exact_group.iteration(), SENDER_CHAIN_RESERVATION_BATCH + 1); + assert!(!cache.needs_pre_wire_flush().await); +} + +fn remove_database(path: &Path) { + let _ = std::fs::remove_file(path); + let base = path.as_os_str().to_string_lossy(); + let _ = std::fs::remove_file(format!("{base}-wal")); + let _ = std::fs::remove_file(format!("{base}-shm")); +} + +#[tokio::test] +#[ignore = "run as a subprocess by the SQLite durability test"] +async fn signal_durability_sqlite_crash_child() { + if std::env::var_os(CHILD_MARKER).is_none() { + return; + } + let database = std::env::var(DATABASE_ENV).expect("child database path"); + crash_child(&database).await; +} + +#[tokio::test] +#[ignore = "run by signal-durability-nightly.yml"] +async fn signal_durability_sqlite_process_restart() { + let database = std::env::temp_dir().join(format!( + "whatsapp-rust-signal-durability-{}.db", + uuid::Uuid::new_v4() + )); + let executable = std::env::current_exe().expect("current test executable"); + let child_database = database.clone(); + let status = tokio::task::spawn_blocking(move || { + Command::new(executable) + .arg("signal_durability_sqlite_crash_child") + .arg("--exact") + .arg("--ignored") + .arg("--nocapture") + .env(CHILD_MARKER, "1") + .env(DATABASE_ENV, child_database) + .status() + }) + .await + .expect("child task") + .expect("start child process"); + #[cfg(unix)] + { + use std::os::unix::process::ExitStatusExt; + assert_eq!(status.signal(), Some(libc::SIGKILL)); + } + #[cfg(not(unix))] + assert_eq!(status.code(), Some(CHILD_EXIT_CODE)); + + verify_recovery(database.to_str().expect("UTF-8 database path")).await; + tokio::task::spawn_blocking(move || remove_database(&database)) + .await + .expect("database cleanup task"); +} diff --git a/wacore/Cargo.toml b/wacore/Cargo.toml index 17126c5ba..6bba1afec 100644 --- a/wacore/Cargo.toml +++ b/wacore/Cargo.toml @@ -96,7 +96,7 @@ sha2 = { workspace = true } [dev-dependencies] divan = { workspace = true } futures = { workspace = true, features = ["executor", "thread-pool"] } -tokio = { workspace = true, features = ["macros", "rt"] } +tokio = { workspace = true, features = ["macros", "rt", "time"] } [[bench]] name = "reporting_token_benchmark" diff --git a/wacore/src/store/signal_cache.rs b/wacore/src/store/signal_cache.rs index 9c9e7c79e..979ef7211 100644 --- a/wacore/src/store/signal_cache.rs +++ b/wacore/src/store/signal_cache.rs @@ -3577,3 +3577,7 @@ mod pre_wire_gate_tests { assert!(!cache.needs_pre_wire_flush().await); } } + +#[cfg(test)] +#[path = "signal_cache_durability_chaos.rs"] +mod durability_chaos_tests; diff --git a/wacore/src/store/signal_cache_durability_chaos.rs b/wacore/src/store/signal_cache_durability_chaos.rs new file mode 100644 index 000000000..5ef7dfbea --- /dev/null +++ b/wacore/src/store/signal_cache_durability_chaos.rs @@ -0,0 +1,815 @@ +use super::*; + +use std::collections::{HashSet, VecDeque}; +use std::time::Duration; + +use anyhow::{Context, ensure}; +use rand::{RngExt, SeedableRng}; + +use crate::libsignal::protocol::{ + ChainKey, CiphertextMessageType, IdentityKey, KeyPair, RootKey, SenderKeyMessage, + SenderKeyStore, SessionState, SignalProtocolError, create_sender_key_distribution_message, + group_decrypt, group_encrypt, process_sender_key_distribution_message, +}; +use crate::store::in_memory::InMemoryBackend; + +const SMOKE_SEEDS: usize = 4; +const SMOKE_STEPS: usize = 64; +const NIGHTLY_SEEDS: usize = 128; +const NIGHTLY_STEPS: usize = 256; +const DEFAULT_SEED: u64 = 0x51A6_DA7A_B1E5_0001; + +type KeyIvFingerprint = ([u8; 32], [u8; 16]); + +struct CachedSenderKeyStore<'a> { + cache: &'a SignalStoreCache, + backend: &'a InMemoryBackend, +} + +#[async_trait::async_trait] +impl SenderKeyStore for CachedSenderKeyStore<'_> { + async fn store_sender_key( + &mut self, + name: &SenderKeyName, + record: SenderKeyRecord, + ) -> crate::libsignal::protocol::error::Result<()> { + self.cache.put_sender_key(name, record).await; + Ok(()) + } + + async fn load_sender_key( + &self, + name: &SenderKeyName, + ) -> crate::libsignal::protocol::error::Result> { + Ok(self + .cache + .get_sender_key(name, self.backend) + .await + .map_err(|error| { + crate::libsignal::protocol::SignalProtocolError::BackendError( + "durability chaos store", + error.into(), + ) + })? + .map(|record| (*record).clone())) + } +} + +#[derive(Clone, Copy, Debug)] +enum Action { + DmSend { fail_gate: bool }, + GroupSend { fail_gate: bool }, + DmCancel, + DeliverGroup { newest: bool }, + Flush, + FailPendingFlush, + CleanReload, + CrashReload, + LossyClear, + DeleteDm, + DeleteGroup, + CheckoutDuringFlush, + RecoverGroup, +} + +#[derive(Clone, Copy)] +enum FlushFailure { + None, + Session, + SenderKey, +} + +struct SplitMix64(u64); + +impl SplitMix64 { + fn next(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut value = self.0; + value = (value ^ (value >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + value = (value ^ (value >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + value ^ (value >> 31) + } + + fn action(&mut self) -> Action { + match self.next() % 25 { + 0 => Action::DmSend { fail_gate: true }, + 1..=6 => Action::DmSend { fail_gate: false }, + 7 => Action::GroupSend { fail_gate: true }, + 8..=13 => Action::GroupSend { fail_gate: false }, + 14 => Action::DmCancel, + 15 => Action::DeliverGroup { + newest: self.next() & 1 == 0, + }, + 16 => Action::Flush, + 17 => Action::FailPendingFlush, + 18 => Action::CleanReload, + 19 => Action::CrashReload, + 20 => Action::LossyClear, + 21 => Action::DeleteDm, + 22 => Action::DeleteGroup, + 23 => Action::CheckoutDuringFlush, + 24 => Action::RecoverGroup, + _ => unreachable!(), + } + } +} + +struct ChaosHarness { + backend: InMemoryBackend, + cache: SignalStoreCache, + receiver_backend: InMemoryBackend, + receiver_cache: SignalStoreCache, + dm_address: ProtocolAddress, + group_name: SenderKeyName, + crypto_rng: rand::rngs::StdRng, + incarnation_generation: u64, + published_dm: HashSet, + published_group: HashSet, + pending_group: VecDeque, +} + +impl ChaosHarness { + async fn new(seed: u64) -> anyhow::Result { + let mut harness = Self { + backend: InMemoryBackend::new(), + cache: SignalStoreCache::with_max_entries_and_incarnation(32, incarnation(seed, 0)), + receiver_backend: InMemoryBackend::new(), + receiver_cache: SignalStoreCache::with_max_entries_and_incarnation( + 32, + incarnation(seed ^ 0xA5A5_A5A5_A5A5_A5A5, 0), + ), + dm_address: ProtocolAddress::new("15550007001".to_string(), 1.into()), + group_name: SenderKeyName::from_parts( + "120363000000070001@g.us", + "15550007002@s.whatsapp.net:0", + ), + crypto_rng: rand::rngs::StdRng::seed_from_u64(seed ^ 0xC4A5_5EED_5AFE_0001), + incarnation_generation: 0, + published_dm: HashSet::new(), + published_group: HashSet::new(), + pending_group: VecDeque::new(), + }; + + ensure!(harness.dm_send(false).await?, "initial DM send was gated"); + harness.sync_group_distribution().await?; + ensure!( + harness.group_send(false).await?, + "initial group send was gated" + ); + harness.deliver_group(false).await?; + harness.assert_invariants().await?; + Ok(harness) + } + + async fn apply(&mut self, action: Action) -> anyhow::Result<()> { + match action { + Action::DmSend { fail_gate } => { + self.dm_send(fail_gate).await?; + } + Action::GroupSend { fail_gate } => { + self.group_send(fail_gate).await?; + } + Action::DmCancel => self.dm_cancel().await?, + Action::DeliverGroup { newest } => { + self.deliver_group(newest).await?; + } + Action::Flush => self.flush_successfully().await?, + Action::FailPendingFlush => self.fail_pending_flush().await?, + Action::CleanReload => self.clean_reload().await?, + Action::CrashReload => self.crash_reload(), + Action::LossyClear => self.lossy_clear().await?, + Action::DeleteDm => self.cache.delete_session(&self.dm_address).await, + Action::DeleteGroup => { + self.cache + .delete_sender_key(self.group_name.cache_key()) + .await; + } + Action::CheckoutDuringFlush => self.checkout_during_flush().await?, + Action::RecoverGroup => self.recover_group().await?, + } + self.assert_invariants().await + } + + async fn dm_send(&mut self, fail_gate: bool) -> anyhow::Result { + let (record, checkout) = self + .cache + .checkout_session(&self.dm_address, &self.backend) + .await?; + let had_session = record.is_some(); + let mut record = record.unwrap_or_else(|| fresh_session(&mut self.crypto_rng)); + let chain = record + .session_state() + .context("DM session state missing")? + .get_sender_chain_key() + .map_err(|_| anyhow::anyhow!("DM sender chain missing"))?; + let keys = chain.message_keys().generate_keys(); + let fingerprint = (*keys.cipher_key(), *keys.iv()); + let next = chain.next_chain_key()?; + record + .session_state_mut() + .context("DM session state missing")? + .set_sender_chain_key(&next) + .map_err(|_| anyhow::anyhow!("DM sender chain update failed"))?; + if chain.index() >= record.reserved_sender_chain_index() { + record.reserve_sender_chain_counters(chain.index()); + } + self.commit_dm(record, checkout, had_session).await?; + + let published = self + .release_wire_gate(if fail_gate { + FlushFailure::Session + } else { + FlushFailure::None + }) + .await?; + if published { + ensure!( + self.published_dm.insert(fingerprint), + "DM key/IV was published twice at counter {}", + chain.index() + ); + } + Ok(published) + } + + async fn dm_cancel(&mut self) -> anyhow::Result<()> { + let (record, checkout) = self + .cache + .checkout_session(&self.dm_address, &self.backend) + .await?; + let Some(mut record) = record else { + self.cache + .cancel_session_checkout(&self.dm_address, checkout); + return Ok(()); + }; + let chain = record + .session_state() + .context("DM session state missing")? + .get_sender_chain_key() + .map_err(|_| anyhow::anyhow!("DM sender chain missing"))?; + let next = chain.next_chain_key()?; + record + .session_state_mut() + .context("DM session state missing")? + .set_sender_chain_key(&next) + .map_err(|_| anyhow::anyhow!("DM sender chain update failed"))?; + if chain.index() >= record.reserved_sender_chain_index() { + record.reserve_sender_chain_counters(chain.index()); + } + self.commit_dm(record, checkout, true).await + } + + async fn commit_dm( + &self, + record: SessionRecord, + checkout: SessionCheckoutKey, + had_session: bool, + ) -> anyhow::Result<()> { + match self.cache.restore_session_from_checkout( + &self.dm_address, + record, + checkout, + had_session, + ) { + SessionCheckoutStoreResult::Stored => Ok(()), + SessionCheckoutStoreResult::Pending(completion) => { + self.cache.complete_session_checkout().await; + ensure!( + completion.load(Ordering::Acquire), + "queued DM checkout restore was rejected" + ); + Ok(()) + } + SessionCheckoutStoreResult::Rejected => { + anyhow::bail!("DM checkout restore was rejected") + } + SessionCheckoutStoreResult::Unhandled(_) => { + anyhow::bail!("cache did not handle a DM checkout restore") + } + } + } + + async fn group_send(&mut self, fail_gate: bool) -> anyhow::Result { + let record = if let Some(record) = self + .cache + .get_sender_key(&self.group_name, &self.backend) + .await? + { + record + } else { + self.sync_group_distribution().await?; + self.cache + .get_sender_key(&self.group_name, &self.backend) + .await? + .context("sender-key setup did not populate the cache")? + }; + let fingerprint = group_key_fingerprint(&record)?; + drop(record); + let message = { + let mut sender = CachedSenderKeyStore { + cache: &self.cache, + backend: &self.backend, + }; + group_encrypt( + &mut sender, + &self.group_name, + b"durability-chaos", + &mut self.crypto_rng, + ) + .await? + }; + let published = self + .release_wire_gate(if fail_gate { + FlushFailure::SenderKey + } else { + FlushFailure::None + }) + .await?; + if published { + ensure!( + self.published_group.insert(fingerprint), + "group key/IV was published twice at chain {} iteration {}", + message.chain_id(), + message.iteration() + ); + self.pending_group.push_back(message); + if self.pending_group.len() > 64 { + self.pending_group.pop_front(); + } + } + Ok(published) + } + + async fn sync_group_distribution(&mut self) -> anyhow::Result<()> { + let distribution = { + let mut sender = CachedSenderKeyStore { + cache: &self.cache, + backend: &self.backend, + }; + create_sender_key_distribution_message( + &self.group_name, + &mut sender, + &mut self.crypto_rng, + ) + .await? + }; + let mut receiver = CachedSenderKeyStore { + cache: &self.receiver_cache, + backend: &self.receiver_backend, + }; + process_sender_key_distribution_message(&self.group_name, &distribution, &mut receiver) + .await?; + self.pending_group.clear(); + Ok(()) + } + + async fn deliver_group(&mut self, newest: bool) -> anyhow::Result { + let Some(message) = (if newest { + self.pending_group.pop_back() + } else { + self.pending_group.pop_front() + }) else { + return Ok(false); + }; + let mut receiver = CachedSenderKeyStore { + cache: &self.receiver_cache, + backend: &self.receiver_backend, + }; + match group_decrypt(message.serialized(), &mut receiver, &self.group_name).await { + Ok(_) => return Ok(false), + Err(SignalProtocolError::InvalidMessage( + CiphertextMessageType::SenderKey, + "message from too far into the future", + )) + | Err(SignalProtocolError::NoSenderKeyState(_)) => {} + Err(error) => return Err(error.into()), + } + + self.sync_group_distribution().await?; + ensure!( + self.group_send(false).await?, + "retry group message remained behind a durability gate" + ); + let retry = self + .pending_group + .pop_back() + .context("group retry message missing")?; + let mut receiver = CachedSenderKeyStore { + cache: &self.receiver_cache, + backend: &self.receiver_backend, + }; + let plaintext = group_decrypt(retry.serialized(), &mut receiver, &self.group_name).await?; + ensure!( + plaintext == b"durability-chaos", + "group retry decrypted the wrong plaintext" + ); + Ok(true) + } + + async fn recover_group(&mut self) -> anyhow::Result<()> { + if self.pending_group.is_empty() { + ensure!( + self.group_send(false).await?, + "group recovery setup remained behind a durability gate" + ); + } + self.receiver_cache + .delete_sender_key(self.group_name.cache_key()) + .await; + ensure!( + self.deliver_group(false).await?, + "lost receiver state did not exercise group recovery" + ); + Ok(()) + } + + async fn release_wire_gate(&self, failure: FlushFailure) -> anyhow::Result { + if !self.cache.needs_pre_wire_flush().await { + return Ok(true); + } + let failure = self.failure_for_pending_gate(failure).await; + self.backend + .set_fail_session_writes(matches!(failure, FlushFailure::Session)); + self.backend + .set_fail_sender_key_writes(matches!(failure, FlushFailure::SenderKey)); + let result = self.cache.flush(&self.backend).await; + self.backend.set_fail_session_writes(false); + self.backend.set_fail_sender_key_writes(false); + + if matches!(failure, FlushFailure::None) { + result?; + ensure!( + !self.cache.needs_pre_wire_flush().await, + "successful pre-wire flush left a gate pending" + ); + return Ok(true); + } + ensure!(result.is_err(), "injected pre-wire flush did not fail"); + ensure!( + self.cache.needs_pre_wire_flush().await, + "failed pre-wire flush released its gate" + ); + Ok(false) + } + + async fn failure_for_pending_gate(&self, preferred: FlushFailure) -> FlushFailure { + if matches!(preferred, FlushFailure::None) { + return preferred; + } + let session_pending = !self + .cache + .lock_sessions() + .await + .reservation_pending + .is_empty(); + let sender_pending = !self + .cache + .sender_keys + .lock() + .await + .wire_gate_pending + .is_empty(); + match preferred { + FlushFailure::Session if session_pending => preferred, + FlushFailure::SenderKey if sender_pending => preferred, + _ if session_pending => FlushFailure::Session, + _ if sender_pending => FlushFailure::SenderKey, + _ => preferred, + } + } + + async fn fail_pending_flush(&self) -> anyhow::Result<()> { + let session_pending = !self + .cache + .lock_sessions() + .await + .reservation_pending + .is_empty(); + let sender_pending = !self + .cache + .sender_keys + .lock() + .await + .wire_gate_pending + .is_empty(); + let failure = if session_pending { + FlushFailure::Session + } else if sender_pending { + FlushFailure::SenderKey + } else { + return Ok(()); + }; + ensure!( + !self.release_wire_gate(failure).await?, + "injected flush unexpectedly released the wire" + ); + Ok(()) + } + + async fn flush_successfully(&self) -> anyhow::Result<()> { + self.cache.flush(&self.backend).await?; + ensure!( + !self.cache.needs_pre_wire_flush().await, + "successful background flush left a gate pending" + ); + Ok(()) + } + + async fn clean_reload(&self) -> anyhow::Result<()> { + let dm_before = self + .cache + .peek_session(&self.dm_address, &self.backend) + .await? + .map(|record| dm_chain_index(&record)) + .transpose()?; + let group_before = self + .cache + .get_sender_key(&self.group_name, &self.backend) + .await? + .map(|record| group_position(&record)) + .transpose()?; + + self.flush_successfully().await?; + self.cache.clear_after_flush().await; + + let dm_after = self + .cache + .peek_session(&self.dm_address, &self.backend) + .await? + .map(|record| dm_chain_index(&record)) + .transpose()?; + let group_after = self + .cache + .get_sender_key(&self.group_name, &self.backend) + .await? + .map(|record| group_position(&record)) + .transpose()?; + ensure!(dm_after == dm_before, "clean DM reload burned a lease"); + ensure!( + group_after == group_before, + "clean group reload burned a lease" + ); + Ok(()) + } + + fn crash_reload(&mut self) { + self.incarnation_generation = self.incarnation_generation.wrapping_add(1); + self.cache = SignalStoreCache::with_max_entries_and_incarnation( + 32, + incarnation(DEFAULT_SEED, self.incarnation_generation), + ); + } + + async fn lossy_clear(&mut self) -> anyhow::Result<()> { + let (record, checkout) = self + .cache + .checkout_session(&self.dm_address, &self.backend) + .await?; + self.incarnation_generation = self.incarnation_generation.wrapping_add(1); + self.cache + .clear_with_incarnation(incarnation( + DEFAULT_SEED ^ 0xFFFF_0000_FFFF_0000, + self.incarnation_generation, + )) + .await; + if let Some(record) = record { + ensure!( + matches!( + self.cache.restore_session_from_checkout( + &self.dm_address, + record, + checkout, + true, + ), + SessionCheckoutStoreResult::Rejected + ), + "lossy clear accepted a stale checkout" + ); + } else { + self.cache + .cancel_session_checkout(&self.dm_address, checkout); + } + ensure!( + !self.cache.needs_pre_wire_flush().await, + "lossy clear retained a stale wire gate" + ); + Ok(()) + } + + async fn checkout_during_flush(&self) -> anyhow::Result<()> { + let (record, checkout) = self + .cache + .checkout_session(&self.dm_address, &self.backend) + .await?; + let dm_was_gated = self + .cache + .lock_sessions() + .await + .reservation_pending + .contains(self.dm_address.as_str()); + self.cache.flush(&self.backend).await?; + match record { + Some(record) => self.commit_dm(record, checkout, true).await?, + None => self + .cache + .cancel_session_checkout(&self.dm_address, checkout), + } + if dm_was_gated { + ensure!( + self.cache.needs_pre_wire_flush().await, + "flush released a checked-out session gate" + ); + } + Ok(()) + } + + async fn assert_invariants(&self) -> anyhow::Result<()> { + let sessions = self.cache.lock_sessions().await; + ensure!( + sessions + .reservation_pending + .iter() + .all(|key| sessions.dirty.contains(key) || sessions.deleted.contains(key)), + "DM gate escaped dirty/tombstone tracking" + ); + ensure!( + sessions + .dirty + .iter() + .all(|key| sessions.cache.contains_key(key.as_ref())), + "dirty DM session was evicted" + ); + ensure!( + sessions + .cache + .values() + .all(|entry| !matches!(entry, SessionEntry::CheckedOut { .. })), + "DM checkout remained stranded" + ); + let session_gate = !sessions.reservation_pending.is_empty(); + drop(sessions); + + ensure!( + self.cache.pending_session_restores().is_empty(), + "queued DM restore remained undrained" + ); + let sender_keys = self.cache.sender_keys.lock().await; + ensure!( + sender_keys + .wire_gate_pending + .iter() + .all(|key| sender_keys.dirty.contains(key)), + "group gate escaped dirty tracking" + ); + ensure!( + sender_keys + .dirty + .iter() + .all(|key| sender_keys.cache.contains_key(key.as_ref())), + "dirty sender key was evicted" + ); + let sender_gate = !sender_keys.wire_gate_pending.is_empty(); + drop(sender_keys); + ensure!( + self.cache.needs_pre_wire_flush().await == (session_gate || sender_gate), + "wire-gate query disagrees with cache state" + ); + Ok(()) + } +} + +fn fresh_session(rng: &mut rand::rngs::StdRng) -> SessionRecord { + let local = IdentityKey::new(KeyPair::generate(rng).public_key); + let remote = IdentityKey::new(KeyPair::generate(rng).public_key); + let base_key = KeyPair::generate(rng).public_key; + let mut root = [0; 32]; + let mut chain = [0; 32]; + rng.fill(&mut root); + rng.fill(&mut chain); + let mut state = SessionState::new(3, &local, &remote, &RootKey::new(root), &base_key); + state.set_sender_chain(&KeyPair::generate(rng), &ChainKey::new(chain, 0)); + SessionRecord::new(state) +} + +fn dm_chain_index(record: &SessionRecord) -> anyhow::Result { + Ok(record + .session_state() + .context("DM session state missing")? + .get_sender_chain_key() + .map_err(|_| anyhow::anyhow!("DM sender chain missing"))? + .index()) +} + +fn group_position(record: &SenderKeyRecord) -> anyhow::Result<(u32, u32)> { + let state = record + .sender_key_state() + .map_err(|_| anyhow::anyhow!("group sender-key state missing"))?; + Ok(( + state.chain_id(), + state + .sender_chain_key() + .context("group sender chain missing")? + .iteration(), + )) +} + +fn group_key_fingerprint(record: &SenderKeyRecord) -> anyhow::Result { + let message_key = record + .sender_key_state() + .map_err(|_| anyhow::anyhow!("group sender-key state missing"))? + .sender_chain_key() + .context("group sender chain missing")? + .sender_message_key(); + let cipher_key: [u8; 32] = message_key + .cipher_key() + .try_into() + .context("group cipher key length")?; + let iv: [u8; 16] = message_key.iv().try_into().context("group IV length")?; + Ok((cipher_key, iv)) +} + +fn incarnation(seed: u64, generation: u64) -> StoreIncarnation { + let mut rng = SplitMix64(seed ^ generation.wrapping_mul(0xD134_2543_DE82_EF95)); + let mut value = [0; 16]; + value[..8].copy_from_slice(&rng.next().to_le_bytes()); + value[8..].copy_from_slice(&rng.next().to_le_bytes()); + value +} + +fn env_u64(name: &str, default: u64) -> u64 { + let value = match std::env::var(name) { + Ok(value) => value, + Err(std::env::VarError::NotPresent) => return default, + Err(std::env::VarError::NotUnicode(value)) => { + panic!("invalid {name}={value:?}: expected Unicode") + } + }; + let parsed = if let Some(hex) = value.strip_prefix("0x") { + u64::from_str_radix(hex, 16) + } else { + value.parse() + }; + parsed.unwrap_or_else(|error| panic!("invalid {name}={value:?}: {error}")) +} + +fn env_usize(name: &str, default: usize) -> usize { + let value = match std::env::var(name) { + Ok(value) => value, + Err(std::env::VarError::NotPresent) => return default, + Err(std::env::VarError::NotUnicode(value)) => { + panic!("invalid {name}={value:?}: expected Unicode") + } + }; + let parsed = value + .parse::() + .unwrap_or_else(|error| panic!("invalid {name}={value:?}: {error}")); + assert!( + (1..=4096).contains(&parsed), + "invalid {name}={value:?}: expected 1..=4096" + ); + parsed +} + +async fn run_seed(seed: u64, steps: usize) -> anyhow::Result<()> { + let mut harness = tokio::time::timeout(Duration::from_secs(5), ChaosHarness::new(seed)) + .await + .context("chaos setup timed out")??; + let mut actions = SplitMix64(seed); + for step in 0..steps { + let action = actions.action(); + let result = tokio::time::timeout(Duration::from_secs(2), harness.apply(action)) + .await + .with_context(|| { + format!("seed=0x{seed:016x} step={step} action={action:?} timed out") + })?; + result.with_context(|| format!("seed=0x{seed:016x} step={step} action={action:?}"))?; + } + Ok(()) +} + +async fn run_matrix(first_seed: u64, seeds: usize, steps: usize) { + for index in 0..seeds { + let seed = first_seed.wrapping_add((index as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15)); + if let Err(error) = run_seed(seed, steps).await { + panic!( + "Signal durability chaos failed: {error:#}\n\ + replay with SIGNAL_CHAOS_SEED=0x{seed:016x} SIGNAL_CHAOS_SEEDS=1 \ + SIGNAL_CHAOS_STEPS={steps}" + ); + } + } +} + +#[tokio::test] +async fn signal_durability_chaos_smoke() { + run_matrix(DEFAULT_SEED, SMOKE_SEEDS, SMOKE_STEPS).await; +} + +#[tokio::test] +#[ignore = "run by signal-durability-nightly.yml"] +async fn signal_durability_chaos_nightly() { + let seed = env_u64("SIGNAL_CHAOS_SEED", DEFAULT_SEED); + let seeds = env_usize("SIGNAL_CHAOS_SEEDS", NIGHTLY_SEEDS); + let steps = env_usize("SIGNAL_CHAOS_STEPS", NIGHTLY_STEPS); + run_matrix(seed, seeds, steps).await; +}