make db_connection build

core/bin/external_node/src/init.rs

@@ -3,7 +3,7 @@
 use anyhow::Context as _;
 use zksync_basic_types::{L1BatchNumber, L2ChainId};
 use zksync_core::sync_layer::genesis::perform_genesis_if_needed;
-use zksync_dal::ConnectionPool;
+use zksync_dal::{ConnectionPool, Server};
 use zksync_health_check::AppHealthCheck;
 use zksync_object_store::ObjectStoreFactory;
 use zksync_snapshots_applier::SnapshotsApplierConfig;
@@ -20,7 +20,7 @@ enum InitDecision {
 }
 
 pub(crate) async fn ensure_storage_initialized(
-    pool: &ConnectionPool,
+    pool: &ConnectionPool<Server>,
     main_node_client: &HttpClient,
     app_health: &AppHealthCheck,
     l2_chain_id: L2ChainId,

core/bin/external_node/src/main.rs

@@ -33,7 +33,7 @@ use zksync_core::{
         MainNodeClient, SyncState,
     },
 };
-use zksync_dal::{healthcheck::ConnectionPoolHealthCheck, ConnectionPool};
+use zksync_dal::{healthcheck::ConnectionPoolHealthCheck, ConnectionPool, Server};
 use zksync_health_check::{AppHealthCheck, HealthStatus, ReactiveHealthCheck};
 use zksync_state::PostgresStorageCaches;
 use zksync_storage::RocksDB;
