Retry proposals without CheckpointAck bundles a validator cannot hold

When a validator rejects a proposal with MissingCrossChainUpdate, the updater
first tries to heal it by pushing the sender chain. That cannot work for a
CheckpointAck bundle on a validator that bootstrapped the sender chain from a
checkpoint later than the ack-sending block: the block sits below its tip, is
vouched for by nothing, and is never resent. With enough such validators (plus
ordinary unavailability), a proposal consuming the ack can never be certified,
and the client would keep re-proposing it forever.

The client now settles exactly the acknowledgement bundles named by such
rejections in its local inbox - removing them from the pending bundles and
advancing the lane's sender_pruned_cursor past them, the state a node that
never received them is in - and retries the proposal without them. Skipping
them is always legal: they are Simple, zero-grant bundles, and validators
that do hold one discard it when the lane's next bundle is consumed; if
another owner's quorum does consume it, the certified block reconciles as a
settled no-op. The settlement is part of the chain state, so it survives
client restarts and one-shot CLI commands instead of costing a failed
proposal round-trip each time. A leftover pending proposal from an
interrupted call gets the same treatment rather than wedging every future
proposal. If nothing else is pending, no block is proposed at all.

Author: afck

linera-chain/src/inbox.rs

@@ -235,6 +235,26 @@ where
         Ok(())
     }
 
+    /// Removes the added bundle at `cursor` and records the lane as settled up to and
+    /// including it, as if the sender had declared it so. Used for a bundle that can
+    /// never be consumed by this chain's blocks — enough validators cannot hold it that
+    /// no proposal consuming it gets certified. A certified consumption by another
+    /// owner's block still reconciles, as a no-op below `sender_pruned_cursor`.
+    pub async fn settle_unavailable_bundle(&mut self, cursor: Cursor) -> Result<(), ViewError> {
+        let bundles = self.added_bundles.elements().await?;
+        self.added_bundles.clear();
+        for bundle in bundles {
+            if bundle.cursor() != cursor {
+                self.added_bundles.push_back(bundle);
+            }
+        }
+        self.note_sender_pruned_below(Cursor {
+            height: cursor.height,
+            index: cursor.index.saturating_add(1),
+        })
+        .await
+    }
+
     /// Consumes a bundle from the inbox.
     ///
     /// Returns `true` if the bundle was already known, i.e. it was present in `added_bundles`.

linera-core/src/chain_worker/state.rs

@@ -15,8 +15,8 @@ use linera_base::prometheus_util::MeasureLatency as _;
 use linera_base::{
     crypto::{CryptoHash, ValidatorPublicKey},
     data_types::{
-        ApplicationDescription, ArithmeticError, Blob, BlockHeight, Epoch, OracleResponse, Round,
-        Timestamp,
+        ApplicationDescription, ArithmeticError, Blob, BlockHeight, Cursor, Epoch, OracleResponse,
+        Round, Timestamp,
     },
     ensure,
     hashed::Hashed,
@@ -1507,6 +1507,22 @@ where
         Ok(CrossChainUpdateResult::Updated(last_updated_height))
     }
 
+    /// Removes an incoming bundle that can never be consumed by this chain's blocks —
+    /// enough validators cannot hold it that no proposal consuming it gets certified —
+    /// and marks the lane as settled up to it, so that a certified consumption by
+    /// another owner's block still reconciles as a no-op.
+    pub(crate) async fn settle_unavailable_bundle(
+        &mut self,
+        origin: ChainId,
+        cursor: Cursor,
+    ) -> Result<(), WorkerError> {
+        let mut inbox = self.chain.inboxes.try_load_entry_mut(&origin).await?;
+        inbox.settle_unavailable_bundle(cursor).await?;
+        drop(inbox);
+        self.save().await?;
+        Ok(())
+    }
+
     /// Handles the cross-chain request confirming that the recipient was updated.
     #[instrument(skip_all, fields(
         chain_id = %self.chain_id(),

linera-core/src/client/chain_client/mod.rs

