Problem: L2 network cannot filter transactions in v24.9.0 (#58)

core/bin/zksync_server/src/main.rs

@@ -44,7 +44,7 @@ struct Cli {
     /// Comma-separated list of components to launch.
     #[arg(
         long,
-        default_value = "api,tree,eth,state_keeper,housekeeper,tee_verifier_input_producer,commitment_generator,da_dispatcher"
+        default_value = "api,tree,eth,state_keeper,housekeeper,tee_verifier_input_producer,commitment_generator,da_dispatcher,deny_list"
     )]
     components: ComponentsToRun,
     /// Path to the yaml config. If set, it will be used instead of env vars.

core/bin/zksync_server/src/node_builder.rs

@@ -287,7 +287,7 @@ impl MainNodeBuilder {
         Ok(self)
     }
 
-    fn add_tx_sender_layer(mut self) -> anyhow::Result<Self> {
+    fn add_tx_sender_layer(mut self, with_denylist: bool) -> anyhow::Result<Self> {
         let sk_config = try_load_config!(self.configs.state_keeper_config);
         let rpc_config = try_load_config!(self.configs.api_config).web3_json_rpc;
         let postgres_storage_caches_config = PostgresStorageCachesConfig {
@@ -297,20 +297,31 @@ impl MainNodeBuilder {
         };
 
         // On main node we always use master pool sink.
-        self.node.add_layer(MasterPoolSinkLayer);
-        self.node.add_layer(TxSenderLayer::new(
-            TxSenderConfig::new(
-                &sk_config,
-                &rpc_config,
-                try_load_config!(self.wallets.state_keeper)
-                    .fee_account
-                    .address(),
-                self.genesis_config.l2_chain_id,
-            ),
-            postgres_storage_caches_config,
-            rpc_config.vm_concurrency_limit(),
-            ApiContracts::load_from_disk_blocking(), // TODO (BFT-138): Allow to dynamically reload API contracts
-        ));
+        let mut master_pool_sink_layer = MasterPoolSinkLayer::default();
+        if with_denylist {
+            let txsink_config = try_load_config!(self.configs.api_config).tx_sink;
+            master_pool_sink_layer = MasterPoolSinkLayer::deny_list(txsink_config.deny_list());
+            tracing::info!(
+                "Add MasterPoolSinkLayer with deny list: {:?}",
+                txsink_config.deny_list().unwrap_or_default()
+            );
+        }
+
+        self.node
+            .add_layer(master_pool_sink_layer)
+            .add_layer(TxSenderLayer::new(
+                TxSenderConfig::new(
+                    &sk_config,
+                    &rpc_config,
+                    try_load_config!(self.wallets.state_keeper)
+                        .fee_account
+                        .address(),
+                    self.genesis_config.l2_chain_id,
+                ),
+                postgres_storage_caches_config,
+                rpc_config.vm_concurrency_limit(),
+                ApiContracts::load_from_disk_blocking(), // TODO (BFT-138): Allow to dynamically reload API contracts
+            ));
         Ok(self)
     }
 
@@ -642,16 +653,22 @@ impl MainNodeBuilder {
                         .add_storage_initialization_layer(LayerKind::Task)?
                         .add_state_keeper_layer()?;
                 }
+                Component::TxSinkDenyList => {
+                    let with_denylist = true;
+                    self = self.add_tx_sender_layer(with_denylist)?;
+                }
                 Component::HttpApi => {
+                    let with_denylist = false;
                     self = self
-                        .add_tx_sender_layer()?
+                        .add_tx_sender_layer(with_denylist)?
                         .add_tree_api_client_layer()?
                         .add_api_caches_layer()?
                         .add_http_web3_api_layer()?;
                 }
                 Component::WsApi => {
+                    let with_denylist = false;
                     self = self
-                        .add_tx_sender_layer()?
+                        .add_tx_sender_layer(with_denylist)?
                         .add_tree_api_client_layer()?
                         .add_api_caches_layer()?
                         .add_ws_web3_api_layer()?;

core/lib/config/src/configs/api.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     fmt,
     net::SocketAddr,
     num::{NonZeroU32, NonZeroUsize},
@@ -24,6 +24,8 @@ pub struct ApiConfig {
     pub healthcheck: HealthCheckConfig,
     /// Configuration options for Merkle tree API.
     pub merkle_tree: MerkleTreeApiConfig,
+    /// Configuration options for the transactions sink.
+    pub tx_sink: TxSinkConfig,
 }
 
 /// Response size limits for specific RPC methods.
@@ -407,6 +409,21 @@ impl MerkleTreeApiConfig {
     }
 }
 
+#[derive(Debug, Deserialize, Clone, PartialEq)]
+pub struct TxSinkConfig {
+    pub deny_list: Option<String>,
+}
+
+impl TxSinkConfig {
+    pub fn deny_list(&self) -> Option<HashSet<Address>> {
+        self.deny_list.as_ref().map(|list| {
+            list.split(',')
+                .map(|element| Address::from_str(element).unwrap())
+                .collect()
+        })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

core/lib/config/src/testonly.rs

@@ -53,6 +53,7 @@ impl Distribution<configs::ApiConfig> for EncodeDist {
             prometheus: self.sample(rng),
             healthcheck: self.sample(rng),
             merkle_tree: self.sample(rng),
+            tx_sink: self.sample(rng),
         }
     }
 }
@@ -135,6 +136,14 @@ impl Distribution<configs::api::MerkleTreeApiConfig> for EncodeDist {
     }
 }
 
+impl Distribution<configs::api::TxSinkConfig> for EncodeDist {
+    fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> configs::api::TxSinkConfig {
+        configs::api::TxSinkConfig {
+            deny_list: self.sample(rng),
+        }
+    }
+}
+
 impl Distribution<configs::PrometheusConfig> for EncodeDist {
     fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> configs::PrometheusConfig {
         configs::PrometheusConfig {

core/lib/env_config/src/api.rs

@@ -1,7 +1,8 @@
 use anyhow::Context as _;
 use zksync_config::configs::{
     api::{
-        ContractVerificationApiConfig, HealthCheckConfig, MerkleTreeApiConfig, Web3JsonRpcConfig,
+        ContractVerificationApiConfig, HealthCheckConfig, MerkleTreeApiConfig, TxSinkConfig,
+        Web3JsonRpcConfig,
     },
     ApiConfig, PrometheusConfig,
 };
@@ -15,6 +16,7 @@ impl FromEnv for ApiConfig {
             prometheus: PrometheusConfig::from_env().context("PrometheusConfig")?,
             healthcheck: HealthCheckConfig::from_env().context("HealthCheckConfig")?,
             merkle_tree: MerkleTreeApiConfig::from_env().context("MerkleTreeApiConfig")?,
+            tx_sink: TxSinkConfig::from_env().context("TxSinkConfig")?,
         })
     }
 }
@@ -44,6 +46,13 @@ impl FromEnv for MerkleTreeApiConfig {
     }
 }
 
