From 2865f383f8e2cc2ac485e4ea4d0146b6a0d5522d Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 27 Jun 2026 19:15:01 -0500 Subject: [PATCH 1/4] fix: exit-for-restart on redb poison instead of bricking forever (#4604) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Problem ------- A single transient redb I/O error permanently and silently bricks a node's contract operations. redb latches an in-memory poison flag (`io_failed`) after any backend read/write error; from then on every `begin_write` returns `StorageError::PreviousIo` ("Previous I/O error occurred. Please close and re-open the database."), so every contract PUT/UPDATE (and the hosting-metadata write on essentially every access) fails forever. Because the process never exits, systemd never restarts it and the exit-42 auto-update hook never fires — the node stays "running" while 100% of contract ops fail. Observed on 0.2.84 (failing non-ECC RAM → btrfs csum EIO → redb poison), but any transient disk EIO / fs hiccup triggers it on healthy hardware too. Solution -------- Option (b): detect the poison at the storage-op layer and EXIT the process so the supervisor recycles the node with a fresh, un-poisoned database handle (and the health/update hooks fire). Chosen over in-process close+reopen (option a) because reopen does NOT interact with the crash-loop protection: on persistent corruption it would churn-reopen forever, recreating the exact "silently broken, supervisor never engages" failure this issue is about. - Precise detection: `storage_error_is_poison` matches only the I/O-poison class (`PreviousIo`, `Io`, `LockPoisoned`, `DatabaseClosed`) by typed variant, never the message string. Benign not-found (`Ok(None)`), `TableDoesNotExist`, `Corrupted`, `ValueTooLarge`, etc. are NOT treated as poison. - Choke point: `ReDb::begin_write` / `begin_read` wrappers route a poison error to the recovery path. `begin_write` is the reliable trigger — redb checks the poison flag on every `begin_write`, and the node writes on essentially every op. - The exit reuses the existing `fatal_listener_exit_code` decision (#4549/#4551), so it inherits the crash-loop protection unchanged: a poison after a healthy uptime exits 42 (burst-exempt, fires the update self-heal); a poison within the 60s healthy window (under a supervisor that understands it) exits the COUNTED code 45, so a database that re-poisons on every boot (persistent corruption) is bounded by `StartLimitBurst` and the #4591 auto-rollback probation instead of restart-looping forever. No new exit code is introduced. - Opt-in (`enable_abort_on_redb_poison`, called only by the real binary) so simulation/integration tests and library embedders never have a storage error kill their host process. Testing ------- - `redb_poison_classifier_is_precise`: produces REAL redb errors via a fault-injecting `StorageBackend` (genuine `Io` then `PreviousIo`) and asserts they classify as poison, while a benign table error does not. - `poisoned_redb_takes_recovery_path_benign_does_not`: a poisoned database routes a write op to the recovery path (observed via a test counter; abort is off in tests) while a benign not-found does not. - `redb_poison_exit_reuses_crash_loop_bounded_codes` / `redb_poison_abort_is_opt_in`: pin the counted-vs-burst-exempt exit-code interaction and the opt-in default. Needs a release to deploy. Closes #4604 Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_014dGjU1Q6Vpk2dm4sUf4pdU --- crates/core/src/bin/freenet.rs | 9 + crates/core/src/contract/storages/redb.rs | 344 +++++++++++++++++++--- crates/core/src/lib.rs | 2 +- crates/core/src/node.rs | 5 +- crates/core/src/node/p2p_impl.rs | 125 ++++++++ 5 files changed, 449 insertions(+), 36 deletions(-) diff --git a/crates/core/src/bin/freenet.rs b/crates/core/src/bin/freenet.rs index 1c981176f8..28057ae57a 100644 --- a/crates/core/src/bin/freenet.rs +++ b/crates/core/src/bin/freenet.rs @@ -255,6 +255,15 @@ async fn run_network_node_with_signals( freenet::enable_fast_crash_exit_code(); } + // #4604: this is the real node process, so let the contract storage layer + // exit-for-restart if redb gets poisoned by a transient I/O error (e.g. a disk + // EIO / filesystem csum failure). Without this the node stays "running" while + // every contract GET/PUT/UPDATE fails forever; with it, the supervisor restarts + // the node with a fresh database handle. The exit code reuses the fatal-listener + // decision above, so a database that re-poisons on every boot (persistent + // corruption) is bounded by the same crash-loop protection rather than looping. + freenet::enable_abort_on_redb_poison(); + // #4073 crash-loop auto-rollback: once this (possibly freshly auto-updated) // node has run healthily for the commit window, clear any post-update // probation marker so ordinary later crashes never trigger a rollback. The diff --git a/crates/core/src/contract/storages/redb.rs b/crates/core/src/contract/storages/redb.rs index f0c3311432..9891614e97 100644 --- a/crates/core/src/contract/storages/redb.rs +++ b/crates/core/src/contract/storages/redb.rs @@ -2,7 +2,10 @@ use std::path::Path; use std::sync::Arc; use freenet_stdlib::prelude::*; -use redb::{Database, DatabaseError, ReadableDatabase, ReadableTable, TableDefinition}; +use redb::{ + Database, DatabaseError, ReadTransaction, ReadableDatabase, ReadableTable, StorageError, + TableDefinition, TransactionError, WriteTransaction, +}; use crate::wasm_runtime::StateStorage; @@ -121,6 +124,51 @@ impl HostingMetadata { } } +/// Does this redb [`StorageError`] indicate the database is *poisoned* by an +/// underlying I/O failure — i.e. unusable until closed and reopened — rather than a +/// benign, app-level condition (a missing key returns `Ok(None)`; a missing/mismatched +/// table is a `TableError`, never one of these)? See issue #4604. +/// +/// redb sets an in-memory poison flag after any I/O error: the triggering op returns +/// [`StorageError::Io`], and EVERY subsequent transaction then returns +/// [`StorageError::PreviousIo`] ("Previous I/O error occurred. Please close and +/// re-open the database.") until the `Database` is dropped and re-created. +/// [`StorageError::LockPoisoned`] (a redb-internal mutex poisoned by a panic) and +/// [`StorageError::DatabaseClosed`] are likewise unrecoverable for the live handle. +/// +/// Matched against the typed variants (not the message string) so it cannot drift +/// with a redb wording change. `StorageError` is `#[non_exhaustive]`; `matches!` +/// keeps every OTHER (benign / app-level) error off the restart path — exactly the +/// precise-detection requirement of #4604. Notably `Corrupted`, `ValueTooLarge`, +/// and the table/type errors are NOT treated as poison. +fn storage_error_is_poison(e: &StorageError) -> bool { + matches!( + e, + StorageError::PreviousIo + | StorageError::Io(_) + | StorageError::LockPoisoned(_) + | StorageError::DatabaseClosed + ) +} + +/// True if a transaction-begin error (`begin_read` / `begin_write`) signals a +/// poisoned database. Once poisoned, EVERY `begin_write` returns +/// `TransactionError::Storage(StorageError::PreviousIo)`, so this is the universal +/// post-poison choke point the #4604 fix keys off. (`ReadTransactionStillInUse` and +/// any future non-storage variant are benign usage errors, not a poison.) +fn transaction_error_is_poison(e: &TransactionError) -> bool { + matches!(e, TransactionError::Storage(s) if storage_error_is_poison(s)) +} + +/// Test-only observable proof that the storage layer routed a detected poison to +/// the recovery (process-exit) path. The real handler ([`abort_process_on_redb_poison`]) +/// exits the process, which a unit test cannot observe; this counter lets the test +/// assert the wrapper recognised poison (and would have exited in production) without +/// killing the test process. +#[cfg(test)] +static POISON_RECOVERY_TRIGGERED: std::sync::atomic::AtomicUsize = + std::sync::atomic::AtomicUsize::new(0); + /// ReDb wraps a redb Database in Arc for thread-safe sharing. /// redb supports MVCC (multiple concurrent readers, single writer) internally, /// so multiple clones of ReDb can safely access the same database. @@ -128,6 +176,42 @@ impl HostingMetadata { pub struct ReDb(Arc); impl ReDb { + /// Begin a write transaction, routing a *poisoned*-database error to the #4604 + /// recovery path (process exit for a supervised restart with a fresh handle) so + /// the node does not fail every contract op forever while looking "running". A + /// benign error is returned unchanged. + /// + /// This is the RELIABLE post-poison choke point: redb latches an in-memory poison + /// flag (`io_failed`) on ANY backend read OR write error, and `begin_write` checks + /// it on every call (returning `PreviousIo`). Because the node writes hosting + /// metadata on essentially every contract access, a poisoned database is detected + /// here within one write — whatever the original error was a read or a write. + fn begin_write(&self) -> Result { + self.0.begin_write().map_err(Self::route_txn_error) + } + + /// Begin a read transaction with the same poison-recovery routing as + /// [`ReDb::begin_write`]. Note redb's `begin_read` does NOT itself check the + /// poison flag (it serves the last committed snapshot from cache), so it only + /// surfaces poison when the read transaction registration itself does I/O; a + /// poisoned read that reaches the backend fails later inside the transaction. + /// Either way the next `begin_write` (above) catches the poison promptly. + fn begin_read(&self) -> Result { + self.0.begin_read().map_err(Self::route_txn_error) + } + + /// If `e` indicates a poisoned database, trigger the #4604 recovery path + /// ([`abort_process_on_redb_poison`], a no-op outside the real node binary). + /// Always returns the error untouched so callers still propagate it. + fn route_txn_error(e: TransactionError) -> TransactionError { + if transaction_error_is_poison(&e) { + #[cfg(test)] + POISON_RECOVERY_TRIGGERED.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + crate::node::abort_process_on_redb_poison(&e.to_string()); + } + e + } + pub async fn new(data_dir: &Path) -> Result { let db_path = data_dir.join("db"); tracing::info!( @@ -337,7 +421,7 @@ impl ReDb { key: &ContractKey, metadata: HostingMetadata, ) -> Result<(), redb::Error> { - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(HOSTING_METADATA_TABLE)?; tbl.insert(key.as_bytes(), metadata.to_bytes().as_slice())?; @@ -350,7 +434,7 @@ impl ReDb { &self, key: &ContractKey, ) -> Result, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(HOSTING_METADATA_TABLE)?; match tbl.get(key.as_bytes())? { Some(v) => Ok(HostingMetadata::from_bytes(v.value())), @@ -360,7 +444,7 @@ impl ReDb { /// Remove hosting metadata for a contract. pub fn remove_hosting_metadata(&self, key: &ContractKey) -> Result<(), redb::Error> { - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(HOSTING_METADATA_TABLE)?; tbl.remove(key.as_bytes())?; @@ -374,7 +458,7 @@ impl ReDb { pub fn load_all_hosting_metadata( &self, ) -> Result, HostingMetadata)>, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(HOSTING_METADATA_TABLE)?; let mut result = Vec::new(); @@ -389,7 +473,7 @@ impl ReDb { /// Get the size of a contract's state (for populating hosting cache). pub fn get_state_size(&self, key: &ContractKey) -> Result, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(STATE_TABLE)?; match tbl.get(key.as_bytes())? { Some(v) => Ok(Some(v.value().len() as u64)), @@ -411,7 +495,7 @@ impl ReDb { key: &ContractKey, state: WrappedState, ) -> Result<(), redb::Error> { - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(STATE_TABLE)?; tbl.insert(key.as_bytes(), state.as_ref())?; @@ -431,7 +515,7 @@ impl ReDb { key: &ContractKey, state: WrappedState, ) -> Result { - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(STATE_TABLE)?; // Check existence within the same write transaction @@ -451,7 +535,7 @@ impl ReDb { /// Used by V2 delegate host functions that need synchronous access during /// WASM `process()` execution. pub fn get_state_sync(&self, key: &ContractKey) -> Result, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let val = { let tbl = txn.open_table(STATE_TABLE)?; tbl.get(key.as_bytes())? @@ -465,7 +549,7 @@ impl ReDb { /// Iterate all contract keys that have stored state. /// Returns the raw key bytes - caller must reconstruct ContractKey. pub fn iter_all_state_keys(&self) -> Result>, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(STATE_TABLE)?; let mut result = Vec::new(); @@ -485,7 +569,7 @@ impl ReDb { instance_id: &ContractInstanceId, code_hash: &CodeHash, ) -> Result<(), redb::Error> { - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(CONTRACT_INDEX_TABLE)?; tbl.insert(instance_id.as_ref(), code_hash.as_ref())?; @@ -498,7 +582,7 @@ impl ReDb { &self, instance_id: &ContractInstanceId, ) -> Result, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(CONTRACT_INDEX_TABLE)?; match tbl.get(instance_id.as_ref())? { Some(v) => { @@ -519,7 +603,7 @@ impl ReDb { &self, instance_id: &ContractInstanceId, ) -> Result<(), redb::Error> { - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(CONTRACT_INDEX_TABLE)?; tbl.remove(instance_id.as_ref())?; @@ -531,7 +615,7 @@ impl ReDb { pub fn load_all_contract_index( &self, ) -> Result, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(CONTRACT_INDEX_TABLE)?; let mut result = Vec::new(); @@ -571,7 +655,7 @@ impl ReDb { key_bytes[..32].copy_from_slice(delegate_key.as_ref()); key_bytes[32..].copy_from_slice(delegate_key.code_hash().as_ref()); - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(DELEGATE_INDEX_TABLE)?; tbl.insert(key_bytes.as_slice(), code_hash.as_ref())?; @@ -588,7 +672,7 @@ impl ReDb { key_bytes[..32].copy_from_slice(delegate_key.as_ref()); key_bytes[32..].copy_from_slice(delegate_key.code_hash().as_ref()); - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(DELEGATE_INDEX_TABLE)?; match tbl.get(key_bytes.as_slice())? { Some(v) => { @@ -610,7 +694,7 @@ impl ReDb { key_bytes[..32].copy_from_slice(delegate_key.as_ref()); key_bytes[32..].copy_from_slice(delegate_key.code_hash().as_ref()); - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(DELEGATE_INDEX_TABLE)?; tbl.remove(key_bytes.as_slice())?; @@ -620,7 +704,7 @@ impl ReDb { /// Load all delegate index entries pub fn load_all_delegate_index(&self) -> Result, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(DELEGATE_INDEX_TABLE)?; let mut result = Vec::new(); @@ -665,7 +749,7 @@ impl ReDb { value_bytes.extend_from_slice(sk); } - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(SECRETS_INDEX_TABLE)?; tbl.insert(key_bytes.as_slice(), value_bytes.as_slice())?; @@ -682,7 +766,7 @@ impl ReDb { key_bytes[..32].copy_from_slice(delegate_key.as_ref()); key_bytes[32..].copy_from_slice(delegate_key.code_hash().as_ref()); - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(SECRETS_INDEX_TABLE)?; match tbl.get(key_bytes.as_slice())? { Some(v) => { @@ -710,7 +794,7 @@ impl ReDb { key_bytes[..32].copy_from_slice(delegate_key.as_ref()); key_bytes[32..].copy_from_slice(delegate_key.code_hash().as_ref()); - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(SECRETS_INDEX_TABLE)?; tbl.remove(key_bytes.as_slice())?; @@ -721,7 +805,7 @@ impl ReDb { /// Load all secrets index entries #[allow(clippy::type_complexity)] pub fn load_all_secrets_index(&self) -> Result)>, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(SECRETS_INDEX_TABLE)?; let mut result = Vec::new(); @@ -780,7 +864,7 @@ impl ReDb { value_bytes.extend_from_slice(sk); } - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; tbl.insert(key_bytes.as_slice(), value_bytes.as_slice())?; @@ -801,7 +885,7 @@ impl ReDb { ) -> Result>, redb::Error> { let key_bytes = Self::user_index_key(delegate_key, user_id); - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; match tbl.get(key_bytes.as_slice())? { Some(v) => { @@ -832,7 +916,7 @@ impl ReDb { ) -> Result<(), redb::Error> { let key_bytes = Self::user_index_key(delegate_key, user_id); - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; tbl.remove(key_bytes.as_slice())?; @@ -846,7 +930,7 @@ impl ReDb { pub fn load_all_user_secrets_index( &self, ) -> Result)>, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; let mut result = Vec::new(); @@ -909,7 +993,7 @@ impl ReDb { instance_id: &ContractInstanceId, kind_byte: u8, ) -> Result<(), redb::Error> { - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(BROKEN_INVARIANTS_TABLE)?; tbl.insert(instance_id.as_ref(), &[kind_byte][..])?; @@ -925,7 +1009,7 @@ impl ReDb { &self, instance_id: &ContractInstanceId, ) -> Result<(), redb::Error> { - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(BROKEN_INVARIANTS_TABLE)?; tbl.remove(instance_id.as_ref())?; @@ -938,7 +1022,7 @@ impl ReDb { /// than failing the entire load — a corrupted entry should not block /// startup, and the worst case is we lose a flag and re-detect it. pub fn load_all_broken_invariants(&self) -> Result, redb::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let tbl = txn.open_table(BROKEN_INVARIANTS_TABLE)?; let mut result = Vec::new(); @@ -973,7 +1057,7 @@ impl StateStorage for ReDb { async fn store(&self, key: ContractKey, state: WrappedState) -> Result<(), Self::Error> { let state_size = state.size() as u64; - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(STATE_TABLE)?; @@ -1007,7 +1091,7 @@ impl StateStorage for ReDb { } async fn get(&self, key: &ContractKey) -> Result, Self::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let val = { let tbl = txn.open_table(STATE_TABLE)?; @@ -1025,7 +1109,7 @@ impl StateStorage for ReDb { key: ContractKey, params: Parameters<'static>, ) -> Result<(), Self::Error> { - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(CONTRACT_PARAMS_TABLE)?; @@ -1038,7 +1122,7 @@ impl StateStorage for ReDb { &'a self, key: &'a ContractKey, ) -> Result>, Self::Error> { - let txn = self.0.begin_read()?; + let txn = self.begin_read()?; let val = { let tbl = txn.open_table(CONTRACT_PARAMS_TABLE)?; @@ -1055,7 +1139,7 @@ impl StateStorage for ReDb { // Delete from all three per-key tables in a single write transaction // so the removal is atomic. `redb`'s `Table::remove` does not error // when the key is absent, so this is naturally idempotent. - let txn = self.0.begin_write()?; + let txn = self.begin_write()?; { let mut tbl = txn.open_table(STATE_TABLE)?; tbl.remove(key.as_bytes())?; @@ -1493,4 +1577,196 @@ mod tests { "malformed key/value rows must be skipped, leaving only the good row" ); } + + // ==================== #4604: redb poison-recovery ==================== + + /// A redb [`redb::StorageBackend`] over an in-memory buffer that can be flipped + /// to return `io::Error` from every I/O method. Used to produce a REAL redb + /// poison deterministically (a genuine `StorageError::Io` that makes redb set its + /// in-memory poison flag, after which every transaction returns + /// `StorageError::PreviousIo`) so the poison-detection and recovery path can be + /// exercised without relying on the error message string. + #[derive(Debug, Clone)] + struct FailingBackend { + inner: Arc, + fail: Arc, + } + + impl FailingBackend { + fn new() -> Self { + Self { + inner: Arc::new(redb::backends::InMemoryBackend::new()), + fail: Arc::new(std::sync::atomic::AtomicBool::new(false)), + } + } + + /// Make every subsequent I/O call fail, simulating a disk EIO / csum failure. + fn start_failing(&self) { + self.fail.store(true, std::sync::atomic::Ordering::SeqCst); + } + + fn check(&self) -> std::io::Result<()> { + if self.fail.load(std::sync::atomic::Ordering::SeqCst) { + Err(std::io::Error::other( + "injected I/O failure (#4604 redb-poison test)", + )) + } else { + Ok(()) + } + } + } + + impl redb::StorageBackend for FailingBackend { + fn len(&self) -> std::io::Result { + self.check()?; + self.inner.len() + } + fn read(&self, offset: u64, out: &mut [u8]) -> std::io::Result<()> { + self.check()?; + self.inner.read(offset, out) + } + fn set_len(&self, len: u64) -> std::io::Result<()> { + self.check()?; + self.inner.set_len(len) + } + fn sync_data(&self) -> std::io::Result<()> { + self.check()?; + self.inner.sync_data() + } + fn write(&self, offset: u64, data: &[u8]) -> std::io::Result<()> { + self.check()?; + self.inner.write(offset, data) + } + } + + /// Open a fully-initialised [`ReDb`] over an arbitrary backend (test-only). + fn open_redb_with_backend(backend: B) -> ReDb { + let db = Database::builder() + .create_with_backend(backend) + .expect("create_with_backend"); + ReDb::initialize_database(db).expect("initialize_database") + } + + /// Poison detection must be PRECISE (issue #4604, requirement 1): it must fire on + /// the real underlying-I/O / poison errors and NOT on benign app-level errors. + /// Uses REAL redb errors produced via the fault-injecting backend, so it is + /// resilient to redb wording changes (we match variants, not strings). + #[test] + fn redb_poison_classifier_is_precise() { + let backend = FailingBackend::new(); + let db = Database::builder() + .create_with_backend(backend.clone()) + .unwrap(); + { + let w = db.begin_write().unwrap(); + w.open_table(STATE_TABLE).unwrap(); + w.commit().unwrap(); + } + + // Benign: opening a non-existent table is a TableError, never an I/O poison. + // (TableDoesNotExist is not a `Storage(..)` variant at all, so it can never + // classify as poison; a `Storage(_)` here would still have to be non-poison.) + { + let r = db.begin_read().unwrap(); + let missing: TableDefinition<&[u8], &[u8]> = TableDefinition::new("nope"); + if let redb::TableError::Storage(s) = r.open_table(missing).unwrap_err() { + assert!( + !storage_error_is_poison(&s), + "a benign table-open storage error must not classify as poison" + ); + } + } + + // Trigger a REAL I/O failure → the triggering op returns StorageError::Io, + // which must classify as poison (the "underlying-I/O-error class"). + backend.start_failing(); + // The injected I/O error surfaces either when `begin_write` does I/O or, more + // commonly, at `commit` — capture whichever StorageError it is. + let storage_err: StorageError = match db.begin_write() { + // begin itself may hit the injected I/O error first. + Err(TransactionError::Storage(s)) => s, + Err(other) => panic!("unexpected begin error: {other:?}"), + Ok(w) => { + { + let mut t = w.open_table(STATE_TABLE).unwrap(); + // Buffered write; the backend failure surfaces at commit. + let _insert = t.insert([1u8, 2, 3].as_slice(), [4u8, 5, 6].as_slice()); + } + match w.commit() { + Ok(()) => { + panic!("commit unexpectedly succeeded while backend was failing") + } + Err(redb::CommitError::Storage(s)) => s, + Err(other) => panic!("unexpected commit error: {other:?}"), + } + } + }; + assert!( + storage_error_is_poison(&storage_err), + "the underlying I/O error (StorageError::Io) must classify as poison" + ); + + // redb is now poisoned: every begin returns PreviousIo, which the universal + // begin_* choke point classifies as poison. + let begin_err = match db.begin_write() { + Ok(_) => panic!("a poisoned database must reject begin_write"), + Err(e) => e, + }; + assert!( + transaction_error_is_poison(&begin_err), + "PreviousIo from a poisoned database's begin_write must classify as poison" + ); + } + + /// End-to-end (issue #4604, requirement 3): a poisoned database routes contract + /// ops to the recovery path (process-exit-for-restart in production) rather than + /// failing forever, while a benign not-found does NOT. The recovery handler is + /// opt-in and OFF in tests, so it returns instead of exiting; the test-only + /// counter proves the `begin_*` wrapper recognised the poison and would have + /// exited under the real node binary. + #[test] + fn poisoned_redb_takes_recovery_path_benign_does_not() { + use std::sync::atomic::Ordering; + + let backend = FailingBackend::new(); + let db = open_redb_with_backend(backend.clone()); + let key = make_test_key(); + + // Benign not-found: Ok(None), recovery path NOT taken. + POISON_RECOVERY_TRIGGERED.store(0, Ordering::SeqCst); + assert!(db.get_state_sync(&key).unwrap().is_none()); + db.store_state_sync(&key, WrappedState::new(vec![1, 2, 3])) + .unwrap(); + assert_eq!( + POISON_RECOVERY_TRIGGERED.load(Ordering::SeqCst), + 0, + "benign not-found / normal ops must NOT take the poison-recovery path" + ); + + // Poison the backend; the next write fails (I/O), and redb latches its + // in-memory poison flag (io_failed) — set on ANY backend read/write error. + backend.start_failing(); + assert!( + db.store_state_sync(&key, WrappedState::new(vec![4, 5, 6])) + .is_err(), + "the injected I/O failure must surface as an error" + ); + + // The database is now poisoned. redb's poison flag is checked by every + // `begin_write` (the node writes hosting metadata on essentially every + // contract op), so the next write returns PreviousIo, which the begin_write + // wrapper routes to the recovery (exit-for-restart) path instead of the old + // fail-forever behaviour. + POISON_RECOVERY_TRIGGERED.store(0, Ordering::SeqCst); + assert!( + db.store_state_sync(&key, WrappedState::new(vec![7, 8, 9])) + .is_err(), + "a poisoned write must return an error, not silently no-op" + ); + assert!( + POISON_RECOVERY_TRIGGERED.load(Ordering::SeqCst) >= 1, + "a poisoned database write must take the recovery (exit-for-restart) path \ + rather than failing forever" + ); + } } diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index d69f681dfe..7d9d09b837 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -21,7 +21,7 @@ mod message; mod node; pub use node::{ EventLoopExitReason, Node, ShutdownHandle, enable_abort_on_fatal_listener_exit, - enable_fast_crash_exit_code, run_local_node, run_network_node, + enable_abort_on_redb_poison, enable_fast_crash_exit_code, run_local_node, run_network_node, }; /// Network operation/transaction state machines. diff --git a/crates/core/src/node.rs b/crates/core/src/node.rs index e095d81129..90dfb192a8 100644 --- a/crates/core/src/node.rs +++ b/crates/core/src/node.rs @@ -80,7 +80,10 @@ mod p2p_impl; mod request_router; pub(crate) mod testing_impl; -pub use p2p_impl::{enable_abort_on_fatal_listener_exit, enable_fast_crash_exit_code}; +pub(crate) use p2p_impl::abort_process_on_redb_poison; +pub use p2p_impl::{ + enable_abort_on_fatal_listener_exit, enable_abort_on_redb_poison, enable_fast_crash_exit_code, +}; pub use request_router::{DeduplicatedRequest, RequestRouter}; /// Handle to trigger graceful shutdown of the node. diff --git a/crates/core/src/node/p2p_impl.rs b/crates/core/src/node/p2p_impl.rs index 10815e77b7..9e906604ab 100644 --- a/crates/core/src/node/p2p_impl.rs +++ b/crates/core/src/node/p2p_impl.rs @@ -139,6 +139,83 @@ pub fn enable_fast_crash_exit_code() { EMIT_FAST_CRASH_EXIT_CODE.store(true, std::sync::atomic::Ordering::Relaxed); } +/// When `true`, the contract storage layer (redb) reacting to a *poisoned* +/// database — one left unusable by a transient underlying I/O error — exits the +/// process so a supervisor restart hands the node a fresh, un-poisoned database +/// handle (#4604). Off by default so simulation / integration tests and library +/// embedders never have a storage error kill their host process; only the real +/// `freenet` binary opts in via [`enable_abort_on_redb_poison`]. +static ABORT_ON_REDB_POISON: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(false); + +/// Process-start instant, captured the first time [`enable_abort_on_redb_poison`] +/// is called (at node entry). Used to pick the poison exit code: a poison that +/// occurs within [`MIN_HEALTHY_UPTIME_FOR_UPDATE_EXIT`] of start is treated as a +/// *fast crash* so a node whose database re-poisons on every boot (persistent +/// disk / RAM corruption) is bounded by the supervisor's crash-loop protection +/// instead of restart-looping forever (#4604, #4591, #4588). +static REDB_POISON_PROCESS_START: std::sync::OnceLock = + std::sync::OnceLock::new(); + +/// Enable the #4604 redb-poison fast-exit. Once the contract storage layer detects +/// that the redb database is poisoned (a transient I/O error left redb returning +/// "Previous I/O error occurred. Please close and re-open the database." from every +/// transaction), the process exits so the service manager restarts it with a fresh +/// handle and the health / auto-update hooks fire — instead of the node staying +/// "running" while 100% of contract operations fail forever. Call once from the +/// production node entry point. See [`ABORT_ON_REDB_POISON`]. +pub fn enable_abort_on_redb_poison() { + REDB_POISON_PROCESS_START.get_or_init(std::time::Instant::now); + ABORT_ON_REDB_POISON.store(true, std::sync::atomic::Ordering::Relaxed); +} + +/// React to a detected redb poison (#4604): exit the process so a supervised +/// restart gives the node a fresh database handle. No-op when the fast-exit is not +/// enabled (tests / embedders), so the caller simply propagates the underlying +/// error as before. +/// +/// The exit code reuses the SAME [`fatal_listener_exit_code`] decision as the +/// #4549 / #4551 fatal-listener path, so the poison exit integrates with the +/// existing crash-loop protection without adding a second policy: +/// - poison after a healthy uptime (>= [`MIN_HEALTHY_UPTIME_FOR_UPDATE_EXIT`]) → +/// [`FATAL_LISTENER_EXIT_CODE`] (42): burst-exempt, fires the `freenet update` +/// self-heal hook. A genuinely transient I/O blip (the #4604 case: a node that +/// ran healthily for hours) restarts cleanly without tripping `StartLimitBurst`. +/// - poison within the healthy-uptime window, when the supervisor understands it → +/// [`FAST_CRASH_EXIT_CODE`] (45): a *counted* failure. A database that re-poisons +/// on every boot (persistent corruption) trips `StartLimitBurst` and is counted +/// by the #4591 auto-rollback probation, so the unit is stopped / rolled back +/// instead of looping forever. +pub(crate) fn abort_process_on_redb_poison(reason: &str) { + if !ABORT_ON_REDB_POISON.load(std::sync::atomic::Ordering::Relaxed) { + return; + } + let uptime = REDB_POISON_PROCESS_START + .get() + .map(std::time::Instant::elapsed) + .unwrap_or_default(); + let exit_code = fatal_listener_exit_code( + uptime, + EMIT_FAST_CRASH_EXIT_CODE.load(std::sync::atomic::Ordering::Relaxed), + ); + eprintln!( + "CRITICAL: contract storage (redb) is poisoned by an I/O error and cannot \ + recover in-process: {reason}. Exiting with code {exit_code} so the service \ + manager restarts the node with a fresh database handle (#4604)." + ); + tracing::error!( + error = %reason, + exit_code, + uptime_secs = uptime.as_secs(), + fast_crash = exit_code == FAST_CRASH_EXIT_CODE, + "redb storage poisoned; forcing process exit for a supervised restart with a \ + fresh database handle (#4604). A persistent-corruption re-poison within the \ + healthy-uptime window exits with the counted fast-crash code so the \ + crash-loop protection bounds the restart loop instead of looping forever." + ); + std::process::exit(exit_code); +} + /// When `true`, a fatal (non-graceful) exit of the network event listener aborts /// the whole process with [`FATAL_LISTENER_EXIT_CODE`] instead of unwinding through /// the normal teardown/return path (#4549). @@ -1054,6 +1131,54 @@ mod tests { ); } + /// #4604: the redb-poison fast-exit reuses [`fatal_listener_exit_code`], so it + /// inherits the crash-loop protection exactly: + /// - A poison after a *healthy* uptime exits 42 (burst-exempt, fires the + /// auto-update hook): a one-off transient I/O blip restarts cleanly. + /// - A poison *within* the healthy window, under a supervisor that understands + /// it, exits the COUNTED code 45 — so a database that re-poisons on every boot + /// (persistent disk / RAM corruption) trips `StartLimitBurst` and the #4591 + /// auto-rollback probation count, bounding the restart loop instead of looping + /// forever. This is the interaction required by issue #4604. + #[test] + fn redb_poison_exit_reuses_crash_loop_bounded_codes() { + // Persistent corruption that re-poisons immediately on boot → counted code + // so the supervisor's crash-loop protection eventually stops / rolls back. + for secs in [0u64, 1, 30, 59] { + assert_eq!( + fatal_listener_exit_code(Duration::from_secs(secs), true), + FAST_CRASH_EXIT_CODE, + "a redb re-poison {secs}s after boot must use the COUNTED fast-crash \ + code so a corrupt node does not restart-loop forever (#4604/#4591)" + ); + } + // A genuinely transient poison after a healthy run → burst-exempt update + // code so the restart self-heals and is never rate-limited. + assert_eq!( + fatal_listener_exit_code(MIN_HEALTHY_UPTIME_FOR_UPDATE_EXIT, true), + FATAL_LISTENER_EXIT_CODE, + "a transient poison after a healthy uptime restarts via the burst-exempt \ + update code (the #4604 real-world case: ran healthily for hours)" + ); + // Both codes are ones the supervisor already handles (declared in the + // generated systemd unit), so no new exit-code contract is introduced. + assert_eq!(FATAL_LISTENER_EXIT_CODE, 42); + assert_eq!(FAST_CRASH_EXIT_CODE, 45); + } + + /// The redb-poison fast-exit must be OFF by default so simulation / integration + /// tests and library embedders never have a storage error kill their process. + /// Only the real binary opts in via [`enable_abort_on_redb_poison`]. + #[test] + fn redb_poison_abort_is_opt_in() { + // NOTE: deliberately does NOT call `enable_abort_on_redb_poison()` (a + // process-global flip would affect other tests in the same binary). + assert!( + !ABORT_ON_REDB_POISON.load(std::sync::atomic::Ordering::Relaxed), + "redb-poison fast-exit must be opt-in (off unless the binary enables it)" + ); + } + /// Verify that a spawned task that panics is detected via JoinHandle. #[tokio::test] async fn test_join_handle_detects_panic() { From f61eee554201223a94d77e703f9ea3c6446defd3 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 27 Jun 2026 19:32:31 -0500 Subject: [PATCH 2/4] fix: also route read-path redb poison to recovery (#4604 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex review of the initial fix flagged that the begin_* wrappers only catch poison at transaction start. redb's `begin_read` does NOT check the poison flag (it serves the last committed snapshot from cache), so a poisoned read fails later at `open_table`/`get`/iteration — a read-only workload could keep erroring until some later write hit `begin_write`. Close the gap: a `read_guarded` helper wraps every read method's body and routes any poison error (surfacing after begin_read) to the same recovery path. Crucially, the umbrella read-path classifier (`redb_error_is_poison`) matches `PreviousIo | LockPoisoned | DatabaseClosed` but NOT `Io` — several read methods SYNTHESIZE `redb::Error::Io(InvalidData)` for a benign malformed-data row (e.g. a wrong-length CodeHash), and treating that as poison would exit-and-restart the node on a single bad row (a crash loop). A genuine backend I/O poison is still caught: redb latches its flag on the first backend Io, so the next backend read returns `PreviousIo` (caught here) and the next `begin_write` returns `PreviousIo` (caught by the transaction-path classifier, which keeps matching `Io` since redb never synthesizes Io there). The classifier test pins both directions, including that a synthetic `Io(InvalidData)` is NOT treated as poison. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_014dGjU1Q6Vpk2dm4sUf4pdU --- crates/core/src/contract/storages/redb.rs | 494 ++++++++++++---------- 1 file changed, 281 insertions(+), 213 deletions(-) diff --git a/crates/core/src/contract/storages/redb.rs b/crates/core/src/contract/storages/redb.rs index 9891614e97..e136dabae6 100644 --- a/crates/core/src/contract/storages/redb.rs +++ b/crates/core/src/contract/storages/redb.rs @@ -160,6 +160,29 @@ fn transaction_error_is_poison(e: &TransactionError) -> bool { matches!(e, TransactionError::Storage(s) if storage_error_is_poison(s)) } +/// True if the umbrella [`redb::Error`] signals a poisoned database. Used to catch +/// poison that surfaces on the READ path AFTER `begin_read` has already succeeded +/// (redb's `begin_read` does not check the poison flag, so a poisoned read fails +/// later at `open_table` / `get` / iteration as a `StorageError` flattened into this +/// umbrella type). +/// +/// IMPORTANT — this deliberately does NOT match `redb::Error::Io`, unlike +/// [`storage_error_is_poison`]. Several read methods in this file SYNTHESIZE +/// `redb::Error::Io(ErrorKind::InvalidData)` for a benign malformed-data row (e.g. a +/// wrong-length `CodeHash`). Treating that as poison would exit-and-restart the node +/// on a single bad row → a crash loop. A genuine backend I/O poison is safe to skip +/// here regardless: redb latches its poison flag on the first backend `Io`, so the +/// very next backend read returns `PreviousIo` (caught here) and the next +/// `begin_write` returns `PreviousIo` (caught by [`storage_error_is_poison`]). So +/// `PreviousIo` (plus the unambiguous `LockPoisoned` / `DatabaseClosed`) is the +/// precise, false-positive-free read-path signal. +fn redb_error_is_poison(e: &redb::Error) -> bool { + matches!( + e, + redb::Error::PreviousIo | redb::Error::LockPoisoned(_) | redb::Error::DatabaseClosed + ) +} + /// Test-only observable proof that the storage layer routed a detected poison to /// the recovery (process-exit) path. The real handler ([`abort_process_on_redb_poison`]) /// exits the process, which a unit test cannot observe; this counter lets the test @@ -212,6 +235,34 @@ impl ReDb { e } + /// Umbrella-error counterpart of [`ReDb::route_txn_error`], for poison that + /// surfaces on the read path AFTER `begin_read` succeeded (at `open_table` / + /// `get` / iteration). Returns the error untouched. + fn route_redb_error(e: redb::Error) -> redb::Error { + if redb_error_is_poison(&e) { + #[cfg(test)] + POISON_RECOVERY_TRIGGERED.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + crate::node::abort_process_on_redb_poison(&e.to_string()); + } + e + } + + /// Run a read-transaction body, routing ANY poison error to the #4604 recovery + /// path — not just a poison at `begin_read`, but one that surfaces later at + /// `open_table` / `get` / iteration (redb's `begin_read` does not check the + /// poison flag, so a poisoned read can fail mid-transaction). This gives + /// read-only workloads the same prompt exit-for-restart that `begin_write` + /// already gives write workloads, instead of failing every read until the next + /// write happens to hit `begin_write`. + fn read_guarded( + &self, + f: impl FnOnce(&ReadTransaction) -> Result, + ) -> Result { + // begin_read already routes a poison at transaction start. + let txn = self.begin_read()?; + f(&txn).map_err(Self::route_redb_error) + } + pub async fn new(data_dir: &Path) -> Result { let db_path = data_dir.join("db"); tracing::info!( @@ -434,12 +485,13 @@ impl ReDb { &self, key: &ContractKey, ) -> Result, redb::Error> { - let txn = self.begin_read()?; - let tbl = txn.open_table(HOSTING_METADATA_TABLE)?; - match tbl.get(key.as_bytes())? { - Some(v) => Ok(HostingMetadata::from_bytes(v.value())), - None => Ok(None), - } + self.read_guarded(|txn| { + let tbl = txn.open_table(HOSTING_METADATA_TABLE)?; + Ok(match tbl.get(key.as_bytes())? { + Some(v) => HostingMetadata::from_bytes(v.value()), + None => None, + }) + }) } /// Remove hosting metadata for a contract. @@ -458,27 +510,26 @@ impl ReDb { pub fn load_all_hosting_metadata( &self, ) -> Result, HostingMetadata)>, redb::Error> { - let txn = self.begin_read()?; - let tbl = txn.open_table(HOSTING_METADATA_TABLE)?; - - let mut result = Vec::new(); - for entry in tbl.iter()? { - let (key, value) = entry?; - if let Some(metadata) = HostingMetadata::from_bytes(value.value()) { - result.push((key.value().to_vec(), metadata)); + self.read_guarded(|txn| { + let tbl = txn.open_table(HOSTING_METADATA_TABLE)?; + + let mut result = Vec::new(); + for entry in tbl.iter()? { + let (key, value) = entry?; + if let Some(metadata) = HostingMetadata::from_bytes(value.value()) { + result.push((key.value().to_vec(), metadata)); + } } - } - Ok(result) + Ok(result) + }) } /// Get the size of a contract's state (for populating hosting cache). pub fn get_state_size(&self, key: &ContractKey) -> Result, redb::Error> { - let txn = self.begin_read()?; - let tbl = txn.open_table(STATE_TABLE)?; - match tbl.get(key.as_bytes())? { - Some(v) => Ok(Some(v.value().len() as u64)), - None => Ok(None), - } + self.read_guarded(|txn| { + let tbl = txn.open_table(STATE_TABLE)?; + Ok(tbl.get(key.as_bytes())?.map(|v| v.value().len() as u64)) + }) } /// Store a contract's state synchronously. @@ -535,29 +586,27 @@ impl ReDb { /// Used by V2 delegate host functions that need synchronous access during /// WASM `process()` execution. pub fn get_state_sync(&self, key: &ContractKey) -> Result, redb::Error> { - let txn = self.begin_read()?; - let val = { + self.read_guarded(|txn| { let tbl = txn.open_table(STATE_TABLE)?; - tbl.get(key.as_bytes())? - }; - match val { - Some(v) => Ok(Some(WrappedState::new(v.value().to_vec()))), - None => Ok(None), - } + Ok(tbl + .get(key.as_bytes())? + .map(|v| WrappedState::new(v.value().to_vec()))) + }) } /// Iterate all contract keys that have stored state. /// Returns the raw key bytes - caller must reconstruct ContractKey. pub fn iter_all_state_keys(&self) -> Result>, redb::Error> { - let txn = self.begin_read()?; - let tbl = txn.open_table(STATE_TABLE)?; + self.read_guarded(|txn| { + let tbl = txn.open_table(STATE_TABLE)?; - let mut result = Vec::new(); - for entry in tbl.iter()? { - let (key, _) = entry?; - result.push(key.value().to_vec()); - } - Ok(result) + let mut result = Vec::new(); + for entry in tbl.iter()? { + let (key, _) = entry?; + result.push(key.value().to_vec()); + } + Ok(result) + }) } // ==================== Contract Index Methods ==================== @@ -582,20 +631,21 @@ impl ReDb { &self, instance_id: &ContractInstanceId, ) -> Result, redb::Error> { - let txn = self.begin_read()?; - let tbl = txn.open_table(CONTRACT_INDEX_TABLE)?; - match tbl.get(instance_id.as_ref())? { - Some(v) => { - let bytes: [u8; 32] = v.value().try_into().map_err(|_| { - redb::Error::Io(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "Invalid CodeHash length", - )) - })?; - Ok(Some(CodeHash::from(&bytes))) + self.read_guarded(|txn| { + let tbl = txn.open_table(CONTRACT_INDEX_TABLE)?; + match tbl.get(instance_id.as_ref())? { + Some(v) => { + let bytes: [u8; 32] = v.value().try_into().map_err(|_| { + redb::Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Invalid CodeHash length", + )) + })?; + Ok(Some(CodeHash::from(&bytes))) + } + None => Ok(None), } - None => Ok(None), - } + }) } /// Remove a contract index entry @@ -615,30 +665,31 @@ impl ReDb { pub fn load_all_contract_index( &self, ) -> Result, redb::Error> { - let txn = self.begin_read()?; - let tbl = txn.open_table(CONTRACT_INDEX_TABLE)?; + self.read_guarded(|txn| { + let tbl = txn.open_table(CONTRACT_INDEX_TABLE)?; - let mut result = Vec::new(); - for entry in tbl.iter()? { - let (key, value) = entry?; - let key_bytes: [u8; 32] = key.value().try_into().map_err(|_| { - redb::Error::Io(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "Invalid ContractInstanceId length", - )) - })?; - let value_bytes: [u8; 32] = value.value().try_into().map_err(|_| { - redb::Error::Io(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "Invalid CodeHash length", - )) - })?; - result.push(( - ContractInstanceId::new(key_bytes), - CodeHash::from(&value_bytes), - )); - } - Ok(result) + let mut result = Vec::new(); + for entry in tbl.iter()? { + let (key, value) = entry?; + let key_bytes: [u8; 32] = key.value().try_into().map_err(|_| { + redb::Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Invalid ContractInstanceId length", + )) + })?; + let value_bytes: [u8; 32] = value.value().try_into().map_err(|_| { + redb::Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Invalid CodeHash length", + )) + })?; + result.push(( + ContractInstanceId::new(key_bytes), + CodeHash::from(&value_bytes), + )); + } + Ok(result) + }) } // ==================== Delegate Index Methods ==================== @@ -672,20 +723,21 @@ impl ReDb { key_bytes[..32].copy_from_slice(delegate_key.as_ref()); key_bytes[32..].copy_from_slice(delegate_key.code_hash().as_ref()); - let txn = self.begin_read()?; - let tbl = txn.open_table(DELEGATE_INDEX_TABLE)?; - match tbl.get(key_bytes.as_slice())? { - Some(v) => { - let bytes: [u8; 32] = v.value().try_into().map_err(|_| { - redb::Error::Io(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "Invalid CodeHash length", - )) - })?; - Ok(Some(CodeHash::from(&bytes))) + self.read_guarded(|txn| { + let tbl = txn.open_table(DELEGATE_INDEX_TABLE)?; + match tbl.get(key_bytes.as_slice())? { + Some(v) => { + let bytes: [u8; 32] = v.value().try_into().map_err(|_| { + redb::Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Invalid CodeHash length", + )) + })?; + Ok(Some(CodeHash::from(&bytes))) + } + None => Ok(None), } - None => Ok(None), - } + }) } /// Remove a delegate index entry @@ -704,30 +756,31 @@ impl ReDb { /// Load all delegate index entries pub fn load_all_delegate_index(&self) -> Result, redb::Error> { - let txn = self.begin_read()?; - let tbl = txn.open_table(DELEGATE_INDEX_TABLE)?; + self.read_guarded(|txn| { + let tbl = txn.open_table(DELEGATE_INDEX_TABLE)?; + + let mut result = Vec::new(); + for entry in tbl.iter()? { + let (key, value) = entry?; + let key_bytes = key.value(); + if key_bytes.len() != 64 { + continue; // Skip malformed entries + } + let delegate_key_bytes: [u8; 32] = key_bytes[..32].try_into().unwrap(); + let code_hash_bytes: [u8; 32] = key_bytes[32..].try_into().unwrap(); + let value_bytes: [u8; 32] = value.value().try_into().map_err(|_| { + redb::Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Invalid CodeHash length", + )) + })?; - let mut result = Vec::new(); - for entry in tbl.iter()? { - let (key, value) = entry?; - let key_bytes = key.value(); - if key_bytes.len() != 64 { - continue; // Skip malformed entries + let delegate_key = + DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); + result.push((delegate_key, CodeHash::from(&value_bytes))); } - let delegate_key_bytes: [u8; 32] = key_bytes[..32].try_into().unwrap(); - let code_hash_bytes: [u8; 32] = key_bytes[32..].try_into().unwrap(); - let value_bytes: [u8; 32] = value.value().try_into().map_err(|_| { - redb::Error::Io(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "Invalid CodeHash length", - )) - })?; - - let delegate_key = - DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); - result.push((delegate_key, CodeHash::from(&value_bytes))); - } - Ok(result) + Ok(result) + }) } // ==================== Secrets Index Methods ==================== @@ -766,26 +819,27 @@ impl ReDb { key_bytes[..32].copy_from_slice(delegate_key.as_ref()); key_bytes[32..].copy_from_slice(delegate_key.code_hash().as_ref()); - let txn = self.begin_read()?; - let tbl = txn.open_table(SECRETS_INDEX_TABLE)?; - match tbl.get(key_bytes.as_slice())? { - Some(v) => { - let value = v.value(); - if value.len() % 32 != 0 { - return Err(redb::Error::Io(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "Invalid secrets index value length", - ))); - } - let mut result = Vec::with_capacity(value.len() / 32); - for chunk in value.chunks(32) { - let arr: [u8; 32] = chunk.try_into().unwrap(); - result.push(arr); + self.read_guarded(|txn| { + let tbl = txn.open_table(SECRETS_INDEX_TABLE)?; + match tbl.get(key_bytes.as_slice())? { + Some(v) => { + let value = v.value(); + if value.len() % 32 != 0 { + return Err(redb::Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Invalid secrets index value length", + ))); + } + let mut result = Vec::with_capacity(value.len() / 32); + for chunk in value.chunks(32) { + let arr: [u8; 32] = chunk.try_into().unwrap(); + result.push(arr); + } + Ok(Some(result)) } - Ok(Some(result)) + None => Ok(None), } - None => Ok(None), - } + }) } /// Remove a secrets index entry @@ -805,34 +859,35 @@ impl ReDb { /// Load all secrets index entries #[allow(clippy::type_complexity)] pub fn load_all_secrets_index(&self) -> Result)>, redb::Error> { - let txn = self.begin_read()?; - let tbl = txn.open_table(SECRETS_INDEX_TABLE)?; + self.read_guarded(|txn| { + let tbl = txn.open_table(SECRETS_INDEX_TABLE)?; + + let mut result = Vec::new(); + for entry in tbl.iter()? { + let (key, value) = entry?; + let key_bytes = key.value(); + if key_bytes.len() != 64 { + continue; // Skip malformed entries + } + let delegate_key_bytes: [u8; 32] = key_bytes[..32].try_into().unwrap(); + let code_hash_bytes: [u8; 32] = key_bytes[32..].try_into().unwrap(); - let mut result = Vec::new(); - for entry in tbl.iter()? { - let (key, value) = entry?; - let key_bytes = key.value(); - if key_bytes.len() != 64 { - continue; // Skip malformed entries - } - let delegate_key_bytes: [u8; 32] = key_bytes[..32].try_into().unwrap(); - let code_hash_bytes: [u8; 32] = key_bytes[32..].try_into().unwrap(); + let value_bytes = value.value(); + if value_bytes.len() % 32 != 0 { + continue; // Skip malformed entries + } + let mut secret_keys = Vec::with_capacity(value_bytes.len() / 32); + for chunk in value_bytes.chunks(32) { + let arr: [u8; 32] = chunk.try_into().unwrap(); + secret_keys.push(arr); + } - let value_bytes = value.value(); - if value_bytes.len() % 32 != 0 { - continue; // Skip malformed entries + let delegate_key = + DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); + result.push((delegate_key, secret_keys)); } - let mut secret_keys = Vec::with_capacity(value_bytes.len() / 32); - for chunk in value_bytes.chunks(32) { - let arr: [u8; 32] = chunk.try_into().unwrap(); - secret_keys.push(arr); - } - - let delegate_key = - DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); - result.push((delegate_key, secret_keys)); - } - Ok(result) + Ok(result) + }) } // ============== Per-User Secrets Index Methods (P1 of #4381) ============== @@ -885,26 +940,27 @@ impl ReDb { ) -> Result>, redb::Error> { let key_bytes = Self::user_index_key(delegate_key, user_id); - let txn = self.begin_read()?; - let tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; - match tbl.get(key_bytes.as_slice())? { - Some(v) => { - let value = v.value(); - if value.len() % 32 != 0 { - return Err(redb::Error::Io(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "Invalid user secrets index value length", - ))); - } - let mut result = Vec::with_capacity(value.len() / 32); - for chunk in value.chunks(32) { - let arr: [u8; 32] = chunk.try_into().unwrap(); - result.push(arr); + self.read_guarded(|txn| { + let tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; + match tbl.get(key_bytes.as_slice())? { + Some(v) => { + let value = v.value(); + if value.len() % 32 != 0 { + return Err(redb::Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Invalid user secrets index value length", + ))); + } + let mut result = Vec::with_capacity(value.len() / 32); + for chunk in value.chunks(32) { + let arr: [u8; 32] = chunk.try_into().unwrap(); + result.push(arr); + } + Ok(Some(result)) } - Ok(Some(result)) + None => Ok(None), } - None => Ok(None), - } + }) } /// Remove a per-user secrets index entry. Called by the inactive-user TTL @@ -930,7 +986,7 @@ impl ReDb { pub fn load_all_user_secrets_index( &self, ) -> Result)>, redb::Error> { - let txn = self.begin_read()?; + self.read_guarded(|txn| { let tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; let mut result = Vec::new(); @@ -977,6 +1033,7 @@ impl ReDb { result.push(((delegate_key, user_id_bytes), secret_keys)); } Ok(result) + }) } // ==================== Broken Invariants Methods ==================== @@ -1022,33 +1079,34 @@ impl ReDb { /// than failing the entire load — a corrupted entry should not block /// startup, and the worst case is we lose a flag and re-detect it. pub fn load_all_broken_invariants(&self) -> Result, redb::Error> { - let txn = self.begin_read()?; - let tbl = txn.open_table(BROKEN_INVARIANTS_TABLE)?; - - let mut result = Vec::new(); - for entry in tbl.iter()? { - let (key, value) = entry?; - let key_bytes: [u8; 32] = match key.value().try_into() { - Ok(b) => b, - Err(_) => { + self.read_guarded(|txn| { + let tbl = txn.open_table(BROKEN_INVARIANTS_TABLE)?; + + let mut result = Vec::new(); + for entry in tbl.iter()? { + let (key, value) = entry?; + let key_bytes: [u8; 32] = match key.value().try_into() { + Ok(b) => b, + Err(_) => { + tracing::warn!( + len = key.value().len(), + "Skipping malformed broken-invariants row (key length)" + ); + continue; + } + }; + let v = value.value(); + if v.len() != 1 { tracing::warn!( - len = key.value().len(), - "Skipping malformed broken-invariants row (key length)" + len = v.len(), + "Skipping malformed broken-invariants row (value length)" ); continue; } - }; - let v = value.value(); - if v.len() != 1 { - tracing::warn!( - len = v.len(), - "Skipping malformed broken-invariants row (value length)" - ); - continue; + result.push((ContractInstanceId::new(key_bytes), v[0])); } - result.push((ContractInstanceId::new(key_bytes), v[0])); - } - Ok(result) + Ok(result) + }) } } @@ -1091,17 +1149,12 @@ impl StateStorage for ReDb { } async fn get(&self, key: &ContractKey) -> Result, Self::Error> { - let txn = self.begin_read()?; - - let val = { + self.read_guarded(|txn| { let tbl = txn.open_table(STATE_TABLE)?; - tbl.get(key.as_bytes())? - }; - - match val { - Some(v) => Ok(Some(WrappedState::new(v.value().to_vec()))), - None => Ok(None), - } + Ok(tbl + .get(key.as_bytes())? + .map(|v| WrappedState::new(v.value().to_vec()))) + }) } async fn store_params( @@ -1122,17 +1175,12 @@ impl StateStorage for ReDb { &'a self, key: &'a ContractKey, ) -> Result>, Self::Error> { - let txn = self.begin_read()?; - - let val = { + self.read_guarded(|txn| { let tbl = txn.open_table(CONTRACT_PARAMS_TABLE)?; - tbl.get(key.as_bytes())? - }; - - match val { - Some(v) => Ok(Some(Parameters::from(v.value().to_vec()))), - None => Ok(None), - } + Ok(tbl + .get(key.as_bytes())? + .map(|v| Parameters::from(v.value().to_vec()))) + }) } async fn remove(&self, key: &ContractKey) -> Result<(), Self::Error> { @@ -1716,6 +1764,26 @@ mod tests { transaction_error_is_poison(&begin_err), "PreviousIo from a poisoned database's begin_write must classify as poison" ); + + // The umbrella read-path classifier must also flag the real PreviousIo... + let umbrella: redb::Error = begin_err.into(); + assert!( + redb_error_is_poison(&umbrella), + "PreviousIo must classify as poison on the umbrella read path too" + ); + + // ...but must NOT flag the synthetic `Io(InvalidData)` that several read + // methods produce for a benign malformed-data row. Misclassifying it would + // exit-and-restart the node on a single bad row (a crash loop) — the + // false-positive this asymmetry exists to prevent (#4604). + let malformed_row = redb::Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Invalid CodeHash length", + )); + assert!( + !redb_error_is_poison(&malformed_row), + "a synthetic Io(InvalidData) malformed-row error must NOT be treated as poison" + ); } /// End-to-end (issue #4604, requirement 3): a poisoned database routes contract From 76eab993acee0ccc7133102b519f9f1db452fad9 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 27 Jun 2026 19:40:21 -0500 Subject: [PATCH 3/4] fix: route commit-time redb poison to recovery on the same op (#4604 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex's second pass: the FIRST backend I/O failure that poisons redb usually surfaces from commit() as CommitError::Storage(Io), which the begin_* wrappers don't see — so the node stayed running on the poisoning write until a later op hit begin_write and tripped PreviousIo. Add `commit_guarded`, used by every write method's commit, which routes a poison commit error to the recovery path on the very op that poisons the handle. The commit path comes straight from redb (never the synthetic Io(InvalidData) of a malformed row), so it matches `Io` via `storage_error_is_poison`. The end-to-end test now asserts the poisoning write triggers recovery on the same op. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_014dGjU1Q6Vpk2dm4sUf4pdU --- crates/core/src/contract/storages/redb.rs | 76 ++++++++++++++++------- 1 file changed, 54 insertions(+), 22 deletions(-) diff --git a/crates/core/src/contract/storages/redb.rs b/crates/core/src/contract/storages/redb.rs index e136dabae6..b04ec0d77e 100644 --- a/crates/core/src/contract/storages/redb.rs +++ b/crates/core/src/contract/storages/redb.rs @@ -263,6 +263,30 @@ impl ReDb { f(&txn).map_err(Self::route_redb_error) } + /// Commit a write transaction, routing a poison error to the #4604 recovery + /// path. The FIRST backend I/O failure usually surfaces HERE (redb reports it as + /// `CommitError::Storage(StorageError::Io)`), on the very op that poisons the + /// handle — so catching it at commit triggers the restart immediately instead of + /// waiting for the next `begin_write` to trip `PreviousIo`. Unlike the read path, + /// commit/begin errors come straight from redb and are never the synthetic + /// `Io(InvalidData)` of a malformed row, so it is safe to match `Io` here via + /// [`storage_error_is_poison`]. + fn commit_guarded(txn: WriteTransaction) -> Result<(), redb::Error> { + match txn.commit() { + Ok(()) => Ok(()), + Err(e) => { + if let redb::CommitError::Storage(s) = &e { + if storage_error_is_poison(s) { + #[cfg(test)] + POISON_RECOVERY_TRIGGERED.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + crate::node::abort_process_on_redb_poison(&e.to_string()); + } + } + Err(e.into()) + } + } + } + pub async fn new(data_dir: &Path) -> Result { let db_path = data_dir.join("db"); tracing::info!( @@ -477,7 +501,7 @@ impl ReDb { let mut tbl = txn.open_table(HOSTING_METADATA_TABLE)?; tbl.insert(key.as_bytes(), metadata.to_bytes().as_slice())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Get hosting metadata for a contract. @@ -501,7 +525,7 @@ impl ReDb { let mut tbl = txn.open_table(HOSTING_METADATA_TABLE)?; tbl.remove(key.as_bytes())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all hosting metadata from the database. @@ -551,7 +575,7 @@ impl ReDb { let mut tbl = txn.open_table(STATE_TABLE)?; tbl.insert(key.as_bytes(), state.as_ref())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Atomically update a contract's state, failing if no prior state exists. @@ -576,7 +600,7 @@ impl ReDb { } tbl.insert(key.as_bytes(), state.as_ref())?; } - txn.commit()?; + Self::commit_guarded(txn)?; Ok(true) } @@ -623,7 +647,7 @@ impl ReDb { let mut tbl = txn.open_table(CONTRACT_INDEX_TABLE)?; tbl.insert(instance_id.as_ref(), code_hash.as_ref())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Get the CodeHash for a ContractInstanceId @@ -658,7 +682,7 @@ impl ReDb { let mut tbl = txn.open_table(CONTRACT_INDEX_TABLE)?; tbl.remove(instance_id.as_ref())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all contract index entries @@ -711,7 +735,7 @@ impl ReDb { let mut tbl = txn.open_table(DELEGATE_INDEX_TABLE)?; tbl.insert(key_bytes.as_slice(), code_hash.as_ref())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Get the CodeHash for a DelegateKey @@ -751,7 +775,7 @@ impl ReDb { let mut tbl = txn.open_table(DELEGATE_INDEX_TABLE)?; tbl.remove(key_bytes.as_slice())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all delegate index entries @@ -807,7 +831,7 @@ impl ReDb { let mut tbl = txn.open_table(SECRETS_INDEX_TABLE)?; tbl.insert(key_bytes.as_slice(), value_bytes.as_slice())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Get the secret key hashes for a DelegateKey @@ -853,7 +877,7 @@ impl ReDb { let mut tbl = txn.open_table(SECRETS_INDEX_TABLE)?; tbl.remove(key_bytes.as_slice())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all secrets index entries @@ -924,7 +948,7 @@ impl ReDb { let mut tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; tbl.insert(key_bytes.as_slice(), value_bytes.as_slice())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Get the secret key hashes for a `(DelegateKey, UserId)` pair. @@ -977,7 +1001,7 @@ impl ReDb { let mut tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; tbl.remove(key_bytes.as_slice())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all per-user secrets index entries as @@ -1055,7 +1079,7 @@ impl ReDb { let mut tbl = txn.open_table(BROKEN_INVARIANTS_TABLE)?; tbl.insert(instance_id.as_ref(), &[kind_byte][..])?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Remove a persisted broken-invariant flag. Paired with @@ -1071,7 +1095,7 @@ impl ReDb { let mut tbl = txn.open_table(BROKEN_INVARIANTS_TABLE)?; tbl.remove(instance_id.as_ref())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all persisted broken-invariant flags. Malformed rows (wrong @@ -1145,7 +1169,7 @@ impl StateStorage for ReDb { tbl.insert(key.as_bytes(), metadata.to_bytes().as_slice())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } async fn get(&self, key: &ContractKey) -> Result, Self::Error> { @@ -1168,7 +1192,7 @@ impl StateStorage for ReDb { let mut tbl = txn.open_table(CONTRACT_PARAMS_TABLE)?; tbl.insert(key.as_bytes(), params.as_ref())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } async fn get_params<'a>( @@ -1200,7 +1224,7 @@ impl StateStorage for ReDb { let mut tbl = txn.open_table(HOSTING_METADATA_TABLE)?; tbl.remove(key.as_bytes())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } } @@ -1811,20 +1835,28 @@ mod tests { "benign not-found / normal ops must NOT take the poison-recovery path" ); - // Poison the backend; the next write fails (I/O), and redb latches its - // in-memory poison flag (io_failed) — set on ANY backend read/write error. + // Poison the backend; the write that triggers the FIRST backend I/O error + // (usually at commit) must take the recovery path on the very op that + // poisons the handle — not wait for a later op. redb also latches its + // in-memory poison flag (io_failed) here, set on ANY backend read/write error. backend.start_failing(); + POISON_RECOVERY_TRIGGERED.store(0, Ordering::SeqCst); assert!( db.store_state_sync(&key, WrappedState::new(vec![4, 5, 6])) .is_err(), "the injected I/O failure must surface as an error" ); + assert!( + POISON_RECOVERY_TRIGGERED.load(Ordering::SeqCst) >= 1, + "the poisoning write (commit-time I/O error) must take the recovery path \ + on the same op, not only on a later one" + ); - // The database is now poisoned. redb's poison flag is checked by every + // The database stays poisoned: redb's poison flag is checked by every // `begin_write` (the node writes hosting metadata on essentially every // contract op), so the next write returns PreviousIo, which the begin_write - // wrapper routes to the recovery (exit-for-restart) path instead of the old - // fail-forever behaviour. + // wrapper also routes to the recovery (exit-for-restart) path instead of the + // old fail-forever behaviour. POISON_RECOVERY_TRIGGERED.store(0, Ordering::SeqCst); assert!( db.store_state_sync(&key, WrappedState::new(vec![7, 8, 9])) From 9029c0c06ab2786644522917257292d4d012182d Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 27 Jun 2026 20:00:14 -0500 Subject: [PATCH 4/4] style: reindent read_guarded closure body + add read-path poison test (#4604) Addresses the rule-review findings on the PR: - load_all_user_secrets_index: reindent the read_guarded closure body to match the other converted methods (rustfmt had silently skipped reformatting this longer fn, so cargo fmt --check passed despite the inconsistent indentation). - Add a deterministic read-path assertion: feed read_guarded a real PreviousIo (from the poisoned handle) and confirm it routes to the recovery path, covering the read_guarded -> route_redb_error wiring in the end-to-end test (previously only the begin_write/commit write path was observed via the counter). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_014dGjU1Q6Vpk2dm4sUf4pdU --- crates/core/src/contract/storages/redb.rs | 112 +++++++++++++--------- 1 file changed, 67 insertions(+), 45 deletions(-) diff --git a/crates/core/src/contract/storages/redb.rs b/crates/core/src/contract/storages/redb.rs index b04ec0d77e..aae6c4a280 100644 --- a/crates/core/src/contract/storages/redb.rs +++ b/crates/core/src/contract/storages/redb.rs @@ -1011,52 +1011,52 @@ impl ReDb { &self, ) -> Result)>, redb::Error> { self.read_guarded(|txn| { - let tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; - - let mut result = Vec::new(); - for entry in tbl.iter()? { - let (key, value) = entry?; - let key_bytes = key.value(); - if key_bytes.len() != 96 { - // Skip malformed entries. The write path always emits a - // 96-byte composite key (DelegateKey(64) || UserId(32)); a - // wrong length means an externally-corrupted or - // future-format row. Drop it rather than panic on the - // fixed-size `try_into`s below, and warn so the corruption - // is visible in monitoring. - tracing::warn!( - len = key_bytes.len(), - "Skipping malformed user-secrets-index row (key length != 96)" - ); - continue; - } - let delegate_key_bytes: [u8; 32] = key_bytes[..32].try_into().unwrap(); - let code_hash_bytes: [u8; 32] = key_bytes[32..64].try_into().unwrap(); - let user_id_bytes: [u8; 32] = key_bytes[64..].try_into().unwrap(); - - let value_bytes = value.value(); - if value_bytes.len() % 32 != 0 { - // Skip malformed entries. The value is a concatenation of - // 32-byte secret-key hashes, so a length not divisible by - // 32 is corruption. Warn and drop rather than splitting a - // partial hash. - tracing::warn!( - len = value_bytes.len(), - "Skipping malformed user-secrets-index row (value length not a multiple of 32)" - ); - continue; - } - let mut secret_keys = Vec::with_capacity(value_bytes.len() / 32); - for chunk in value_bytes.chunks(32) { - let arr: [u8; 32] = chunk.try_into().unwrap(); - secret_keys.push(arr); - } + let tbl = txn.open_table(USER_SECRETS_INDEX_TABLE)?; - let delegate_key = - DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); - result.push(((delegate_key, user_id_bytes), secret_keys)); - } - Ok(result) + let mut result = Vec::new(); + for entry in tbl.iter()? { + let (key, value) = entry?; + let key_bytes = key.value(); + if key_bytes.len() != 96 { + // Skip malformed entries. The write path always emits a + // 96-byte composite key (DelegateKey(64) || UserId(32)); a + // wrong length means an externally-corrupted or + // future-format row. Drop it rather than panic on the + // fixed-size `try_into`s below, and warn so the corruption + // is visible in monitoring. + tracing::warn!( + len = key_bytes.len(), + "Skipping malformed user-secrets-index row (key length != 96)" + ); + continue; + } + let delegate_key_bytes: [u8; 32] = key_bytes[..32].try_into().unwrap(); + let code_hash_bytes: [u8; 32] = key_bytes[32..64].try_into().unwrap(); + let user_id_bytes: [u8; 32] = key_bytes[64..].try_into().unwrap(); + + let value_bytes = value.value(); + if value_bytes.len() % 32 != 0 { + // Skip malformed entries. The value is a concatenation of + // 32-byte secret-key hashes, so a length not divisible by + // 32 is corruption. Warn and drop rather than splitting a + // partial hash. + tracing::warn!( + len = value_bytes.len(), + "Skipping malformed user-secrets-index row (value length not a multiple of 32)" + ); + continue; + } + let mut secret_keys = Vec::with_capacity(value_bytes.len() / 32); + for chunk in value_bytes.chunks(32) { + let arr: [u8; 32] = chunk.try_into().unwrap(); + secret_keys.push(arr); + } + + let delegate_key = + DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); + result.push(((delegate_key, user_id_bytes), secret_keys)); + } + Ok(result) }) } @@ -1868,5 +1868,27 @@ mod tests { "a poisoned database write must take the recovery (exit-for-restart) path \ rather than failing forever" ); + + // Read path: redb's `begin_read` does not check the poison flag, so a poison + // surfaces inside the read body (at open_table/get/iter) as a `PreviousIo`. + // Feed `read_guarded` a real `PreviousIo` (obtained from the now-poisoned + // handle's begin_write) to prove the read path routes to recovery too, not + // just the write path. (A poisoned read served from cache would succeed and + // is not a failure; this exercises the failing-read body deterministically.) + let previous_io: redb::Error = match db.begin_write() { + Ok(_) => panic!("database should still be poisoned"), + Err(e) => e.into(), + }; + assert!( + redb_error_is_poison(&previous_io), + "a poisoned handle's PreviousIo must classify as poison on the read path" + ); + POISON_RECOVERY_TRIGGERED.store(0, Ordering::SeqCst); + let routed: Result<(), redb::Error> = db.read_guarded(|_txn| Err(previous_io)); + assert!(routed.is_err()); + assert!( + POISON_RECOVERY_TRIGGERED.load(Ordering::SeqCst) >= 1, + "read_guarded must route a poison surfacing inside the read body to recovery" + ); } }