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..aae6c4a280 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,74 @@ 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)) +} + +/// 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 +/// 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 +199,94 @@ 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 + } + + /// 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) + } + + /// 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!( @@ -337,12 +496,12 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Get hosting metadata for a contract. @@ -350,22 +509,23 @@ impl ReDb { &self, key: &ContractKey, ) -> Result, redb::Error> { - let txn = self.0.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. 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all hosting metadata from the database. @@ -374,27 +534,26 @@ impl ReDb { pub fn load_all_hosting_metadata( &self, ) -> Result, HostingMetadata)>, redb::Error> { - let txn = self.0.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.0.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. @@ -411,12 +570,12 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Atomically update a contract's state, failing if no prior state exists. @@ -431,7 +590,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 @@ -441,7 +600,7 @@ impl ReDb { } tbl.insert(key.as_bytes(), state.as_ref())?; } - txn.commit()?; + Self::commit_guarded(txn)?; Ok(true) } @@ -451,29 +610,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.0.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.0.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 ==================== @@ -485,12 +642,12 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Get the CodeHash for a ContractInstanceId @@ -498,20 +655,21 @@ impl ReDb { &self, instance_id: &ContractInstanceId, ) -> Result, redb::Error> { - let txn = self.0.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 @@ -519,42 +677,43 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all contract index entries pub fn load_all_contract_index( &self, ) -> Result, redb::Error> { - let txn = self.0.begin_read()?; - 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) + 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) + }) } // ==================== Delegate Index Methods ==================== @@ -571,12 +730,12 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Get the CodeHash for a DelegateKey @@ -588,20 +747,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.0.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 @@ -610,40 +770,41 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all delegate index entries pub fn load_all_delegate_index(&self) -> Result, redb::Error> { - let txn = self.0.begin_read()?; - 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", - )) - })?; + 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 delegate_key = - DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); - result.push((delegate_key, CodeHash::from(&value_bytes))); - } - Ok(result) + let delegate_key = + DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); + result.push((delegate_key, CodeHash::from(&value_bytes))); + } + Ok(result) + }) } // ==================== Secrets Index Methods ==================== @@ -665,12 +826,12 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Get the secret key hashes for a DelegateKey @@ -682,26 +843,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.0.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", - ))); + 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)) } - 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)) + None => Ok(None), } - None => Ok(None), - } + }) } /// Remove a secrets index entry @@ -710,45 +872,46 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// 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 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(); + 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 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 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) + let delegate_key = + DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); + result.push((delegate_key, secret_keys)); + } + Ok(result) + }) } // ============== Per-User Secrets Index Methods (P1 of #4381) ============== @@ -780,12 +943,12 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Get the secret key hashes for a `(DelegateKey, UserId)` pair. @@ -801,26 +964,27 @@ impl ReDb { ) -> Result>, redb::Error> { let key_bytes = Self::user_index_key(delegate_key, user_id); - let txn = self.0.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 @@ -832,12 +996,12 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all per-user secrets index entries as @@ -846,53 +1010,54 @@ impl ReDb { pub fn load_all_user_secrets_index( &self, ) -> Result)>, redb::Error> { - let txn = self.0.begin_read()?; - 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); - } + 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 delegate_key = - DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); - result.push(((delegate_key, user_id_bytes), secret_keys)); - } - Ok(result) + let delegate_key = + DelegateKey::new(delegate_key_bytes, CodeHash::from(&code_hash_bytes)); + result.push(((delegate_key, user_id_bytes), secret_keys)); + } + Ok(result) + }) } // ==================== Broken Invariants Methods ==================== @@ -909,12 +1074,12 @@ 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][..])?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Remove a persisted broken-invariant flag. Paired with @@ -925,12 +1090,12 @@ 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())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } /// Load all persisted broken-invariant flags. Malformed rows (wrong @@ -938,33 +1103,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.0.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) + }) } } @@ -973,7 +1139,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)?; @@ -1003,21 +1169,16 @@ 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> { - let txn = self.0.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( @@ -1025,37 +1186,32 @@ 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)?; tbl.insert(key.as_bytes(), params.as_ref())?; } - txn.commit().map_err(Into::into) + Self::commit_guarded(txn) } async fn get_params<'a>( &'a self, key: &'a ContractKey, ) -> Result>, Self::Error> { - let txn = self.0.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> { // 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())?; @@ -1068,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) } } @@ -1493,4 +1649,246 @@ 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" + ); + + // 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 + /// 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 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 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 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])) + .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" + ); + + // 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" + ); + } } 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() {