Boolean Logic Transaction Filtering

rust/Cargo.toml

@@ -1,7 +1,7 @@
 [workspace]
 resolver = "2"
 
-members = ["indexer-metrics", "moving-average", "processor", "server-framework"]
+members = ["indexer-metrics", "moving-average", "processor", "server-framework", "transaction-filter"]
 
 [workspace.package]
 authors = ["Aptos Labs <[email protected]>"]
@@ -67,6 +67,8 @@ jemallocator = { version = "0.5.0", features = [
 ] }
 kanal = { version = "0.1.0-pre8", features = ["async"] }
 once_cell = "1.10.0"
+# SIMD for string search
+memchr = "2.7.2"
 num_cpus = "1.16.0"
 pbjson = "0.5.1"
 prometheus = { version = "0.13.0", default-features = false }
@@ -86,6 +88,7 @@ sha2 = "0.9.3"
 sha3 = "0.9.1"
 strum = { version = "0.24.1", features = ["derive"] }
 tempfile = "3.3.0"
+thiserror = "1.0.61"
 toml = "0.7.4"
 tracing-subscriber = { version = "0.3.17", features = ["json", "env-filter"] }
 tokio = { version = "1.35.1", features = ["full"] }

rust/processor/src/utils/util.rs

@@ -53,8 +53,11 @@ pub struct MultisigPayloadClean {
 }
 
 /// Standardizes all addresses and table handles to be length 66 (0x-64 length hash)
+#[inline]
 pub fn standardize_address(handle: &str) -> String {
-    if let Some(handle) = handle.strip_prefix("0x") {
+    if handle.len() == 66 {
+        handle.to_string()
+    } else if let Some(handle) = handle.strip_prefix("0x") {
         format!("0x{:0>64}", handle)
     } else {
         format!("0x{:0>64}", handle)

rust/transaction-filter/Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "transaction-filter"
+version = "0.1.0"
+
+# Workspace inherited keys
+authors = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+repository = { workspace = true }
+rust-version = { workspace = true }
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+anyhow = { workspace = true }
+aptos-protos = { workspace = true }
+# SIMD for string search. TODO: benchmark this on various real inputs to see if it's worth it
+memchr = { workspace = true }
+
+prost = { workspace = true }
+
+serde = { workspace = true }
+serde_json = { workspace = true }
+
+thiserror = { workspace = true }
+
+[dev-dependencies]
+# we only decompress the fixture protos in test
+lz4 = "1.24.0"
+

rust/transaction-filter/src/errors.rs

@@ -0,0 +1,81 @@
+use serde::{Serialize, Serializer};
+use std::fmt::Display;
+use thiserror::Error as ThisError;
+
+#[derive(Debug, Serialize)]
+pub struct FilterStepTrace {
+    pub serialized_filter: String,
+    pub filter_type: String,
+}
+
+impl Display for FilterStepTrace {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}:   {}", self.filter_type, self.serialized_filter)
+    }
+}
+#[derive(Debug)]
+pub struct SerializableError {
+    pub inner: Box<dyn std::error::Error + Send + Sync>,
+}
+
+/// Custom error that allows for keeping track of the filter type/path that caused the error
+#[derive(Debug, Serialize, ThisError)]
+pub struct FilterError {
+    pub filter_path: Vec<FilterStepTrace>,
+    pub error: SerializableError,
+}
+
+impl FilterError {
+    pub fn new(error: Box<dyn std::error::Error + Send + Sync>) -> Self {
+        Self {
+            filter_path: Vec::new(),
+            error: SerializableError::new(error),
+        }
+    }
+
+    pub fn add_trace(&mut self, serialized_filter: String, filter_type: String) {
+        self.filter_path.push(FilterStepTrace {
+            serialized_filter,
+            filter_type,
+        });
+    }
+}
+
+impl From<anyhow::Error> for FilterError {
+    fn from(error: anyhow::Error) -> Self {
+        Self::new(error.into())
+    }
+}
+
+impl Display for FilterError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let trace_path = self
+            .filter_path
+            .iter()
+            .map(|trace| format!("{}", trace))
+            .collect::<Vec<String>>()
+            .join("\n");
+        write!(
+            f,
+            "Filter Error: {:?}\nTrace Path:\n{}",
+            self.error.inner, trace_path
+        )
+    }
+}
+
+impl SerializableError {
+    fn new(error: Box<dyn std::error::Error + Send + Sync>) -> Self {
+        SerializableError { inner: error }
+    }
+}
+
+// Implement Serialize for the wrapper
+impl Serialize for SerializableError {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        // Serialize the error as its string representation
+        serializer.serialize_str(&self.inner.to_string())
+    }
+}