Add FLAG_FREE_REJECT: with it, rejecting messages is free of charge. (#4876)

## Motivation

In #4875, we make rejecting messages free of charge.

## Proposal

Backport this to the testnet.

Since it changes execution semantics, it needs to be behind a new flag
`FLAG_FREE_REJECT.linera.network`.

## Test Plan

CI

## Release Plan

- These changes should
    - be released in a new SDK,
    - be released in a validator hotfix,
    - be activated via a resource policy change in a new epoch.

## Links

- PR to main: #4875
- [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,9 @@ 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,
+    FLAG_FREE_REJECT,
 };
 use linera_views::context::Context;
 use tracing::instrument;
@@ -287,6 +288,14 @@ 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
+                && resource_controller
+                    .policy()
+                    .http_request_allow_list
+                    .contains(FLAG_FREE_REJECT)
+            {
+                continue; // Bouncing messages are free.
+            }
             resource_controller
                 .track_message(&message_out.message)
                 .with_execution_context(context)?;

linera-core/src/unit_tests/client_tests.rs

@@ -25,6 +25,7 @@ use linera_chain::{
 use linera_execution::{
     committee::Committee, system::SystemOperation, ExecutionError, Message, MessageKind, Operation,
     QueryOutcome, ResourceControlPolicy, SystemMessage, SystemQuery, SystemResponse,
+    FLAG_FREE_REJECT,
 };
 use linera_storage::Storage;
 use rand::Rng;
@@ -44,8 +45,8 @@ use crate::test_utils::ScyllaDbStorageBuilder;
 use crate::test_utils::ServiceStorageBuilder;
 use crate::{
     client::{
-        BlanketMessagePolicy, ChainClient, ChainClientError, ClientOutcome, MessageAction,
-        MessagePolicy,
+        BlanketMessagePolicy, ChainClient, ChainClientError, ChainClientOptions, ClientOutcome,
+        MessageAction, MessagePolicy,
     },
     local_node::LocalNodeError,
     node::{
@@ -2835,3 +2836,51 @@ 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 {
+            http_request_allow_list: BTreeSet::from([FLAG_FREE_REJECT.to_string()]),
+            ..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,
+            ChainClientOptions {
+                message_policy: MessagePolicy::new(BlanketMessagePolicy::Reject, None),
+                ..ChainClientOptions::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

@@ -1029,11 +1029,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: ChainClientOptions,
     ) -> anyhow::Result<ChainClient<B::Storage>> {
         // Note that new clients are only given the genesis store: they must figure out
         // the rest by asking validators.
@@ -1051,7 +1052,7 @@ where
             format!("Client node for {:.8}", chain_id),
             Duration::from_secs(30),
             Duration::from_secs(1),
-            ChainClientOptions::test_default(),
+            options,
             crate::client::RequestsSchedulerConfig::default(),
         ));
         Ok(client.create_chain_client(
@@ -1064,6 +1065,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,
+            ChainClientOptions::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,

linera-execution/src/lib.rs

@@ -90,6 +90,10 @@ const MAX_STREAM_NAME_LEN: usize = 64;
 /// returned to be all zeros.
 // Note: testnet-only! This should not survive to mainnet.
 pub const FLAG_ZERO_HASH: &str = "FLAG_ZERO_HASH.linera.network";
+/// The flag that deactivates charging for bouncing messages. If this is present, outgoing
+/// messages are free of charge if they are bouncing, and operation outcomes are counted only
+/// by payload size, so that rejecting messages is free.
+pub const FLAG_FREE_REJECT: &str = "FLAG_FREE_REJECT.linera.network";
 
 /// An implementation of [`UserContractModule`].
 #[derive(Clone)]