@@ -21,7 +21,7 @@ use linera_base::{
     crypto::{signer, CryptoHash, Signer, ValidatorPublicKey},
     data_types::{
         Amount, ApplicationDescription, ApplicationPermissions, ArithmeticError, Blob, BlobContent,
-        BlockHeight, ChainDescription, Epoch, MessagePolicy, Round, TimeDelta, Timestamp,
+        BlockHeight, ChainDescription, Cursor, Epoch, MessagePolicy, Round, TimeDelta, Timestamp,
     },
     ensure,
     identifiers::{
@@ -427,6 +427,22 @@ impl<Env: Environment> ChainClient<Env> {
             .proposal_mutex()
     }
 
+    /// Settles acknowledgement bundles that a validator rejected as unavailable in the
+    /// local inbox: they are removed from the pending bundles, durably, so future
+    /// proposals leave them out without a failed round-trip.
+    async fn settle_unavailable_acks(
+        &self,
+        acks: BTreeMap<(ChainId, BlockHeight), Cursor>,
+    ) -> Result<(), Error> {
+        for ((origin, _), cursor) in acks {
+            self.client
+                .local_node
+                .settle_unavailable_bundle(self.chain_id, origin, cursor)
+                .await?;
+        }
+        Ok(())
+    }
+
     /// Returns the pending proposal, if any.
     #[instrument(level = "trace", skip(self))]
     pub async fn pending_proposal(&self) -> Option<PendingProposal> {
@@ -1452,22 +1468,41 @@ impl<Env: Environment> ChainClient<Env> {
         // Even if there is no pending proposal, this still calls
         // `request_leader_timeout_if_needed` which ensures the local chain state
         // is synchronized with the current consensus round.
+        let pending_acks = proposal_guard
+            .as_ref()
+            .map(|pending| ack_bundles_in(&pending.block.transactions))
+            .unwrap_or_default();
         match self
             .process_pending_block_without_prepare(&mut proposal_guard)
-            .await?
+            .await
         {
-            ClientOutcome::Committed(Some(certificate)) => {
+            Ok(ClientOutcome::Committed(Some(certificate))) => {
                 return Ok(self.classify_committed(certificate, &operations));
             }
-            ClientOutcome::WaitForTimeout(timeout) => {
+            Ok(ClientOutcome::WaitForTimeout(timeout)) => {
                 return Ok(ClientOutcome::WaitForTimeout(timeout))
             }
-            ClientOutcome::Conflict(certificate) => {
+            Ok(ClientOutcome::Conflict(certificate)) => {
                 return Ok(ClientOutcome::Conflict(certificate))
             }
-            ClientOutcome::Committed(None) => {}
+            Ok(ClientOutcome::Committed(None)) => {}
+            Err(error) => {
+                // The leftover proposal itself may contain an unavailable
+                // acknowledgement bundle; discard it rather than stay wedged. The loop
+                // below rebuilds the block without the bundle.
+                let unavailable = unavailable_ack_bundles(&error, &pending_acks);
+                if unavailable.is_empty() {
+                    return Err(error);
+                }
+                warn!(
+                    chain_id = %self.chain_id,
+                    ?unavailable,
+                    "discarding a pending proposal with unavailable acknowledgement bundles",
+                );
+                self.settle_unavailable_acks(unavailable).await?;
+                *proposal_guard = None;
+            }
         }
-
         loop {
             // Collect pending messages and epoch changes after acquiring the lock to avoid
             // race conditions where messages valid for one block height are proposed at a
@@ -1483,13 +1518,38 @@ impl<Env: Environment> ChainClient<Env> {
                 )));
             }
 
+            let proposed_acks = ack_bundles_in(&transactions);
+
             self.new_pending_block(transactions, blobs.clone(), &mut proposal_guard)
                 .await?;
 
-            match self
+            let outcome = match self
                 .process_pending_block_without_prepare(&mut proposal_guard)
-                .await?
+                .await
             {
+                Ok(outcome) => outcome,
+                Err(error) => {
+                    let unavailable = unavailable_ack_bundles(&error, &proposed_acks);
+                    if unavailable.is_empty() {
+                        return Err(error);
+                    }
+                    // A validator rejected an acknowledgement bundle it does not hold —
+                    // e.g. it bootstrapped the sender chain from a checkpoint past the
+                    // acknowledgement, which nothing resends. Settle the bundle in the
+                    // local inbox and retry without it: validators that do hold it
+                    // discard it when the lane's next bundle is consumed.
+                    warn!(
+                        chain_id = %self.chain_id,
+                        ?unavailable,
+                        "retrying the proposal without unavailable acknowledgement bundles",
+                    );
+                    self.settle_unavailable_acks(unavailable).await?;
+                    *proposal_guard = None;
+                    continue;
+                }
+            };
+
+            match outcome {
                 ClientOutcome::Committed(Some(certificate)) => {
                     return Ok(self.classify_committed(certificate, &operations));
                 }
@@ -3529,6 +3589,60 @@ impl<Env: Environment> ChainClient<Env> {
     }
 }
 
+/// Returns the transactions' incoming bundles that consist solely of `CheckpointAck`
+/// messages, keyed by `(origin, height)` — the fields named by
+/// [`NodeError::MissingCrossChainUpdate`] — with their inbox cursors as values.
+fn ack_bundles_in(transactions: &[Transaction]) -> BTreeMap<(ChainId, BlockHeight), Cursor> {
+    transactions
+        .iter()
+        .filter_map(|transaction| match transaction {
+            Transaction::ReceiveMessages(incoming)
+                if incoming
+                    .bundle
+                    .messages
+                    .iter()
+                    .all(|posted| posted.message.is_checkpoint_ack()) =>
+            {
+                Some((
+                    (incoming.origin, incoming.bundle.height),
+                    incoming.bundle.cursor(),
+                ))
+            }
+            _ => None,
+        })
+        .collect()
+}
+
+/// Returns the subset of `ack_bundles` — the proposed bundles consisting solely of
+/// `CheckpointAck` messages, keyed by `(origin, height)` — that a validator rejected
+/// with [`NodeError::MissingCrossChainUpdate`]. An acknowledgement's bundle has no
+/// availability guarantee, so a validator that does not hold it never will; a proposal
+/// containing it can only be certified by retrying without it.
+fn unavailable_ack_bundles(
+    error: &Error,
+    ack_bundles: &BTreeMap<(ChainId, BlockHeight), Cursor>,
+) -> BTreeMap<(ChainId, BlockHeight), Cursor> {
+    let Error::CommunicationError(error) = error else {
+        return BTreeMap::new();
+    };
+    let node_errors = match error {
+        CommunicationError::Trusted(error) => vec![error],
+        CommunicationError::Sample(samples) => samples.iter().map(|(error, _)| error).collect(),
+        CommunicationError::NoConsensus(_, _) => Vec::new(),
+    };
+    node_errors
+        .into_iter()
+        .filter_map(|error| match error {
+            NodeError::MissingCrossChainUpdate { origin, height, .. } => {
+                let key = (*origin, *height);
+                let cursor = ack_bundles.get(&key)?;
+                Some((key, *cursor))
+            }
+            _ => None,
+        })
+        .collect()
+}
+
 #[cfg(with_testing)]
 impl<Env: Environment> ChainClient<Env> {
     /// Processes a notification received from the given validator.

linera-core/src/local_node.rs

@@ -491,6 +491,21 @@ where
         Ok(self.node.state.reset_and_reexecute_chain(chain_id).await?)
     }
 
+    /// Removes an incoming bundle of `chain_id` that can never be consumed by its
+    /// blocks, and marks the lane from `origin` as settled up to it.
+    pub async fn settle_unavailable_bundle(
+        &self,
+        chain_id: ChainId,
+        origin: ChainId,
+        cursor: linera_base::data_types::Cursor,
+    ) -> Result<(), LocalNodeError> {
+        Ok(self
+            .node
+            .state
+            .settle_unavailable_bundle(chain_id, origin, cursor)
+            .await?)
+    }
+
     /// Gets received certificate trackers.
     pub async fn get_received_certificate_trackers(
         &self,

linera-core/src/unit_tests/client_tests.rs

@@ -4573,6 +4573,102 @@ where
     Ok(())
 }
 
+/// A proposal consuming a `CheckpointAck` bundle that some validator can never hold
+/// must be retried without it. Validator 0 misses the recipient's ack-sending
+/// checkpoint and is later bootstrapped from the recipient's *next* checkpoint, so the
+/// ack block stays below its tip: pushing certificates cannot heal it, and it rejects
+/// every ack-consuming proposal with `MissingCrossChainUpdate`. With a second validator
+/// offline, such a proposal cannot reach quorum — the client must drop the skippable
+/// ack bundle and move on instead of wedging.
+#[test_case(MemoryStorageBuilder::default(); "memory")]
+#[cfg_attr(feature = "rocksdb", test_case(RocksDbStorageBuilder::new().await; "rocks_db"))]
+#[test_log::test(tokio::test)]
+async fn test_unavailable_checkpoint_ack_is_skipped<B>(storage_builder: B) -> anyhow::Result<()>
+where
+    B: StorageBuilder,
+{
+    let signer = InMemorySigner::new(None);
+    let mut builder = TestBuilder::new(storage_builder, 4, 0, signer).await?;
+    // Validator 0 sees nothing of what follows: not the transfer, and crucially not the
+    // recipient's checkpoints.
+    builder.set_fault_type([0], FaultType::NoChains);
+    let producer = builder.add_root_chain(1, Amount::from_tokens(7)).await?;
+    let recipient = builder.add_root_chain(2, Amount::ZERO).await?;
+    let recipient_id = recipient.chain_id();
+
+    // The recipient consumes a transfer and checkpoints, acknowledging it; then burns
+    // and checkpoints again, so its latest checkpoint sits above the ack-sending block.
+    producer
+        .transfer_to_account(
+            AccountOwner::CHAIN,
+            Amount::from_tokens(2),
+            Account::chain(recipient_id),
+        )
+        .await
+        .unwrap_ok_committed();
+    recipient.synchronize_from_validators().await?;
+    recipient.process_inbox().await?;
+    recipient.checkpoint().await.unwrap().unwrap();
+    recipient
+        .burn(AccountOwner::CHAIN, Amount::ONE)
+        .await
+        .unwrap_ok_committed();
+    recipient.checkpoint().await.unwrap().unwrap();
+
+    // Validator 0 comes back and validator 1 goes offline. The recipient's next block
+    // bootstraps validator 0 from the latest checkpoint — past the ack-sending block,
+    // which nothing can deliver to it anymore.
+    builder.set_fault_type([0], FaultType::Honest);
+    builder.set_fault_type([1], FaultType::Offline);
+    recipient
+        .burn(AccountOwner::CHAIN, Amount::ONE)
+        .await
+        .unwrap_ok_committed();
+
+    // The producer's inbox holds only the unavailable ack. Consuming it can never be
+    // certified (validator 0 rejects it, validator 1 is offline), so the client skips
+    // the bundle; with nothing else to include, no block is proposed at all.
+    producer.synchronize_from_validators().await?;
+    let (certificates, _) = producer.process_inbox().await?;
+    assert!(
+        certificates.is_empty(),
+        "the unconsumable ack must be skipped, not wedge the client",
+    );
+
+    // The skip is durable: the ack was settled in the local inbox, so a fresh
+    // `process_inbox` — e.g. after a client restart — has nothing to retry.
+    {
+        let chain = producer
+            .client
+            .local_node
+            .chain_state_view(producer.chain_id())
+            .await?;
+        let inbox = chain
+            .inboxes
+            .try_load_entry(&recipient_id)
+            .await?
+            .expect("the producer has an inbox for the recipient");
+        assert_eq!(
+            inbox.added_bundles.count(),
+            0,
+            "the settled ack must be gone from the local inbox",
+        );
+    }
+
+    // The chain stays fully usable: a new transfer commits — without the ack.
+    let certificate = producer
+        .transfer_to_account(
+            AccountOwner::CHAIN,
+            Amount::ONE,
+            Account::chain(recipient_id),
+        )
+        .await
+        .unwrap_ok_committed();
+    assert!(!certificate.block().consumes_checkpoint_ack());
+
+    Ok(())
+}
+
 /// Verifies the push side of the checkpoint flow: a validator that missed the whole
 /// chain is brought up to speed by the proposing client pushing only the latest
 /// checkpoint plus the pre-checkpoint sender blocks it certifies, not every

linera-core/src/worker.rs

@@ -17,8 +17,8 @@ use futures::{
 use linera_base::{
     crypto::{CryptoError, CryptoHash, ValidatorPublicKey},
     data_types::{
-        ApplicationDescription, ArithmeticError, Blob, BlockHeight, Epoch, Round, TimeDelta,
-        Timestamp,
+        ApplicationDescription, ArithmeticError, Blob, BlockHeight, Cursor, Epoch, Round,
+        TimeDelta, Timestamp,
     },
     doc_scalar,
     identifiers::{AccountOwner, ApplicationId, BlobId, ChainId, EventId, StreamId},
@@ -1577,6 +1577,20 @@ where
             .await
     }
 
+    /// Removes an incoming bundle of `chain_id` that can never be consumed by its
+    /// blocks, and marks the lane from `origin` as settled up to it. See
+    /// [`ChainWorkerState::settle_unavailable_bundle`].
+    pub async fn settle_unavailable_bundle(
+        &self,
+        chain_id: ChainId,
+        origin: ChainId,
+        cursor: Cursor,
+    ) -> Result<(), WorkerError> {
+        let state = self.get_or_create_chain_worker(chain_id).await?;
+        let mut guard = handle::write_lock(&state).await?;
+        guard.settle_unavailable_bundle(origin, cursor).await
+    }
+
     /// Test helper that runs `ChainWorkerState::reset_and_reexecute_chain` for the given
     /// chain (the same routine the corruption-recovery path invokes).
     #[cfg(with_testing)]