@@ -59,7 +59,7 @@ async fn build_state_keeper(
     action_queue: ActionQueue,
     state_keeper_db_path: String,
     config: &ExternalNodeConfig,
-    connection_pool: ConnectionPool,
+    connection_pool: ConnectionPool<Server>,
     sync_state: SyncState,
     l2_erc20_bridge_addr: Address,
     miniblock_sealer_handle: MiniblockSealerHandle,
@@ -111,7 +111,7 @@ async fn build_state_keeper(
 
 async fn init_tasks(
     config: &ExternalNodeConfig,
-    connection_pool: ConnectionPool,
+    connection_pool: ConnectionPool<Server>,
     main_node_client: HttpClient,
     task_handles: &mut Vec<task::JoinHandle<anyhow::Result<()>>>,
     app_health: &AppHealthCheck,

core/bin/snapshots_creator/src/creator.rs

@@ -5,7 +5,7 @@ use std::sync::Arc;
 use anyhow::Context as _;
 use tokio::sync::Semaphore;
 use zksync_config::SnapshotsCreatorConfig;
-use zksync_dal::{BasicStorageProcessor, ConnectionPool};
+use zksync_dal::{BasicStorageProcessor, ConnectionPool, Server, ServerProcessor};
 use zksync_object_store::ObjectStore;
 use zksync_types::{
     snapshots::{
@@ -60,14 +60,14 @@ impl SnapshotProgress {
 #[derive(Debug)]
 pub(crate) struct SnapshotCreator {
     pub blob_store: Arc<dyn ObjectStore>,
-    pub master_pool: ConnectionPool,
-    pub replica_pool: ConnectionPool,
+    pub master_pool: ConnectionPool<Server>,
+    pub replica_pool: ConnectionPool<Server>,
     #[cfg(test)]
     pub event_listener: Box<dyn HandleEvent>,
 }
 
 impl SnapshotCreator {
-    async fn connect_to_replica(&self) -> anyhow::Result<BasicStorageProcessor<'_>> {
+    async fn connect_to_replica(&self) -> anyhow::Result<ServerProcessor<'_>> {
         self.replica_pool
             .access_storage_tagged("snapshots_creator")
             .await
@@ -192,7 +192,7 @@ impl SnapshotCreator {
         config: &SnapshotsCreatorConfig,
         min_chunk_count: u64,
         latest_snapshot: Option<&SnapshotMetadata>,
-        conn: &mut BasicStorageProcessor<'_>,
+        conn: &mut ServerProcessor<'_>,
     ) -> anyhow::Result<Option<SnapshotProgress>> {
         // We subtract 1 so that after restore, EN node has at least one L1 batch to fetch
         let sealed_l1_batch_number = conn.blocks_dal().get_sealed_l1_batch_number().await?;

core/lib/circuit_breaker/src/l1_txs.rs

@@ -1,10 +1,10 @@
-use zksync_dal::ConnectionPool;
+use zksync_dal::{ConnectionPool, Server};
 
 use crate::{CircuitBreaker, CircuitBreakerError};
 
 #[derive(Debug)]
 pub struct FailedL1TransactionChecker {
-    pub pool: ConnectionPool,
+    pub pool: ConnectionPool<Server>,
 }
 
 #[async_trait::async_trait]

core/lib/circuit_breaker/src/replication_lag.rs

@@ -1,10 +1,10 @@
-use zksync_dal::ConnectionPool;
+use zksync_dal::{ConnectionPool, Server};
 
 use crate::{CircuitBreaker, CircuitBreakerError};
 
 #[derive(Debug)]
 pub struct ReplicationLagChecker {
-    pub pool: ConnectionPool,
+    pub pool: ConnectionPool<Server>,
     pub replication_lag_limit_sec: Option<u32>,
 }
 

core/lib/dal/src/basic_witness_input_producer_dal.rs

@@ -1,7 +1,7 @@
 use std::time::{Duration, Instant};
 
 use sqlx::postgres::types::PgInterval;
-use zksync_db_connection::instrument::InstrumentExt;
+use zksync_db_connection::{instrument::InstrumentExt, processor::StorageProcessor};
 use zksync_types::L1BatchNumber;
 
 use crate::{

core/lib/dal/src/blocks_dal.rs

@@ -7,7 +7,9 @@ use std::{
 use anyhow::Context as _;
 use bigdecimal::{BigDecimal, FromPrimitive, ToPrimitive};
 use zksync_db_connection::{
-    instrument::InstrumentExt, match_query_as, processor::StorageProcessor,
+    instrument::InstrumentExt,
+    match_query_as,
+    processor::{BasicStorageProcessor, StorageProcessor},
 };
 use zksync_types::{
     aggregated_operations::AggregatedActionType,

core/lib/dal/src/blocks_web3_dal.rs

@@ -1,5 +1,7 @@
 use zksync_db_connection::{
-    instrument::InstrumentExt, match_query_as, processor::StorageProcessor,
+    instrument::InstrumentExt,
+    match_query_as,
+    processor::{BasicStorageProcessor, StorageProcessor},
 };
 use zksync_system_constants::EMPTY_UNCLES_HASH;
 use zksync_types::{

core/lib/dal/src/consensus_dal.rs

@@ -1,7 +1,7 @@
 use anyhow::Context as _;
 use zksync_consensus_roles::validator;
 use zksync_consensus_storage::ReplicaState;
-use zksync_db_connection::processor::StorageProcessor;
+use zksync_db_connection::processor::{BasicStorageProcessor, StorageProcessor};
 use zksync_types::MiniblockNumber;
 
 pub use crate::models::consensus::Payload;

core/lib/dal/src/contract_verification_dal.rs

@@ -5,7 +5,7 @@ use std::{
 
 use anyhow::Context as _;
 use sqlx::postgres::types::PgInterval;
-use zksync_db_connection::processor::StorageProcessor;
+use zksync_db_connection::processor::{BasicStorageProcessor, StorageProcessor};
 use zksync_types::{
     contract_verification_api::{
         DeployContractCalldata, VerificationIncomingRequest, VerificationInfo, VerificationRequest,

core/lib/dal/src/eth_sender_dal.rs

@@ -2,7 +2,10 @@ use std::{convert::TryFrom, str::FromStr};
 
 use anyhow::Context as _;
 use sqlx::types::chrono::{DateTime, Utc};
-use zksync_db_connection::{match_query_as, processor::StorageProcessor};
+use zksync_db_connection::{
+    match_query_as,
+    processor::{BasicStorageProcessor, StorageProcessor},
+};
 use zksync_types::{
     aggregated_operations::AggregatedActionType,
     eth_sender::{EthTx, EthTxBlobSidecar, TxHistory, TxHistoryToSend},

core/lib/dal/src/events_dal.rs

@@ -1,7 +1,10 @@
 use std::{collections::HashMap, fmt};
 
 use sqlx::types::chrono::Utc;
-use zksync_db_connection::processor::StorageProcessor;
+use zksync_db_connection::{
+    processor::{BasicStorageProcessor, StorageProcessor},
+    write_str, writeln_str,
+};
 use zksync_system_constants::L1_MESSENGER_ADDRESS;
 use zksync_types::{
     api,

core/lib/dal/src/events_web3_dal.rs

@@ -3,7 +3,10 @@ use sqlx::{
     query::{Query, QueryAs},
     Postgres, Row,
 };
-use zksync_db_connection::{instrument::InstrumentExt, processor::StorageProcessor};
+use zksync_db_connection::{
+    instrument::InstrumentExt,
+    processor::{BasicStorageProcessor, StorageProcessor},
+};
 use zksync_types::{
     api::{GetLogsFilter, Log},
     Address, MiniblockNumber, H256,

core/lib/dal/src/factory_deps_dal.rs

@@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet};
 
 use anyhow::Context as _;
 use zksync_contracts::{BaseSystemContracts, SystemContractCode};
-use zksync_db_connection::processor::StorageProcessor;
+use zksync_db_connection::processor::{BasicStorageProcessor, StorageProcessor};
 use zksync_types::{MiniblockNumber, H256, U256};
 use zksync_utils::{bytes_to_be_words, bytes_to_chunks};
 

core/lib/dal/src/lib.rs

@@ -55,15 +55,25 @@ mod tests;
 
 pub struct Server(());
 
+#[derive(Debug)]
 pub struct ServerProcessor<'a>(BasicStorageProcessor<'a>);
 
+impl<'a> From<BasicStorageProcessor<'a>> for ServerProcessor<'a> {
+    fn from(processor: BasicStorageProcessor<'a>) -> Self {
+        Self(processor)
+    }
+}
+
 impl StorageKind for Server {
     type Processor<'a> = ServerProcessor<'a>;
 }
 
+#[async_trait]
 impl<'a> StorageProcessor for ServerProcessor<'a> {
+    type Processor<'b> = ServerProcessor<'b> where Self: 'b;
+
     async fn start_transaction(&mut self) -> sqlx::Result<ServerProcessor<'_>> {
-        self.0.start_transaction()
+        self.0.start_transaction().await.map(ServerProcessor::from)
     }
 
     /// Checks if the `StorageProcessor` is currently within database transaction.
@@ -75,18 +85,6 @@ impl<'a> StorageProcessor for ServerProcessor<'a> {
         self.0.commit()
     }
 
-    fn from_pool(
-        connection: PoolConnection<Postgres>,
-        tags: Option<StorageProcessorTags>,
-        traced_connections: Option<&'a TracedConnections>,
-    ) -> Self {
-        Self(BasicStorageProcessor::from_pool(
-            connection,
-            tags,
-            traced_connections,
-        ))
-    }
-
     fn conn(&mut self) -> &mut PgConnection {
         self.0.conn_and_tags().0
     }

core/lib/dal/src/proof_generation_dal.rs

@@ -1,7 +1,7 @@
 use std::time::Duration;
 
 use strum::{Display, EnumString};
-use zksync_db_connection::processor::StorageProcessor;
+use zksync_db_connection::processor::{BasicStorageProcessor, StorageProcessor};
 use zksync_types::L1BatchNumber;
 
 use crate::{time_utils::pg_interval_from_duration, ServerProcessor, SqlxError};

core/lib/dal/src/protocol_versions_dal.rs

@@ -2,7 +2,7 @@ use std::convert::TryInto;
 
 use anyhow::Context as _;
 use zksync_contracts::{BaseSystemContracts, BaseSystemContractsHashes};
-use zksync_db_connection::processor::StorageProcessor;
+use zksync_db_connection::processor::{BasicStorageProcessor, StorageProcessor};
 use zksync_types::{
     protocol_version::{L1VerifierConfig, ProtocolUpgradeTx, ProtocolVersion, VerifierParams},
     Address, ProtocolVersionId, H256,

core/lib/dal/src/protocol_versions_web3_dal.rs

@@ -1,4 +1,4 @@
-use zksync_db_connection::processor::StorageProcessor;
+use zksync_db_connection::processor::{BasicStorageProcessor, StorageProcessor};
 use zksync_types::api::ProtocolVersion;
 
 use crate::{models::storage_protocol_version::StorageProtocolVersion, ServerProcessor};

core/lib/dal/src/snapshot_recovery_dal.rs

@@ -1,4 +1,4 @@
-use zksync_db_connection::processor::StorageProcessor;
+use zksync_db_connection::processor::{BasicStorageProcessor, StorageProcessor};
 use zksync_types::{
     snapshots::SnapshotRecoveryStatus, L1BatchNumber, MiniblockNumber, ProtocolVersionId, H256,
 };

core/lib/dal/src/snapshots_creator_dal.rs

@@ -4,7 +4,7 @@ use zksync_types::{
     StorageKey, H256,
 };
 
-use crate::{instrument::InstrumentExt, ServerProcessor};
+use crate::ServerProcessor;
 
 #[derive(Debug)]
 pub struct SnapshotsCreatorDal<'a, 'c> {
@@ -247,7 +247,7 @@ mod tests {
     }
 
     async fn assert_logs_for_snapshot(
-        conn: &mut StorageProcessor<'_>,
+        conn: &mut ServerProcessor<'_>,
         miniblock_number: MiniblockNumber,
         l1_batch_number: L1BatchNumber,
         expected_logs: &[StorageLog],

core/lib/dal/src/snapshots_dal.rs

@@ -1,4 +1,7 @@
-use zksync_db_connection::{instrument::InstrumentExt, processor::StorageProcessor};
+use zksync_db_connection::{
+    instrument::InstrumentExt,
+    processor::{BasicStorageProcessor, StorageProcessor},
+};
 use zksync_types::{
     snapshots::{AllSnapshots, SnapshotMetadata},
     L1BatchNumber,

core/lib/dal/src/storage_dal.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 
 use itertools::Itertools;
-use zksync_db_connection::processor::StorageProcessor;
+use zksync_db_connection::processor::{BasicStorageProcessor, StorageProcessor};
 use zksync_types::{StorageKey, StorageLog, StorageValue, H256};
 
 use crate::ServerProcessor;

core/lib/dal/src/storage_logs_dal.rs

@@ -1,7 +1,11 @@
 use std::{collections::HashMap, ops, time::Instant};
 
 use sqlx::{types::chrono::Utc, Row};
-use zksync_db_connection::{instrument::InstrumentExt, processor::StorageProcessor};
+use zksync_db_connection::{
+    instrument::InstrumentExt,
+    processor::{BasicStorageProcessor, StorageProcessor},
+    write_str, writeln_str,
+};
 use zksync_types::{
     get_code_key, snapshots::SnapshotStorageLog, AccountTreeId, Address, L1BatchNumber,
     MiniblockNumber, StorageKey, StorageLog, FAILED_CONTRACT_DEPLOYMENT_BYTECODE_HASH, H160, H256,
@@ -841,7 +845,11 @@ mod tests {
     use super::*;
     use crate::{tests::create_miniblock_header, ConnectionPool};
 
-    async fn insert_miniblock(conn: &mut StorageProcessor<'_>, number: u32, logs: Vec<StorageLog>) {
+    async fn insert_miniblock(
+        conn: &mut BasicStorageProcessor<'_>,
+        number: u32,
+        logs: Vec<StorageLog>,
+    ) {
         let header = L1BatchHeader::new(
             L1BatchNumber(number),
             0,
@@ -917,7 +925,7 @@ mod tests {
     }
 
     async fn test_rollback(
-        conn: &mut StorageProcessor<'_>,
+        conn: &mut BasicStorageProcessor<'_>,
         key: StorageKey,
         second_key: StorageKey,
     ) {
@@ -1145,7 +1153,7 @@ mod tests {
         }
     }
 
-    async fn prepare_tree_entries(conn: &mut StorageProcessor<'_>, count: u8) -> Vec<H256> {
+    async fn prepare_tree_entries(conn: &mut BasicStorageProcessor<'_>, count: u8) -> Vec<H256> {
         conn.protocol_versions_dal()
             .save_protocol_version_with_tx(ProtocolVersion::default())
             .await;

core/lib/dal/src/storage_logs_dedup_dal.rs

@@ -1,7 +1,7 @@
 use std::collections::HashSet;
 
 use sqlx::types::chrono::Utc;
-use zksync_db_connection::processor::StorageProcessor;
+use zksync_db_connection::processor::{BasicStorageProcessor, StorageProcessor};
 use zksync_types::{
     snapshots::SnapshotStorageLog, zk_evm_types::LogQuery, AccountTreeId, Address, L1BatchNumber,
     StorageKey, H256,

core/lib/dal/src/storage_web3_dal.rs

@@ -1,6 +1,9 @@
 use std::{collections::HashMap, ops};
 
-use zksync_db_connection::{instrument::InstrumentExt, processor::StorageProcessor};
+use zksync_db_connection::{
+    instrument::InstrumentExt,
+    processor::{BasicStorageProcessor, StorageProcessor},
+};
 use zksync_types::{
     get_code_key, get_nonce_key,
     utils::{decompose_full_nonce, storage_key_for_standard_token_balance},

core/lib/dal/src/system_dal.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 
 use sqlx::Row;
-use zksync_db_connection::instrument::InstrumentExt;
+use zksync_db_connection::{instrument::InstrumentExt, processor::StorageProcessor};
 
 #[derive(Debug)]
 pub(crate) struct TableSize {