Allow filtering incoming messages by application.

CLI.md

@@ -169,6 +169,8 @@ Client implementation and command-line tool for the Linera blockchain
     Don't include any messages in blocks, and don't make any decision whether to accept or reject
 
 * `--restrict-chain-ids-to <RESTRICT_CHAIN_IDS_TO>` — A set of chains to restrict incoming messages from. By default, messages from all chains are accepted. To reject messages from all chains, specify an empty string
+* `--accept-messages-with-application-ids <ACCEPT_MESSAGES_WITH_APPLICATION_IDS>` — A set of application IDs. If specified, only bundles with at least one message from one of these applications will be accepted
+* `--reject-messages-with-other-application-ids <REJECT_MESSAGES_WITH_OTHER_APPLICATION_IDS>` — A set of application IDs. If specified, only bundles where all messages are from one of these applications will be accepted
 * `--timings` — Enable timing reports during operations
 * `--timing-interval <TIMING_INTERVAL>` — Interval in seconds between timing reports (defaults to 5)
 

linera-client/src/client_options.rs

@@ -5,7 +5,7 @@ use std::{collections::HashSet, fmt, iter, path::PathBuf};
 
 use linera_base::{
     data_types::{ApplicationPermissions, TimeDelta},
-    identifiers::{AccountOwner, ApplicationId, ChainId},
+    identifiers::{AccountOwner, ApplicationId, ChainId, GenericApplicationId},
     ownership::{ChainOwnership, TimeoutConfig},
     time::Duration,
 };
