[Data Service] Implement simple upstream transaction filtering

There are two types of transaction filtering we will support in the future:
1. Per stream configuration: The downstream declares what txns they want to receive.
2. Global configuration: At the data service level we refuse to include full txns for all streams.

This PR implements the second of these, using @CapCap's work here:
aptos-labs/aptos-indexer-processors#398.

Rather than not sending txns at all if they match the blocklist filters, we just omit the writesets and events. Not sending the txns entirely would cause issues with processors, which today assume that they will receive all txns.

Cargo.toml

@@ -473,6 +473,7 @@ ark-groth16 = "0.4.0"
 ark-serialize = "0.4.0"
 ark-std = { version = "0.4.0", features = ["getrandom"] }
 aptos-moving-average = { git = "https://github.com/aptos-labs/aptos-indexer-processors.git", rev = "4801acae7aea30d7e96bbfbe5ec5b04056dfa4cf" }
+transaction-filter = { git = "https://github.com/aptos-labs/aptos-indexer-processors.git", rev = "35c362a10224ba0a22af8d03c515652c6bc62eb4" }
 assert_approx_eq = "1.1.0"
 assert_unordered = "0.3.5"
 async-channel = "1.7.1"

ecosystem/indexer-grpc/indexer-grpc-data-service/Cargo.toml

@@ -31,6 +31,7 @@ tokio = { workspace = true }
 tokio-stream = { workspace = true }
 tonic = { workspace = true }
 tonic-reflection = { workspace = true }
+transaction-filter = { workspace = true }
 tracing = { workspace = true }
 uuid = { workspace = true }
 

ecosystem/indexer-grpc/indexer-grpc-data-service/src/config.rs

@@ -16,6 +16,7 @@ use aptos_protos::{
 use serde::{Deserialize, Serialize};
 use std::{collections::HashSet, net::SocketAddr, sync::Arc};
 use tonic::{codec::CompressionEncoding, transport::Server};
+use transaction_filter::BooleanTransactionFilter;
 
 pub const SERVER_NAME: &str = "idxdatasvc";
 
@@ -69,9 +70,17 @@ pub struct IndexerGrpcDataServiceConfig {
     pub enable_cache_compression: bool,
     #[serde(default)]
     pub in_memory_cache_config: InMemoryCacheConfig,
-    /// Sender addresses to ignore. Transactions from these addresses will not be indexed.
-    #[serde(default = "IndexerGrpcDataServiceConfig::default_sender_addresses_to_ignore")]
-    pub sender_addresses_to_ignore: Vec<String>,
+    /// Any transaction that matches this filter will be stripped, meaning we remove
+    /// events and writesets from it before sending it downstream. This should only be
+    /// used in an emergency situation, e.g. when txns related to a certain module are
+    /// too large and are causing issues for the data service. Learn more here:
+    ///
+    /// https://www.notion.so/aptoslabs/Runbook-c006a37259394ac2ba904d6b54d180fa?pvs=4#171c210964ec42a89574fc80154f9e85
+    ///
+    /// Generally you will want to start with this with an OR, and then list out
+    /// separate filters that describe each type of txn we want to strip.
+    #[serde(default = "IndexerGrpcDataServiceConfig::default_txns_to_strip_filter")]
+    pub txns_to_strip_filter: BooleanTransactionFilter,
 }
 
 impl IndexerGrpcDataServiceConfig {
@@ -84,7 +93,7 @@ impl IndexerGrpcDataServiceConfig {
         redis_read_replica_address: RedisUrl,
         enable_cache_compression: bool,
         in_memory_cache_config: InMemoryCacheConfig,
-        sender_addresses_to_ignore: Vec<String>,
+        txns_to_strip_filter: BooleanTransactionFilter,
     ) -> Self {
         Self {
             data_service_grpc_tls_config,
@@ -97,7 +106,7 @@ impl IndexerGrpcDataServiceConfig {
             redis_read_replica_address,
             enable_cache_compression,
             in_memory_cache_config,
-            sender_addresses_to_ignore,
+            txns_to_strip_filter,
         }
     }
 
@@ -109,8 +118,9 @@ impl IndexerGrpcDataServiceConfig {
         false
     }
 
-    pub const fn default_sender_addresses_to_ignore() -> Vec<String> {
-        vec![]
+    pub const fn default_txns_to_strip_filter() -> BooleanTransactionFilter {
+        // This filter matches no txns.
+        BooleanTransactionFilter::new_or(vec![])
     }
 }
 
@@ -162,10 +172,7 @@ impl RunnableConfig for IndexerGrpcDataServiceConfig {
             self.redis_read_replica_address.clone(),
             self.file_store_config.clone(),
             self.data_service_response_channel_size,
-            self.sender_addresses_to_ignore
-                .clone()
-                .into_iter()
-                .collect::<HashSet<_>>(),
+            self.txns_to_strip_filter.clone(),
             cache_storage_format,
             Arc::new(in_memory_cache),
         )?;