+impl FromEnv for TxSinkConfig {
+    /// Loads configuration from env variables.
+    fn from_env() -> anyhow::Result<Self> {
+        envy_load("tx_sink", "API_TX_SINK_")
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::num::{NonZeroU32, NonZeroUsize};
@@ -112,6 +121,9 @@ mod tests {
                 hard_time_limit_ms: Some(2_000),
             },
             merkle_tree: MerkleTreeApiConfig { port: 8082 },
+            tx_sink: TxSinkConfig {
+                deny_list: Some("0x1234567890abcdef".to_string()),
+            },
         }
     }
 
@@ -158,6 +170,7 @@ mod tests {
             API_HEALTHCHECK_SLOW_TIME_LIMIT_MS=250
             API_HEALTHCHECK_HARD_TIME_LIMIT_MS=2000
             API_MERKLE_TREE_PORT=8082
+            API_TX_SINK_DENY_LIST="0x1234567890abcdef"
         "#;
         lock.set_env(config);
 

core/lib/protobuf_config/src/api.rs

@@ -17,6 +17,7 @@ impl ProtoRepr for proto::Api {
             prometheus: read_required_repr(&self.prometheus).context("prometheus")?,
             healthcheck: read_required_repr(&self.healthcheck).context("healthcheck")?,
             merkle_tree: read_required_repr(&self.merkle_tree).context("merkle_tree")?,
+            tx_sink: read_required_repr(&self.tx_sink).context("tx_sink")?,
         })
     }
 
