Cleanups; fix potential race condition (#5936)

## Motivation

In #5935 a few
questions were raised after it was already merged. Whether we
`rollback()` is irrelevant if the state is poisoned because we drop that
view anyway. Also, the port needlessly introduced a few changes compared
to `testnet_conway`.

## Proposal

Bring the code more in line with `testnet_conway`.

Make sure worker states only get evicted if they are really poisoned.

## Test Plan

CI

## Release Plan

- Backport the worker fix.

## Links

- Addresses some comments from
#5935.
- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

---------

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

Author: afck

linera-core/src/chain_worker/state.rs

@@ -114,13 +114,6 @@ where
     poisoned: bool,
 }
 
-/// Whether the block was processed or skipped. Used for metrics.
-pub enum BlockOutcome {
-    Processed,
-    Preprocessed,
-    Skipped,
-}
-
 /// The result of processing a cross-chain update.
 pub(crate) enum CrossChainUpdateResult {
     /// The update was applied and the chain was saved up to the given height.
@@ -136,6 +129,13 @@ pub(crate) enum CrossChainUpdateResult {
     },
 }
 
+/// Whether the block was processed or skipped. Used for metrics.
+pub enum BlockOutcome {
+    Processed,
+    Preprocessed,
+    Skipped,
+}
+
 impl<StorageClient> ChainWorkerState<StorageClient>
 where
     StorageClient: Storage + Clone + 'static,
@@ -192,16 +192,15 @@ 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) {
-        if !self.poisoned {
-            self.chain.rollback();
-        }
+        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
+    /// Returns `WorkerError::PoisonedWorker` if the worker is poisoned due to a journal
+    /// resolution failure.
+    pub(crate) fn check_not_poisoned(&self) -> Result<(), WorkerError> {
+        ensure!(!self.poisoned, WorkerError::PoisonedWorker);
+        Ok(())
     }
 
     /// Updates the last-access timestamp to the current time.
@@ -345,7 +344,6 @@ where
         Ok(maybe_blobs.into_iter().collect())
     }
 
-    /// Loads pending cross-chain requests, and adds `NewRound` notifications where appropriate.
     /// Creates cross-chain requests for a single recipient from its outbox.
     #[instrument(skip_all, fields(
         chain_id = %self.chain_id()
@@ -2147,16 +2145,16 @@ where
         chain_id = %self.chain_id()
     ))]
     async fn save(&mut self) -> Result<(), WorkerError> {
-        if let Err(e) = self.chain.save().await {
-            if e.must_reload_view() {
+        if let Err(error) = self.chain.save().await {
+            if error.must_reload_view() {
                 tracing::error!(
-                    error = ?e,
+                    ?error,
                     chain_id = %self.chain_id(),
                     "Journal resolution failed; marking worker as poisoned"
                 );
                 self.poisoned = true;
             }
-            return Err(e.into());
+            return Err(error.into());
         }
         Ok(())
     }

linera-core/src/worker.rs

@@ -412,6 +412,8 @@ pub enum WorkerError {
     MissingNetworkDescription,
     #[error("thread error: {0}")]
     Thread(#[from] web_thread_pool::Error),
+    #[error("Chain worker was poisoned by a journal resolution failure")]
+    PoisonedWorker,
 }
 
 impl WorkerError {
@@ -447,15 +449,23 @@ impl WorkerError {
             | WorkerError::MissingNetworkDescription
             | WorkerError::Thread(_)
             | WorkerError::ReadCertificatesError(_)
-            | WorkerError::IncorrectOutcome { .. } => true,
+            | WorkerError::IncorrectOutcome { .. }
+            | WorkerError::PoisonedWorker => true,
             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())
+    /// Returns `true` if this error indicates that the chain worker's in-memory
+    /// state may be inconsistent and must be evicted from the cache.
+    fn must_reload_view(&self) -> bool {
+        matches!(
+            self,
+            WorkerError::PoisonedWorker
+                | WorkerError::ViewError(ViewError::StoreError {
+                    must_reload_view: true,
+                    ..
+                })
+        )
     }
 }
 
@@ -773,17 +783,16 @@ where
         Fut: std::future::Future<Output = Result<R, WorkerError>>,
     {
         let state = self.get_or_create_chain_worker(chain_id).await?;
+        let state_ref = &state;
         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());
-            }
+            let guard = handle::read_lock(state_ref).await;
+            guard.check_not_poisoned()?;
             f(guard).await
         }))
         .await;
-        if let Err(e) = &result {
-            if e.must_reload_view() {
-                self.evict_poisoned_worker(chain_id);
+        if let Err(error) = &result {
+            if error.must_reload_view() {
+                self.evict_poisoned_worker(chain_id, &state);
             }
         }
         result
@@ -800,35 +809,34 @@ where
         Fut: std::future::Future<Output = Result<R, WorkerError>>,
     {
         let state = self.get_or_create_chain_worker(chain_id).await?;
+        let state_ref = &state;
         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());
-            }
+            let guard = handle::write_lock(state_ref).await;
+            guard.check_not_poisoned()?;
             f(guard).await
         }))
         .await;
-        if let Err(e) = &result {
-            if e.must_reload_view() {
-                self.evict_poisoned_worker(chain_id);
+        if let Err(error) = &result {
+            if error.must_reload_view() {
+                self.evict_poisoned_worker(chain_id, &state);
             }
         }
         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) {
+    /// Evicts a poisoned chain worker from the cache, but only if the entry still
+    /// points to the same instance. This avoids removing a fresh replacement that
+    /// another task may have already loaded.
+    fn evict_poisoned_worker(&self, chain_id: ChainId, poisoned: &ChainWorkerArc<StorageClient>) {
         tracing::warn!(%chain_id, "Evicting poisoned chain worker from cache");
         let pin = self.chain_workers.pin();
-        pin.remove(&chain_id);
+        let weak_poisoned = Arc::downgrade(poisoned);
+        let _ = pin.remove_if(&chain_id, |_key, future| {
+            future
+                .peek()
+                .and_then(|r| r.clone().ok())
+                .is_some_and(|weak| weak.ptr_eq(&weak_poisoned))
+        });
     }
 
     /// Gets or creates a chain worker for the given chain.