Avoid using shared chain state views except for GraphQL. (#4797)

## Motivation

With the exception of a GraphQL query, the chain state view should only
be accessed by the corresponding chain worker. GraphQL is why the worker
can return a shared chain state view behind a lock, but we are using
that in many other places, too. After
#4769, this is even
more dangerous, as an actor could be dropped and restarted while shared
views are still being used.

## Proposal

Replace several `chain_state_view()` calls with chain info or new chain
worker requests.

Also, extend a comment. (See
#4793 (comment).)

## Test Plan

CI

## Release Plan

- These changes should be backported to `testnet_conway`, then
    - be released in a new SDK.

## Links

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

Author: afck

linera-chain/src/chain.rs

@@ -655,6 +655,10 @@ where
     }
 
     /// Removes the incoming message bundles in the block from the inboxes.
+    ///
+    /// If `must_be_present` is `true`, an error is returned if any of the bundles have not been
+    /// added to the inbox yet. So this should be `true` if the bundles are in a block _proposal_,
+    /// and `false` if the block is already confirmed.
     #[instrument(target = "telemetry_only", skip_all, fields(
         chain_id = %self.chain_id(),
     ))]

linera-core/src/chain_worker/actor.rs

@@ -157,6 +157,36 @@ where
         new_trackers: BTreeMap<ValidatorPublicKey, u64>,
         callback: oneshot::Sender<Result<(), WorkerError>>,
     },
+
+    /// Get preprocessed block hashes in a given height range.
+    GetPreprocessedBlockHashes {
+        start: BlockHeight,
+        end: BlockHeight,
+        #[debug(skip)]
+        callback: oneshot::Sender<Result<Vec<CryptoHash>, WorkerError>>,
+    },
+
+    /// Get the next block height to receive from an inbox.
+    GetInboxNextHeight {
+        origin: ChainId,
+        #[debug(skip)]
+        callback: oneshot::Sender<Result<BlockHeight, WorkerError>>,
+    },
+
+    /// Get locking blobs for specific blob IDs.
+    GetLockingBlobs {
+        blob_ids: Vec<BlobId>,
+        #[debug(skip)]
+        callback: oneshot::Sender<Result<Option<Vec<Blob>>, WorkerError>>,
+    },
+
+    /// Read a range from the confirmed log.
+    ReadConfirmedLog {
+        start: BlockHeight,
+        end: BlockHeight,
+        #[debug(skip)]
+        callback: oneshot::Sender<Result<Vec<CryptoHash>, WorkerError>>,
+    },
 }
 
 /// The actor worker type.

linera-core/src/chain_worker/state.rs

@@ -208,6 +208,26 @@ where
                         .await,
                 )
                 .is_ok(),
+            ChainWorkerRequest::GetPreprocessedBlockHashes {
+                start,
+                end,
+                callback,
+            } => callback
+                .send(self.get_preprocessed_block_hashes(start, end).await)
+                .is_ok(),
+            ChainWorkerRequest::GetInboxNextHeight { origin, callback } => callback
+                .send(self.get_inbox_next_height(origin).await)
+                .is_ok(),
+            ChainWorkerRequest::GetLockingBlobs { blob_ids, callback } => callback
+                .send(self.get_locking_blobs(blob_ids).await)
+                .is_ok(),
+            ChainWorkerRequest::ReadConfirmedLog {
+                start,
+                end,
+                callback,
+            } => callback
+                .send(self.read_confirmed_log(start, end).await)
+                .is_ok(),
         };
 
         if !responded {
@@ -1030,6 +1050,92 @@ where
         Ok(())
     }
 
