No fees for rejecting (#4875)

## Motivation

We charge fees for all outgoing messages, including bounced ones. That
means that a chain with very low balance can become unable to reject
tracked messages.

The client doesn't expect this, and keeps looping and setting the
failing message to `Reject`, even if it already is.

## Proposal

Don't charge for bouncing messages.

Also, charge only for the length of the operation outcome payload,
rather than the length of its serialized size, so that an empty outcome
is free, too.

Finally, we now break out of the loop in the client if a _rejected_
message fails.

## Test Plan

A new test was added that reproduced the issue.

## Release Plan

- Nothing to do / These changes follow the usual release cycle.
- I will prepare a separate PR for a migration of `testnet_conway` to
this new execution semantics. We can decide later whether we want to do
that.

## Links

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

Author: afck

linera-chain/src/block_tracker.rs

@@ -13,8 +13,8 @@ use linera_base::{
 };
 use linera_execution::{
     execution_state_actor::ExecutionStateActor, ExecutionRuntimeContext, ExecutionStateView,
-    MessageContext, OperationContext, OutgoingMessage, ResourceController, ResourceTracker,
-    SystemExecutionStateView, TransactionOutcome, TransactionTracker,
+    MessageContext, MessageKind, OperationContext, OutgoingMessage, ResourceController,
+    ResourceTracker, SystemExecutionStateView, TransactionOutcome, TransactionTracker,
 };
 use linera_views::context::Context;
 use tracing::instrument;
@@ -287,6 +287,9 @@ impl<'resources, 'blobs> BlockExecutionTracker<'resources, 'blobs> {
         let mut resource_controller = self.resource_controller.with_state(view).await?;
 
         for message_out in &txn_outcome.outgoing_messages {
+            if message_out.kind == MessageKind::Bouncing {
+                continue; // Bouncing messages are free.
+            }
             resource_controller
                 .track_message(&message_out.message)
                 .with_execution_context(context)?;
@@ -357,18 +360,6 @@ impl<'resources, 'blobs> BlockExecutionTracker<'resources, 'blobs> {
             .collect()
     }
 
-    /// Returns the execution context for the current transaction.
-    pub fn chain_execution_context(&self, transaction: &Transaction) -> ChainExecutionContext {
-        match transaction {
-            Transaction::ReceiveMessages(_) => {
-                ChainExecutionContext::IncomingBundle(self.transaction_index)
-            }
-            Transaction::ExecuteOperation(_) => {
-                ChainExecutionContext::Operation(self.transaction_index)
-            }
-        }
-    }
-
     /// Returns a mutable reference to the resource controller.
     pub fn resource_controller_mut(
         &mut self,
@@ -399,6 +390,18 @@ impl<'resources, 'blobs> BlockExecutionTracker<'resources, 'blobs> {
             self.operation_results,
         )
     }
+
+    /// Returns the execution context for the current transaction.
+    fn chain_execution_context(&self, transaction: &Transaction) -> ChainExecutionContext {
+        match transaction {
+            Transaction::ReceiveMessages(_) => {
+                ChainExecutionContext::IncomingBundle(self.transaction_index)
+            }
+            Transaction::ExecuteOperation(_) => {
+                ChainExecutionContext::Operation(self.transaction_index)
+            }
+        }
+    }
 }
 
 pub(crate) type FinalizeExecutionResult = (

linera-core/src/client/mod.rs

@@ -1450,26 +1450,29 @@ impl<Env: Environment> Client<Env> {
                         .transactions
                         .get_mut(*index as usize)
                         .expect("Transaction at given index should exist");
-                    let Transaction::ReceiveMessages(message) = transaction else {
+                    let Transaction::ReceiveMessages(incoming_bundle) = transaction else {
                         panic!(
                             "Expected incoming bundle at transaction index {}, found operation",
                             index
                         );
                     };
                     ensure!(
-                        !message.bundle.is_protected(),
+                        !incoming_bundle.bundle.is_protected(),
                         chain_client::Error::BlockProposalError(
                             "Protected incoming message failed to execute locally"
                         )
                     );
+                    if incoming_bundle.action == MessageAction::Reject {
+                        return result;
+                    }
                     // Reject the faulty message from the block and continue.
                     // TODO(#1420): This is potentially a bit heavy-handed for
                     // retryable errors.
                     info!(
-                        %error, origin = ?message.origin,
-                        "Message failed to execute locally and will be rejected."
+                        %error, %index, origin = ?incoming_bundle.origin,
+                        "Message bundle failed to execute locally and will be rejected."
                     );
-                    message.action = MessageAction::Reject;
+                    incoming_bundle.action = MessageAction::Reject;
                     continue;
                 }
             }

linera-core/src/unit_tests/client_tests.rs

@@ -2933,3 +2933,48 @@ where
 
     Ok(())
 }
+
+#[test_case(MemoryStorageBuilder::default(); "memory")]
+#[cfg_attr(feature = "storage-service", test_case(ServiceStorageBuilder::new(); "storage_service"))]
+#[cfg_attr(feature = "rocksdb", test_case(RocksDbStorageBuilder::new().await; "rocks_db"))]
+#[cfg_attr(feature = "dynamodb", test_case(DynamoDbStorageBuilder::default(); "dynamo_db"))]
+#[cfg_attr(feature = "scylladb", test_case(ScyllaDbStorageBuilder::default(); "scylla_db"))]
+#[test_log::test(tokio::test)]
+async fn test_rejected_message_bundles_are_free<B>(storage_builder: B) -> anyhow::Result<()>
+where
+    B: StorageBuilder,
+{
+    let signer = InMemorySigner::new(None);
+    let mut builder = TestBuilder::new(storage_builder, 4, 1, signer)
+        .await?
+        .with_policy(ResourceControlPolicy::testnet());
+    let admin = builder.add_root_chain(1, Amount::from_tokens(2)).await?;
+    let user = builder.add_root_chain(1, Amount::ZERO).await?;
+    let user_reject = builder
+        .make_client_with_options(
+            user.chain_id(),
+            None,
+            BlockHeight::ZERO,
+            chain_client::Options {
+                message_policy: MessagePolicy::new(BlanketMessagePolicy::Reject, None),
+                ..chain_client::Options::test_default()
+            },
+        )
+        .await?;
+
+    let recipient = Account::chain(user.chain_id());
+    admin
+        .transfer(AccountOwner::CHAIN, Amount::ONE, recipient)
+        .await
+        .unwrap_ok_committed();
+
+    user_reject.synchronize_from_validators().await?;
+    let (certificates, _) = user_reject.process_inbox().await.unwrap();
+    assert_eq!(certificates.len(), 1);
+    assert_matches!(
+        &certificates[0].block().body.transactions[0],
+        Transaction::ReceiveMessages(bundle) if bundle.action == MessageAction::Reject
+    );
+
+    Ok(())
+}

linera-core/src/unit_tests/test_utils.rs

@@ -1042,11 +1042,12 @@ where
         Ok(self.genesis_storage_builder.build(storage).await)
     }
 
-    pub async fn make_client(
+    pub async fn make_client_with_options(
         &mut self,
         chain_id: ChainId,
         block_hash: Option<CryptoHash>,
         block_height: BlockHeight,
+        options: chain_client::Options,
     ) -> anyhow::Result<ChainClient<B::Storage>> {
         // Note that new clients are only given the genesis store: they must figure out
         // the rest by asking validators.
@@ -1064,7 +1065,7 @@ where
             format!("Client node for {:.8}", chain_id),
             Duration::from_secs(30),
             Duration::from_secs(1),
-            chain_client::Options::test_default(),
+            options,
             5_000,
             10_000,
             crate::client::RequestsSchedulerConfig::default(),
@@ -1079,6 +1080,21 @@ where
         ))
     }
 
+    pub async fn make_client(
+        &mut self,
+        chain_id: ChainId,
+        block_hash: Option<CryptoHash>,
+        block_height: BlockHeight,
+    ) -> anyhow::Result<ChainClient<B::Storage>> {
+        self.make_client_with_options(
+            chain_id,
+            block_hash,
+            block_height,
+            chain_client::Options::test_default(),
+        )
+        .await
+    }
+
     /// Tries to find a (confirmation) certificate for the given chain_id and block height.
     pub async fn check_that_validators_have_certificate(
         &self,