[testnet] Fix poisoned worker race condition. (#5936) (#5940)

Backport of #5936.

## Motivation

#5936 fixed a potential race condition in case of a DB journaling error.

## Proposal

Ensure that the state that's removed from the map is really the poisoned
one.

Other minor cleanups to be closer to the `main` branch.

## Test Plan

CI

## Release Plan

- Validator hotfix.

## Links

- PR to `main`: #5936
- [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/handle.rs

@@ -82,11 +82,6 @@ impl<S: Storage + Clone + 'static> DerefMut for RollbackGuard<S> {
 
 impl<S: Storage + Clone + 'static> Drop for RollbackGuard<S> {
     fn drop(&mut self) {
-        if self.0.is_poisoned() {
-            // The view is in an inconsistent state due to a journal resolution failure.
-            // Don't rollback — the worker will be evicted and reloaded.
-            return;
-        }
         self.0.rollback();
     }
 }

linera-core/src/chain_worker/state.rs

@@ -195,9 +195,11 @@ where
         self.chain.rollback();
     }
 
-    /// Returns whether this worker is poisoned (view is inconsistent).
-    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.
@@ -2131,16 +2133,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,8 +412,8 @@ pub enum WorkerError {
 
     #[error("Fallback mode is not available on this network")]
     NoFallbackMode,
-    #[error("Chain worker for {chain_id} is poisoned and must be reloaded")]
-    WorkerPoisoned { chain_id: ChainId },
+    #[error("Chain worker was poisoned by a journal resolution failure")]
+    PoisonedWorker,
 }
 
 impl WorkerError {
@@ -449,15 +449,22 @@ impl WorkerError {
             | WorkerError::Thread(_)
             | WorkerError::ReadCertificatesError(_)
             | WorkerError::IncorrectOutcome { .. }
-            | WorkerError::WorkerPoisoned { .. } => true,
+            | 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,
+                    ..
+                })
+        )
     }
 }
 
@@ -756,16 +763,19 @@ 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 guard = handle::read_lock(&state).await;
-            if guard.is_poisoned() {
-                self.chain_workers.pin().remove(&chain_id);
-                drop(guard);
-                return Err(WorkerError::WorkerPoisoned { chain_id });
-            }
+        let state_ref = &state;
+        let result = Box::pin(wrap_future(async move {
+            let guard = handle::read_lock(state_ref).await;
+            guard.check_not_poisoned()?;
             f(guard).await
         }))
-        .await
+        .await;
+        if let Err(error) = &result {
+            if error.must_reload_view() {
+                self.evict_poisoned_worker(chain_id, &state);
+            }
+        }
+        result
     }
 
     /// Acquires a write lock on the chain worker and executes the given closure.
@@ -779,26 +789,34 @@ 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 guard = handle::write_lock(&state).await;
-            if guard.is_poisoned() {
-                self.chain_workers.pin().remove(&chain_id);
-                drop(guard);
-                return Err(WorkerError::WorkerPoisoned { chain_id });
-            }
-            let result = f(guard).await;
-            if result.is_err() {
-                // Check if the operation poisoned the worker. If so, evict it
-                // so the next request reloads from storage.
-                let guard = state.read().await;
-                if guard.is_poisoned() {
-                    self.chain_workers.pin().remove(&chain_id);
-                    drop(guard);
-                }
-            }
-            result
+        let state_ref = &state;
+        let result = Box::pin(wrap_future(async move {
+            let guard = handle::write_lock(state_ref).await;
+            guard.check_not_poisoned()?;
+            f(guard).await
         }))
-        .await
+        .await;
+        if let Err(error) = &result {
+            if error.must_reload_view() {
+                self.evict_poisoned_worker(chain_id, &state);
+            }
+        }
+        result
+    }
+
+    /// 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();
+        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.