+    /// Returns the preprocessed block hashes in the given height range.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        start = %start,
+        end = %end
+    ))]
+    async fn get_preprocessed_block_hashes(
+        &self,
+        start: BlockHeight,
+        end: BlockHeight,
+    ) -> Result<Vec<CryptoHash>, WorkerError> {
+        let mut hashes = Vec::new();
+        let mut height = start;
+        while height < end {
+            match self.chain.preprocessed_blocks.get(&height).await? {
+                Some(hash) => hashes.push(hash),
+                None => break,
+            }
+            height = height.try_add_one()?;
+        }
+        Ok(hashes)
+    }
+
+    /// Returns the next block height to receive from an inbox.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        origin = %origin
+    ))]
+    async fn get_inbox_next_height(&self, origin: ChainId) -> Result<BlockHeight, WorkerError> {
+        Ok(match self.chain.inboxes.try_load_entry(&origin).await? {
+            Some(inbox) => inbox.next_block_height_to_receive()?,
+            None => BlockHeight::ZERO,
+        })
+    }
+
+    /// Returns the locking blobs for the given blob IDs.
+    /// Returns `Ok(None)` if any of the blobs is not found.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        num_blob_ids = %blob_ids.len()
+    ))]
+    async fn get_locking_blobs(
+        &self,
+        blob_ids: Vec<BlobId>,
+    ) -> Result<Option<Vec<Blob>>, WorkerError> {
+        let mut blobs = Vec::new();
+        for blob_id in blob_ids {
+            match self.chain.manager.locking_blobs.get(&blob_id).await? {
+                None => return Ok(None),
+                Some(blob) => blobs.push(blob),
+            }
+        }
+        Ok(Some(blobs))
+    }
+
+    /// Reads a range from the confirmed log.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id(),
+        start = %start,
+        end = %end
+    ))]
+    async fn read_confirmed_log(
+        &self,
+        start: BlockHeight,
+        end: BlockHeight,
+    ) -> Result<Vec<CryptoHash>, WorkerError> {
+        let start_usize = usize::try_from(start)?;
+        let end_usize = usize::try_from(end)?;
+        let log_heights: Vec<_> = (start_usize..end_usize).collect();
+        let hashes = self
+            .chain
+            .confirmed_log
+            .multi_get(log_heights.clone())
+            .await?
+            .into_iter()
+            .enumerate()
+            .map(|(i, maybe_hash)| {
+                maybe_hash.ok_or_else(|| WorkerError::ConfirmedLogEntryNotFound {
+                    height: BlockHeight(u64::try_from(start_usize + i).unwrap_or(u64::MAX)),
+                    chain_id: self.chain_id(),
+                })
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        Ok(hashes)
+    }
+
     /// Attempts to vote for a leader timeout, if possible.
     #[instrument(target = "telemetry_only", skip_all, fields(
         chain_id = %self.chain_id(),

linera-core/src/client/mod.rs

@@ -402,19 +402,12 @@ impl<Env: Environment> Client<Env> {
     ) -> Result<Option<Box<ChainInfo>>, ChainClientError> {
         let mut last_info = None;
         // First load any blocks from local storage, if available.
-        let mut hashes = Vec::new();
-        let mut next_height = BlockHeight::ZERO;
-        {
-            let chain = self.local_node.chain_state_view(chain_id).await?;
-            next_height = next_height.max(chain.tip_state.get().next_block_height);
-            while next_height < stop {
-                let Some(hash) = chain.preprocessed_blocks.get(&next_height).await? else {
-                    break;
-                };
-                hashes.push(hash);
-                next_height = next_height.try_add_one()?;
-            }
-        }
+        let chain_info = self.local_node.chain_info(chain_id).await?;
+        let mut next_height = chain_info.next_block_height;
+        let hashes = self
+            .local_node
+            .get_preprocessed_block_hashes(chain_id, next_height, stop)
+            .await?;
         let certificates = self
             .storage_client()
             .read_certificates(hashes.clone())
@@ -3971,11 +3964,11 @@ impl<Env: Environment> ChainClient<Env> {
         &self,
         origin: ChainId,
     ) -> Result<BlockHeight, ChainClientError> {
-        let chain = self.chain_state_view().await?;
-        Ok(match chain.inboxes.try_load_entry(&origin).await? {
-            Some(inbox) => inbox.next_block_height_to_receive()?,
-            None => BlockHeight::ZERO,
-        })
+        Ok(self
+            .client
+            .local_node
+            .get_inbox_next_height(self.chain_id, origin)
+            .await?)
     }
 
     #[instrument(level = "trace", skip(remote_node, local_node, notification))]

linera-core/src/local_node.rs

@@ -187,15 +187,12 @@ where
         blob_ids: impl IntoIterator<Item = &BlobId>,
         chain_id: ChainId,
     ) -> Result<Option<Vec<Blob>>, LocalNodeError> {
-        let chain = self.chain_state_view(chain_id).await?;
-        let mut blobs = Vec::new();
-        for blob_id in blob_ids {
-            match chain.manager.locking_blobs.get(blob_id).await? {
-                None => return Ok(None),
-                Some(blob) => blobs.push(blob),
-            }
-        }
-        Ok(Some(blobs))
+        let blob_ids_vec: Vec<_> = blob_ids.into_iter().copied().collect();
+        Ok(self
+            .node
+            .state
+            .get_locking_blobs(chain_id, blob_ids_vec)
+            .await?)
     }
 
     /// Writes the given blobs to storage if there is an appropriate blob state.
@@ -304,6 +301,31 @@ where
             .await?;
         Ok(())
     }
+
+    pub async fn get_preprocessed_block_hashes(
+        &self,
+        chain_id: ChainId,
+        start: BlockHeight,
+        end: BlockHeight,
+    ) -> Result<Vec<linera_base::crypto::CryptoHash>, LocalNodeError> {
+        Ok(self
+            .node
+            .state
+            .get_preprocessed_block_hashes(chain_id, start, end)
+            .await?)
+    }
+
+    pub async fn get_inbox_next_height(
+        &self,
+        chain_id: ChainId,
+        origin: ChainId,
+    ) -> Result<BlockHeight, LocalNodeError> {
+        Ok(self
+            .node
+            .state
+            .get_inbox_next_height(chain_id, origin)
+            .await?)
+    }
 }
 
 /// Extension trait for [`ChainInfo`]s from our local node. These should always be valid and

