Journaling fix (#5892) (#5935)

Port of #5892.

## Motivation

#5891 added logs but now we would like to prevent a possible DB
corruption happening with journaling.

## Proposal

* Distinguish DB errors that may force a reload of the view
* This is the case for certain journaling errors during save().

## Test Plan

CI

## Release Plan

- Nothing to do.

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: afck

linera-core/src/chain_worker/state.rs

@@ -110,6 +110,8 @@ where
     chain_modes: Option<Arc<sync::RwLock<BTreeMap<ChainId, ListeningMode>>>>,
     delivery_notifier: DeliveryNotifier,
     knows_chain_is_active: bool,
+    /// Set to `true` if a journal resolution failure has left storage potentially inconsistent.
+    poisoned: bool,
 }
 
 /// Whether the block was processed or skipped. Used for metrics.
@@ -170,6 +172,7 @@ where
             chain_modes,
             delivery_notifier,
             knows_chain_is_active: false,
+            poisoned: false,
         })
     }
 
@@ -189,8 +192,16 @@ where
     }
 
     /// Rolls back any uncommitted changes to the chain state.
+    /// Does nothing if the worker is poisoned (the view is in an inconsistent state).
     pub(crate) fn rollback(&mut self) {
-        self.chain.rollback();
+        if !self.poisoned {
+            self.chain.rollback();
+        }
+    }
+
+    /// Returns `true` if the worker is poisoned due to a journal resolution failure.
+    pub(crate) fn is_poisoned(&self) -> bool {
+        self.poisoned
     }
 
     /// Updates the last-access timestamp to the current time.
@@ -2130,11 +2141,23 @@ where
     }
 
     /// Stores the chain state in persistent storage.
+    ///
+    /// If the save fails, the worker is marked as poisoned and must be reloaded.
     #[instrument(skip_all, fields(
         chain_id = %self.chain_id()
     ))]
     async fn save(&mut self) -> Result<(), WorkerError> {
-        self.chain.save().await?;
+        if let Err(e) = self.chain.save().await {
+            if e.must_reload_view() {
+                tracing::error!(
+                    error = ?e,
+                    chain_id = %self.chain_id(),
+                    "Journal resolution failed; marking worker as poisoned"
+                );
+                self.poisoned = true;
+            }
+            return Err(e.into());
+        }
         Ok(())
     }
 }

linera-core/src/worker.rs

@@ -451,6 +451,12 @@ impl WorkerError {
             WorkerError::ChainError(chain_error) => chain_error.is_local(),
         }
     }