@@ -26,6 +27,7 @@ impl ProtoRepr for proto::Api {
             prometheus: Some(ProtoRepr::build(&this.prometheus)),
             healthcheck: Some(ProtoRepr::build(&this.healthcheck)),
             merkle_tree: Some(ProtoRepr::build(&this.merkle_tree)),
+            tx_sink: Some(ProtoRepr::build(&this.tx_sink)),
         }
     }
 }
@@ -271,3 +273,18 @@ impl ProtoRepr for proto::MerkleTreeApi {
         }
     }
 }
+
+impl ProtoRepr for proto::TxSink {
+    type Type = api::TxSinkConfig;
+    fn read(&self) -> anyhow::Result<Self::Type> {
+        Ok(Self::Type {
+            deny_list: self.deny_list.clone(),
+        })
+    }
+
+    fn build(this: &Self::Type) -> Self {
+        Self {
+            deny_list: this.deny_list.clone(),
+        }
+    }
+}

core/lib/protobuf_config/src/proto/config/api.proto

@@ -42,11 +42,10 @@ message Web3JsonRpc {
   repeated MaxResponseSizeOverride max_response_body_size_overrides = 31;
   repeated string api_namespaces = 32; // Optional, if empty all namespaces are available
   optional bool extended_api_tracing = 33; // optional, default false
-  reserved 15; reserved "l1_to_l2_transactions_compatibility_mode";
+  reserved 15;
+  reserved "l1_to_l2_transactions_compatibility_mode";
 }
 