linera-core/src/worker.rs

@@ -1195,6 +1195,87 @@ where
         })
         .await
     }
+
+    /// Gets preprocessed block hashes in a given height range.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        nickname = %self.nickname,
+        chain_id = %chain_id,
+        start = %start,
+        end = %end
+    ))]
+    pub async fn get_preprocessed_block_hashes(
+        &self,
+        chain_id: ChainId,
+        start: BlockHeight,
+        end: BlockHeight,
+    ) -> Result<Vec<CryptoHash>, WorkerError> {
+        self.query_chain_worker(chain_id, move |callback| {
+            ChainWorkerRequest::GetPreprocessedBlockHashes {
+                start,
+                end,
+                callback,
+            }
+        })
+        .await
+    }
+
+    /// Gets the next block height to receive from an inbox.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        nickname = %self.nickname,
+        chain_id = %chain_id,
+        origin = %origin
+    ))]
+    pub async fn get_inbox_next_height(
+        &self,
+        chain_id: ChainId,
+        origin: ChainId,
+    ) -> Result<BlockHeight, WorkerError> {
+        self.query_chain_worker(chain_id, move |callback| {
+            ChainWorkerRequest::GetInboxNextHeight { origin, callback }
+        })
+        .await
+    }
+
+    /// Gets locking blobs for specific blob IDs.
+    /// Returns `Ok(None)` if any of the blobs is not found.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        nickname = %self.nickname,
+        chain_id = %chain_id,
+        num_blob_ids = %blob_ids.len()
+    ))]
+    pub async fn get_locking_blobs(
+        &self,
+        chain_id: ChainId,
+        blob_ids: Vec<BlobId>,
+    ) -> Result<Option<Vec<Blob>>, WorkerError> {
+        self.query_chain_worker(chain_id, move |callback| {
+            ChainWorkerRequest::GetLockingBlobs { blob_ids, callback }
+        })
+        .await
+    }
+
+    /// Reads a range from the confirmed log.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        nickname = %self.nickname,
+        chain_id = %chain_id,
+        start = %start,
+        end = %end
+    ))]
+    pub async fn read_confirmed_log(
+        &self,
+        chain_id: ChainId,
+        start: BlockHeight,
+        end: BlockHeight,
+    ) -> Result<Vec<CryptoHash>, WorkerError> {
+        self.query_chain_worker(chain_id, move |callback| {
+            ChainWorkerRequest::ReadConfirmedLog {
+                start,
+                end,
+                callback,
+            }
+        })
+        .await
+    }
 }
 
 #[cfg(with_testing)]

linera-service/src/node_service.rs

@@ -661,10 +661,7 @@ where
         let client = self.context.lock().await.make_chain_client(chain_id);
         let hash = match hash {
             Some(hash) => Some(hash),
-            None => {
-                let view = client.chain_state_view().await?;
-                view.tip_state.get().block_hash
-            }
+            None => client.chain_info().await?.block_hash,
         };
         if let Some(hash) = hash {
             let block = client.read_confirmed_block(hash).await?;
@@ -699,10 +696,7 @@ where
         let limit = limit.unwrap_or(10);
         let from = match from {
             Some(from) => Some(from),
-            None => {
-                let view = client.chain_state_view().await?;
-                view.tip_state.get().block_hash
-            }
+            None => client.chain_info().await?.block_hash,
         };
         let Some(from) = from else {
             return Ok(vec![]);