+
+    /// Returns `true` if this error was caused by a journal resolution failure,
+    /// which may leave storage in an inconsistent state requiring a view reload.
+    pub fn must_reload_view(&self) -> bool {
+        matches!(self, WorkerError::ViewError(e) if e.must_reload_view())
+    }
 }
 
 impl From<ChainError> for WorkerError {
@@ -767,11 +773,20 @@ where
         Fut: std::future::Future<Output = Result<R, WorkerError>>,
     {
         let state = self.get_or_create_chain_worker(chain_id).await?;
-        Box::pin(wrap_future(async move {
+        let result = Box::pin(wrap_future(async move {
             let guard = handle::read_lock(&state).await;
+            if guard.is_poisoned() {
+                return Err(Self::poisoned_worker_error());
+            }
             f(guard).await
         }))
-        .await
+        .await;
+        if let Err(e) = &result {
+            if e.must_reload_view() {
+                self.evict_poisoned_worker(chain_id);
+            }
+        }
+        result
     }
 
     /// Acquires a write lock on the chain worker and executes the given closure.
@@ -785,11 +800,35 @@ where
         Fut: std::future::Future<Output = Result<R, WorkerError>>,
     {
         let state = self.get_or_create_chain_worker(chain_id).await?;
-        Box::pin(wrap_future(async move {
+        let result = Box::pin(wrap_future(async move {
             let guard = handle::write_lock(&state).await;
+            if guard.is_poisoned() {
+                return Err(Self::poisoned_worker_error());
+            }
             f(guard).await
         }))
-        .await
+        .await;
+        if let Err(e) = &result {
+            if e.must_reload_view() {
+                self.evict_poisoned_worker(chain_id);
+            }
+        }
+        result
+    }
+
+    fn poisoned_worker_error() -> WorkerError {
+        WorkerError::ViewError(ViewError::StoreError {
+            backend: "journaling",
+            error: "Worker is poisoned due to a journal resolution failure".into(),
+            must_reload_view: true,
+        })
+    }
+
+    /// Evicts a poisoned chain worker from the cache so it gets reloaded on the next request.
+    fn evict_poisoned_worker(&self, chain_id: ChainId) {
+        tracing::warn!(%chain_id, "Evicting poisoned chain worker from cache");
+        let pin = self.chain_workers.pin();
+        pin.remove(&chain_id);
     }
 
     /// Gets or creates a chain worker for the given chain.

linera-views/src/backends/dual.rs

@@ -413,4 +413,12 @@ where
     E2: KeyValueStoreError,
 {
     const BACKEND: &'static str = "dual_store";
+
+    fn must_reload_view(&self) -> bool {
+        match self {
+            DualStoreError::First(e) => e.must_reload_view(),
+            DualStoreError::Second(e) => e.must_reload_view(),
+            _ => false,
+        }
+    }
 }

linera-views/src/backends/dynamo_db.rs

@@ -44,7 +44,7 @@ use crate::store::TestKeyValueDatabase;
 use crate::{
     batch::SimpleUnorderedBatch,
     common::get_uleb128_size,
-    journaling::{JournalConsistencyError, JournalingKeyValueDatabase},
+    journaling::JournalingKeyValueDatabase,
     lru_caching::{LruCachingConfig, LruCachingDatabase},
     store::{
         DirectWritableKeyValueStore, KeyValueDatabase, KeyValueStoreError, ReadableKeyValueStore,
@@ -994,10 +994,6 @@ pub enum DynamoDbStoreInternalError {
     #[error("The key prefix must have at most 1024 bytes")]
     KeyPrefixTooLong,
 
-    /// The journal is not coherent
-    #[error(transparent)]
-    JournalConsistencyError(#[from] JournalConsistencyError),
-
     /// The length of the value should be at most 400 KB.
     #[error("The DynamoDB value should be less than 400 KB")]
     ValueLengthTooLarge,
@@ -1093,7 +1089,7 @@ impl KeyValueStoreError for DynamoDbStoreInternalError {
 
 #[cfg(with_testing)]
 impl TestKeyValueDatabase for JournalingKeyValueDatabase<DynamoDbDatabaseInternal> {
-    async fn new_test_config() -> Result<DynamoDbStoreInternalConfig, DynamoDbStoreInternalError> {
+    async fn new_test_config() -> Result<DynamoDbStoreInternalConfig, Self::Error> {
         Ok(DynamoDbStoreInternalConfig {
             use_dynamodb_local: true,
             max_concurrent_queries: Some(TEST_DYNAMO_DB_MAX_CONCURRENT_QUERIES),
@@ -1103,7 +1099,8 @@ impl TestKeyValueDatabase for JournalingKeyValueDatabase<DynamoDbDatabaseInterna
 }
 
 /// The combined error type for [`DynamoDbDatabase`].
-pub type DynamoDbStoreError = ValueSplittingError<DynamoDbStoreInternalError>;
+pub type DynamoDbStoreError =
+    ValueSplittingError<crate::journaling::JournalingError<DynamoDbStoreInternalError>>;
 
 /// The config type for [`DynamoDbDatabase`]`
 pub type DynamoDbStoreConfig = LruCachingConfig<DynamoDbStoreInternalConfig>;