feat(gRPC): add improved transaction filter (#9086)

muXxer · muXxer · commit 4b9863074007 · 2025-11-06T20:33:10.000+02:00
# Description of change

This PR adds a new transaction filter and a new event filter for gRPC
(not used yet by the API).
It combines the two existing json-rpc filters into one, but also adds
the possibility to filter for transactions that match a certain event
filter. Also, the possibility to chain the filters was added (`Any`,
`All`, `Or`, `And`) similar to the event filters.

## How the change has been tested

- [x] Basic tests (linting, compilation, formatting, unit/integration
tests)
- [ ] Patch-specific tests (correctness, functionality coverage)
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [ ] I have checked that new and existing unit tests pass locally with
my changes
diff --git a/crates/iota-core/src/event_filter.rs b/crates/iota-core/src/event_filter.rs
@@ -0,0 +1,107 @@
+// Copyright (c) Mysten Labs, Inc.
+// Modifications Copyright (c) 2025 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+use iota_json_rpc_types::{Filter, IotaEvent};
+use iota_metrics::monitored_scope;
+use iota_types::{
+    base_types::{IotaAddress, ObjectID},
+    error::IotaResult,
+};
+use move_core_types::{identifier::Identifier, language_storage::StructTag};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub enum GrpcEventFilter {
+    // Logical AND of several filters.
+    All(Vec<GrpcEventFilter>),
+    // Logical OR of several filters.
+    Any(Vec<GrpcEventFilter>),
+    // Logical NOT of a filter.
+    Not(Box<GrpcEventFilter>),
+
+    /// Filter by sender address.
+    Sender(IotaAddress),
+
+    /// Return events emitted in a specified Move package + module (optional).
+    /// If the event is defined in PackageA::ModuleA but emitted in a tx with
+    /// PackageB::ModuleB, filtering `MovePackageAndModule` by PackageB::ModuleB
+    /// returns the event. Filtering `MoveEventPackageAndModule` by
+    /// PackageA::ModuleA returns the event too.
+    MovePackageAndModule {
+        /// the Move package ID
+        package: ObjectID,
+        /// the module name (optional)
+        module: Option<Identifier>,
+    },
+    /// Return events with the given Move package + module (optional) where the
+    /// event struct is defined. If the event is defined in
+    /// PackageA::ModuleA but emitted in a tx with PackageB::ModuleB, filtering
+    /// `MoveEventPackageAndModule` by PackageA::ModuleA returns the
+    /// event. Filtering `MovePackageAndModule` by PackageB::ModuleB returns the
+    /// event too.
+    MoveEventPackageAndModule {
+        /// the Move package ID
+        package: ObjectID,
+        /// the module name (optional)
+        module: Option<Identifier>,
+    },
+    /// Return events with the given Move event struct name (struct tag).
+    /// For example, if the event is defined in `0xabcd::MyModule`, and named
+    /// `Foo`, then the struct tag is `0xabcd::MyModule::Foo`.
+    MoveEventType(StructTag),
+    /// Return events whose JSON representation contains the given field path
+    /// with the specified value (optional). The path should be a JSON pointer
+    /// as defined in RFC 6901.
+    MoveEventField {
+        path: String,
+        value: Option<Value>,
+    },
+}
+
+impl GrpcEventFilter {
+    fn try_matches(&self, item: &IotaEvent) -> IotaResult<bool> {
+        Ok(match self {
+            GrpcEventFilter::All(filters) => filters.iter().all(|f| f.matches(item)),
+            GrpcEventFilter::Any(filters) => filters.iter().any(|f| f.matches(item)),
+            GrpcEventFilter::Not(filter) => !filter.matches(item),
+
+            GrpcEventFilter::Sender(sender) => &item.sender == sender,
+
+            GrpcEventFilter::MovePackageAndModule { package, module } => {
+                &item.package_id == package
+                    && (module.is_none()
+                        || matches!(module,  Some(m2) if m2 == &item.transaction_module))
+            }
+            GrpcEventFilter::MoveEventPackageAndModule { package, module } => {
+                &ObjectID::from(item.type_.address) == package
+                    && (module.is_none() || matches!(module,  Some(m2) if m2 == &item.type_.module))
+            }
+            GrpcEventFilter::MoveEventType(event_type) => &item.type_ == event_type,
+            GrpcEventFilter::MoveEventField { path, value } => {
+                let json_ptr_value = item.parsed_json.pointer(path);
+                if value.is_none() {
+                    // If no value is specified, just check for the existence of the field.
+                    json_ptr_value.is_some()
+                } else {
+                    matches!(json_ptr_value, Some(v) if v == value.as_ref().unwrap())
+                }
+            }
+        })
+    }
+
+    pub fn and(self, other_filter: GrpcEventFilter) -> Self {
+        Self::All(vec![self, other_filter])
+    }
+    pub fn or(self, other_filter: GrpcEventFilter) -> Self {
+        Self::Any(vec![self, other_filter])
+    }
+}
+
+impl Filter<IotaEvent> for GrpcEventFilter {
+    fn matches(&self, item: &IotaEvent) -> bool {
+        let _scope = monitored_scope("GrpcEventFilter::matches");
+        self.try_matches(item).unwrap_or_default()
+    }
+}
diff --git a/crates/iota-core/src/lib.rs b/crates/iota-core/src/lib.rs
@@ -19,6 +19,7 @@ pub(crate) mod consensus_types;
 pub mod consensus_validator;
 pub mod db_checkpoint_handler;
 pub mod epoch;
+pub mod event_filter;
 pub mod execution_cache;
 mod execution_driver;
 mod fallback_fetch;
@@ -42,6 +43,7 @@ pub mod streamer;
 pub mod subscription_handler;
 pub mod test_utils;
 pub mod traffic_controller;
+pub mod transaction_filter;
 mod transaction_input_loader;
 mod transaction_manager;
 pub mod transaction_orchestrator;
diff --git a/crates/iota-core/src/subscription_handler.rs b/crates/iota-core/src/subscription_handler.rs
@@ -16,7 +16,10 @@ use prometheus::{
 use tokio_stream::Stream;
 use tracing::{error, instrument, trace};
 
-use crate::streamer::Streamer;
+use crate::{
+    streamer::Streamer,
+    transaction_filter::{GrpcTransactionFilter, TransactionDataWithEffectsAndEvents},
+};
 
 #[cfg(test)]
 #[path = "unit_tests/subscription_handler_tests.rs"]
@@ -70,14 +73,28 @@ pub struct SubscriptionHandler {
     event_streamer: Streamer<IotaEvent, IotaEvent, EventFilter>,
     transaction_streamer:
         Streamer<EffectsWithInput, IotaTransactionBlockEffects, TransactionFilter>,
+    grpc_transaction_streamer: Streamer<
+        TransactionDataWithEffectsAndEvents,
+        IotaTransactionBlockEffects,
+        GrpcTransactionFilter,
+    >,
 }
 
 impl SubscriptionHandler {
     pub fn new(registry: &Registry) -> Self {
         let metrics = Arc::new(SubscriptionMetrics::new(registry));
         Self {
             event_streamer: Streamer::spawn(EVENT_DISPATCH_BUFFER_SIZE, metrics.clone(), "event"),
-            transaction_streamer: Streamer::spawn(EVENT_DISPATCH_BUFFER_SIZE, metrics, "tx"),
+            transaction_streamer: Streamer::spawn(
+                EVENT_DISPATCH_BUFFER_SIZE,
+                metrics.clone(),
+                "tx",
+            ),
+            grpc_transaction_streamer: Streamer::spawn(
+                EVENT_DISPATCH_BUFFER_SIZE,
+                metrics,
+                "grpc_tx",
+            ),
         }
     }
 }
@@ -96,6 +113,17 @@ impl SubscriptionHandler {
             "Processing tx/event subscription"
         );
 
+        if let Err(e) =
+            self.grpc_transaction_streamer
+                .try_send(TransactionDataWithEffectsAndEvents {
+                    tx_data: input.clone(),
+                    effects: effects.clone(),
+                    events: events.clone(),
+                })
+        {
+            error!(error =? e, "Failed to send transaction to grpc dispatch");
+        }
+
         if let Err(e) = self.transaction_streamer.try_send(EffectsWithInput {
             input: input.clone(),
             effects: effects.clone(),
@@ -123,4 +151,11 @@ impl SubscriptionHandler {
     ) -> impl Stream<Item = IotaTransactionBlockEffects> {
         self.transaction_streamer.subscribe(filter)
     }
+
+    pub fn subscribe_grpc_transactions(
+        &self,
+        filter: GrpcTransactionFilter,
+    ) -> impl Stream<Item = IotaTransactionBlockEffects> {
+        self.grpc_transaction_streamer.subscribe(filter)
+    }
 }
diff --git a/crates/iota-core/src/transaction_filter.rs b/crates/iota-core/src/transaction_filter.rs
@@ -0,0 +1,146 @@
+// Copyright (c) Mysten Labs, Inc.
+// Modifications Copyright (c) 2025 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+use iota_json_rpc_types::{
+    Filter, IotaTransactionBlockEffects, IotaTransactionBlockEffectsAPI,
+    IotaTransactionBlockEvents, IotaTransactionKind, OwnedObjectRef,
+};
+use iota_metrics::monitored_scope;
+use iota_types::{
+    base_types::{IotaAddress, ObjectID},
+    object::Owner,
+    transaction::{TransactionData, TransactionDataAPI},
+};
+use serde::{Deserialize, Serialize};
+
+use crate::event_filter::GrpcEventFilter;
+
+#[derive(Clone)]
+pub struct TransactionDataWithEffectsAndEvents {
+    pub tx_data: TransactionData,
+    pub effects: IotaTransactionBlockEffects,
+    pub events: IotaTransactionBlockEvents,
+}
+
+impl From<TransactionDataWithEffectsAndEvents> for IotaTransactionBlockEffects {
+    fn from(e: TransactionDataWithEffectsAndEvents) -> Self {
+        e.effects
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub enum GrpcTransactionFilter {
+    // Logical AND of several filters.
+    All(Vec<GrpcTransactionFilter>),
+    // Logical OR of several filters.
+    Any(Vec<GrpcTransactionFilter>),
+    // Logical NOT of a filter.
+    Not(Box<GrpcTransactionFilter>),
+
+    /// Filter transactions of any given kind in the input.
+    TransactionKind(Vec<IotaTransactionKind>),
+
+    /// Filter by sender address.
+    Sender(IotaAddress),
+    /// Filter by recipient address. The recipient is determined by
+    /// checking the owners of mutated and unwrapped objects.
+    Receiver(IotaAddress),
+
+    /// Filter by input object.
+    InputObject(ObjectID),
+    /// Filter by changed object, including created, mutated and unwrapped
+    /// objects.
+    ChangedObject(ObjectID),
+    /// Filter transactions that wrapped or deleted the specified object.
+    /// Includes transactions that either created and immediately wrapped
+    /// the object or unwrapped and immediately deleted it.
+    /// TODO: @infra: do we need that now that we have the AffectedObject
+    /// filter?
+    WrappedOrDeletedObject(ObjectID),
+    /// Filter for transactions that touch this object.
+    AffectedObject(ObjectID),
+
+    /// Filter by move package, module (optional) and function (optional).
+    MoveCall {
+        /// the Move package ID
+        package: ObjectID,
+        /// the module name
+        module: Option<String>,
+        /// the function name
+        function: Option<String>,
+    },
+
+    /// Filter transactions that contain events matching the given event filter.
+    Events(GrpcEventFilter),
+}
+
+impl Filter<TransactionDataWithEffectsAndEvents> for GrpcTransactionFilter {
+    fn matches(&self, item: &TransactionDataWithEffectsAndEvents) -> bool {
+        let _scope = monitored_scope("GrpcTransactionFilter::matches");
+        match self {
+            GrpcTransactionFilter::All(filters) => filters.iter().all(|f| f.matches(item)),
+            GrpcTransactionFilter::Any(filters) => filters.iter().any(|f| f.matches(item)),
+            GrpcTransactionFilter::Not(filter) => !filter.matches(item),
+
+            GrpcTransactionFilter::TransactionKind(kinds) => kinds
+                .iter()
+                .any(|kind| kind == &IotaTransactionKind::from(item.tx_data.kind())),
+
+            GrpcTransactionFilter::Sender(a) => &item.tx_data.sender() == a,
+            GrpcTransactionFilter::Receiver(a) => {
+                let mutated: &[OwnedObjectRef] = item.effects.mutated();
+                mutated.iter().chain(item.effects.unwrapped().iter()).any(|oref: &OwnedObjectRef| {
+                    matches!(oref.owner, Owner::AddressOwner(owner) if owner == *a)
+                })
+            }
+
+            GrpcTransactionFilter::InputObject(o) => {
+                let Ok(input_objects) = item.tx_data.input_objects() else {
+                    return false;
+                };
+                input_objects.iter().any(|object| object.object_id() == *o)
+            }
+            GrpcTransactionFilter::ChangedObject(o) => item
+                .effects
+                .mutated()
+                .iter()
+                .any(|oref: &OwnedObjectRef| &oref.reference.object_id == o),
+            GrpcTransactionFilter::WrappedOrDeletedObject(o) => item
+                .effects
+                .wrapped()
+                .iter()
+                .chain(item.effects.deleted().iter())
+                .chain(item.effects.unwrapped_then_deleted().iter())
+                .any(|oref| &oref.object_id == o),
+            GrpcTransactionFilter::AffectedObject(o) => item
+                .effects
+                .created()
+                .iter()
+                .chain(item.effects.mutated().iter())
+                .chain(item.effects.unwrapped().iter())
+                .map(|oref: &OwnedObjectRef| &oref.reference)
+                .chain(item.effects.shared_objects().iter())
+                .chain(item.effects.deleted().iter())
+                .chain(item.effects.unwrapped_then_deleted().iter())
+                .chain(item.effects.wrapped().iter())
+                .any(|oref| &oref.object_id == o),
+
+            GrpcTransactionFilter::MoveCall {
+                package,
+                module,
+                function,
+            } => item.tx_data.move_calls().into_iter().any(|(p, m, f)| {
+                p == package
+                    && (module.is_none() || matches!(module,  Some(m2) if m2 == &m.to_string()))
+                    && (function.is_none() || matches!(function, Some(f2) if f2 == &f.to_string()))
+            }),
+
+            GrpcTransactionFilter::Events(event_filter) => item
+                .events
+                .data
+                .iter()
+                .any(|event| event_filter.matches(event)),
+        }
+    }
+}