-
-
 message HealthCheck {
   optional uint32 port = 1; // required; u16
   optional uint64 slow_time_limit_ms = 2; // optional; ms
@@ -57,9 +56,14 @@ message MerkleTreeApi {
   optional uint32 port = 1; // required; u16
 }
 
+message TxSink {
+  optional string deny_list = 1; // optional
+}
+
 message Api {
   optional Web3JsonRpc web3_json_rpc = 1; // required
   optional utils.Prometheus prometheus = 3; // required
   optional HealthCheck healthcheck = 4; // required
   optional MerkleTreeApi merkle_tree = 5; // required
+  optional TxSink tx_sink = 6; // optional
 }

core/lib/protobuf_config/src/proto/config/general.proto

@@ -2,25 +2,25 @@ syntax = "proto3";
 
 package zksync.config.general;
 
-import "zksync/config/prover.proto";
 import "zksync/config/api.proto";
+import "zksync/config/base_token_adjuster.proto";
 import "zksync/config/chain.proto";
+import "zksync/config/circuit_breaker.proto";
+import "zksync/config/commitment_generator.proto";
 import "zksync/config/contract_verifier.proto";
+import "zksync/config/da_dispatcher.proto";
 import "zksync/config/database.proto";
-import "zksync/config/circuit_breaker.proto";
 import "zksync/config/eth_sender.proto";
+import "zksync/config/external_price_api_client.proto";
 import "zksync/config/house_keeper.proto";
+import "zksync/config/object_store.proto";
 import "zksync/config/observability.proto";
+import "zksync/config/prover.proto";
+import "zksync/config/pruning.proto";
+import "zksync/config/snapshot_recovery.proto";
 import "zksync/config/snapshots_creator.proto";
 import "zksync/config/utils.proto";
-import "zksync/config/da_dispatcher.proto";
 import "zksync/config/vm_runner.proto";
-import "zksync/config/commitment_generator.proto";
-import "zksync/config/snapshot_recovery.proto";
-import "zksync/config/pruning.proto";
-import "zksync/config/object_store.proto";
-import "zksync/config/base_token_adjuster.proto";
-import "zksync/config/external_price_api_client.proto";
 
 message GeneralConfig {
   optional config.database.Postgres postgres = 1;

core/lib/protobuf_config/src/tests.rs

@@ -11,6 +11,7 @@ fn test_encoding() {
     test_encode_all_formats::<ReprConv<proto::api::Web3JsonRpc>>(rng);
     test_encode_all_formats::<ReprConv<proto::api::HealthCheck>>(rng);
     test_encode_all_formats::<ReprConv<proto::api::MerkleTreeApi>>(rng);
+    test_encode_all_formats::<ReprConv<proto::api::TxSink>>(rng);
     test_encode_all_formats::<ReprConv<proto::api::Api>>(rng);
     test_encode_all_formats::<ReprConv<proto::utils::Prometheus>>(rng);
     test_encode_all_formats::<ReprConv<proto::chain::StateKeeper>>(rng);

core/lib/zksync_core_leftovers/src/lib.rs

@@ -62,6 +62,8 @@ pub enum Component {
     BaseTokenRatioPersister,
     /// VM runner-based component that saves VM execution data for basic witness generation.
     VmRunnerBwip,
+    /// Component for filtering L2 transactions by denylist
+    TxSinkDenyList,
 }
 
 #[derive(Debug)]
@@ -106,6 +108,7 @@ impl FromStr for Components {
                 Ok(Components(vec![Component::BaseTokenRatioPersister]))
             }
             "vm_runner_bwip" => Ok(Components(vec![Component::VmRunnerBwip])),
+            "deny_list" => Ok(Components(vec![Component::TxSinkDenyList])),
             other => Err(format!("{} is not a valid component name", other)),
         }
     }

core/node/api_server/src/tx_sender/master_pool_sink.rs

@@ -1,4 +1,7 @@
-use std::collections::hash_map::{Entry, HashMap};
+use std::collections::{
+    hash_map::{Entry, HashMap},
+    HashSet,
+};
 
 use tokio::sync::Mutex;
 use zksync_dal::{transactions_dal::L2TxSubmissionResult, ConnectionPool, Core, CoreDal};
@@ -13,13 +16,15 @@ use crate::web3::metrics::API_METRICS;
 pub struct MasterPoolSink {
     master_pool: ConnectionPool<Core>,
     inflight_requests: Mutex<HashMap<(Address, Nonce), H256>>,
+    deny_list: HashSet<Address>,
 }
 
 impl MasterPoolSink {
-    pub fn new(master_pool: ConnectionPool<Core>) -> Self {
+    pub fn new(master_pool: ConnectionPool<Core>, deny_list: Option<HashSet<Address>>) -> Self {
         Self {
             master_pool,
             inflight_requests: Mutex::new(HashMap::new()),
+            deny_list: deny_list.unwrap_or_default(),
         }
     }
 }
@@ -33,6 +38,10 @@ impl TxSink for MasterPoolSink {
     ) -> Result<L2TxSubmissionResult, SubmitTxError> {
         let address_and_nonce = (tx.initiator_account(), tx.nonce());
 
+        if self.deny_list.contains(&address_and_nonce.0) {
+            return Err(SubmitTxError::SenderInDenyList(address_and_nonce.0));
+        }
+
         let mut lock = self.inflight_requests.lock().await;
         match lock.entry(address_and_nonce) {
             Entry::Occupied(entry) => {

core/node/api_server/src/tx_sender/mod.rs

@@ -63,7 +63,7 @@ pub async fn build_tx_sender(
     storage_caches: PostgresStorageCaches,
 ) -> anyhow::Result<(TxSender, VmConcurrencyBarrier)> {
     let sequencer_sealer = SequencerSealer::new(state_keeper_config.clone());
-    let master_pool_sink = MasterPoolSink::new(master_pool);
+    let master_pool_sink = MasterPoolSink::new(master_pool, None);
     let tx_sender_builder = TxSenderBuilder::new(
         tx_sender_config.clone(),
         replica_pool.clone(),

core/node/api_server/src/tx_sender/result.rs

@@ -1,6 +1,6 @@
 use thiserror::Error;
 use zksync_multivm::interface::{ExecutionResult, VmExecutionResultAndLogs};
-use zksync_types::{l2::error::TxCheckError, U256};
+use zksync_types::{l2::error::TxCheckError, Address, U256};
 use zksync_web3_decl::error::EnrichedClientError;
 
 use crate::execution_sandbox::{SandboxExecutionError, ValidationError};
@@ -75,6 +75,8 @@ pub enum SubmitTxError {
     /// Catch-all internal error (e.g., database error) that should not be exposed to the caller.
     #[error("internal error")]
     Internal(#[from] anyhow::Error),
+    #[error("sender address {0} is in deny list")]
+    SenderInDenyList(Address),
 }
 
 impl SubmitTxError {
@@ -108,6 +110,7 @@ impl SubmitTxError {
             Self::ProxyError(_) => "proxy-error",
             Self::FailedToPublishCompressedBytecodes => "failed-to-publish-compressed-bytecodes",
             Self::Internal(_) => "internal",
+            Self::SenderInDenyList(_) => "sender-in-deny-list",
         }
     }