@@ -135,6 +135,16 @@ pub struct ClientContextOptions {
     #[arg(long, value_parser = util::parse_chain_set)]
     pub restrict_chain_ids_to: Option<HashSet<ChainId>>,
 
+    /// A set of application IDs. If specified, only bundles with at least one message from one of
+    /// these applications will be accepted.
+    #[arg(long, value_parser = util::parse_app_set)]
+    pub accept_messages_with_application_ids: Option<HashSet<GenericApplicationId>>,
+
+    /// A set of application IDs. If specified, only bundles where all messages are from one of
+    /// these applications will be accepted.
+    #[arg(long, value_parser = util::parse_app_set)]
+    pub reject_messages_with_other_application_ids: Option<HashSet<GenericApplicationId>>,
+
     /// Enable timing reports during operations
     #[cfg(not(web))]
     #[arg(long)]
@@ -247,6 +257,8 @@ impl ClientContextOptions {
         let message_policy = MessagePolicy::new(
             self.blanket_message_policy,
             self.restrict_chain_ids_to.clone(),
+            self.accept_messages_with_application_ids.clone(),
+            self.reject_messages_with_other_application_ids.clone(),
         );
         let cross_chain_message_delivery =
             CrossChainMessageDelivery::new(self.wait_for_outgoing_messages);

linera-client/src/util.rs

@@ -7,7 +7,7 @@ use futures::future;
 use linera_base::{
     crypto::CryptoError,
     data_types::{TimeDelta, Timestamp},
-    identifiers::ChainId,
+    identifiers::{ApplicationId, ChainId, GenericApplicationId},
     time::Duration,
 };
 use linera_core::{data_types::RoundTimeout, node::NotificationStream, worker::Reason};
@@ -32,6 +32,16 @@ pub fn parse_chain_set(s: &str) -> Result<HashSet<ChainId>, CryptoError> {
     }
 }
 
+pub fn parse_app_set(s: &str) -> anyhow::Result<HashSet<GenericApplicationId>> {
+    s.trim()
+        .split(",")
+        .map(|app_str| {
+            GenericApplicationId::from_str(app_str)
+                .or_else(|_| Ok(ApplicationId::from_str(app_str)?.into()))
+        })
+        .collect()
+}
+
 pub fn parse_ascii_alphanumeric_string(s: &str) -> Result<String, &'static str> {
     if s.chars().all(|x| x.is_ascii_alphanumeric()) {
         Ok(s.to_string())

linera-core/Cargo.toml

@@ -98,6 +98,7 @@ alloy-primitives.workspace = true
 assert_matches.workspace = true
 counter.workspace = true
 criterion.workspace = true
+crowd-funding.workspace = true
 fungible.workspace = true
 hex-game.workspace = true
 linera-core = { path = ".", default-features = false, features = ["test"] }

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

@@ -521,12 +521,7 @@ impl<Env: Environment> ChainClient<Env> {
         Ok(info
             .requested_pending_message_bundles
             .into_iter()
-            .filter_map(|mut bundle| {
-                self.options
-                    .message_policy
-                    .must_handle(&mut bundle)
-                    .then_some(bundle)
-            })
+            .filter_map(|bundle| self.options.message_policy.apply(bundle))
             .take(self.options.max_pending_message_bundles)
             .collect())
     }

linera-core/src/client/mod.rs

@@ -19,7 +19,7 @@ use linera_base::{
     crypto::{CryptoHash, Signer, ValidatorPublicKey},
     data_types::{ArithmeticError, Blob, BlockHeight, ChainDescription, Epoch},
     ensure,
-    identifiers::{AccountOwner, BlobId, BlobType, ChainId, StreamId},
+    identifiers::{AccountOwner, BlobId, BlobType, ChainId, GenericApplicationId, StreamId},
     time::Duration,
 };
 #[cfg(not(target_arch = "wasm32"))]
@@ -139,6 +139,10 @@ pub struct MessagePolicy {
     /// accepted. `Option::None` means that messages from all chains are accepted. An empty
     /// `HashSet` denotes that messages from no chains are accepted.
     restrict_chain_ids_to: Option<HashSet<ChainId>>,
+    /// A collection of applications: If `Some`, only bundles with at least one message by any
+    /// of these applications will be accepted.
+    accept_messages_with_application_ids: Option<HashSet<GenericApplicationId>>,
+    reject_messages_with_other_application_ids: Option<HashSet<GenericApplicationId>>,
 }
 
 #[derive(Copy, Clone, Debug, clap::ValueEnum)]
@@ -157,10 +161,14 @@ impl MessagePolicy {
     pub fn new(
         blanket: BlanketMessagePolicy,
         restrict_chain_ids_to: Option<HashSet<ChainId>>,
+        accept_messages_with_application_ids: Option<HashSet<GenericApplicationId>>,
+        reject_messages_with_other_application_ids: Option<HashSet<GenericApplicationId>>,
     ) -> Self {
         Self {
             blanket,
             restrict_chain_ids_to,
+            accept_messages_with_application_ids,
+            reject_messages_with_other_application_ids,
         }
     }
 
@@ -169,22 +177,42 @@ impl MessagePolicy {
         Self {
             blanket: BlanketMessagePolicy::Accept,
             restrict_chain_ids_to: None,
+            accept_messages_with_application_ids: None,
+            reject_messages_with_other_application_ids: None,
         }
     }
 
     #[instrument(level = "trace", skip(self))]
-    fn must_handle(&self, bundle: &mut IncomingBundle) -> bool {
+    fn apply(&self, mut bundle: IncomingBundle) -> Option<IncomingBundle> {
+        if let Some(chain_ids) = &self.restrict_chain_ids_to {
+            if !chain_ids.contains(&bundle.origin) {
+                return None;
+            }
+        }
+        if let Some(app_ids) = &self.accept_messages_with_application_ids {
+            if !bundle
+                .messages()
+                .any(|posted_msg| app_ids.contains(&posted_msg.message.application_id()))
+            {
+                return None;
+            }
+        }
+        if let Some(app_ids) = &self.reject_messages_with_other_application_ids {
+            if !bundle
+                .messages()
+                .all(|posted_msg| app_ids.contains(&posted_msg.message.application_id()))
+            {
+                return None;
+            }
+        }
         if self.is_reject() {
             if bundle.bundle.is_skippable() {
-                return false;
+                return None;
             } else if !bundle.bundle.is_protected() {
                 bundle.action = MessageAction::Reject;
             }
         }
-        match &self.restrict_chain_ids_to {
-            None => true,
-            Some(chains) => chains.contains(&bundle.origin),
-        }
+        Some(bundle)
     }
 
     #[instrument(level = "trace", skip(self))]

linera-core/src/unit_tests/client_tests.rs

@@ -2697,7 +2697,8 @@ where
         Amount::from_tokens(3)
     );
 
-    receiver.options_mut().message_policy = MessagePolicy::new(BlanketMessagePolicy::Ignore, None);
+    receiver.options_mut().message_policy =
+        MessagePolicy::new(BlanketMessagePolicy::Ignore, None, None, None);
     receiver.synchronize_from_validators().await?;
     assert!(receiver.process_inbox().await?.0.is_empty());
     // The message was ignored.
@@ -2708,7 +2709,8 @@ where
         Amount::from_tokens(3)
     );
 
-    receiver.options_mut().message_policy = MessagePolicy::new(BlanketMessagePolicy::Reject, None);
+    receiver.options_mut().message_policy =
+        MessagePolicy::new(BlanketMessagePolicy::Reject, None, None, None);
     let certs = receiver.process_inbox().await?.0;
     assert_eq!(certs.len(), 1);
     sender.synchronize_from_validators().await?;
@@ -2742,6 +2744,8 @@ where
     receiver.options_mut().message_policy = MessagePolicy::new(
         BlanketMessagePolicy::Accept,
         Some([sender.chain_id()].into_iter().collect()),
+        None,
+        None,
     );
     receiver.synchronize_from_validators().await?;
     let certs = receiver.process_inbox().await?.0;
@@ -2750,7 +2754,8 @@ where
     assert_eq!(receiver.local_balance().await.unwrap(), Amount::ONE);
 
     // Let's accept the other one, too.
-    receiver.options_mut().message_policy = MessagePolicy::new(BlanketMessagePolicy::Accept, None);
+    receiver.options_mut().message_policy =
+        MessagePolicy::new(BlanketMessagePolicy::Accept, None, None, None);
     let certs = receiver.process_inbox().await?.0;
     assert_eq!(certs.len(), 1);
     assert_eq!(
@@ -3121,7 +3126,7 @@ where
             None,
             BlockHeight::ZERO,
             chain_client::Options {
-                message_policy: MessagePolicy::new(BlanketMessagePolicy::Reject, None),
+                message_policy: MessagePolicy::new(BlanketMessagePolicy::Reject, None, None, None),
                 ..chain_client::Options::test_default()
             },
         )