feat(api): Monitor server RPC errors (matter-labs#1203)

## What ❔

- Introduces RPC-level metadata middleware.
- Uses this middleware to monitor server RPC errors and report stats as
metrics and logs.

## Why ❔

Increased API server observability.

## Checklist

- [x] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [x] Tests for the changes have been added / updated.
- [x] Documentation comments have been added / updated.
- [x] Code has been formatted via `zk fmt` and `zk lint`.
- [x] Spellcheck has been run via `zk spellcheck`.
- [x] Linkcheck has been run via `zk linkcheck`.

checks-config/era.dic

@@ -201,7 +201,7 @@ Serde
 tokenize
 EOAs
 zeroized
-value
+cardinality
 
 // zkSync-related words
 matterlabs

core/lib/web3_decl/Cargo.toml

@@ -10,19 +10,19 @@ keywords = ["blockchain", "zksync"]
 categories = ["cryptography"]
 
 [dependencies]
-itertools = "0.10.1"
+anyhow = "1.0"
 serde = "1.0"
-serde_json = "1.0"
 rlp = "0.5.0"
 thiserror = "1.0"
-bigdecimal = { version = "0.3.0", features = ["serde"] }
 jsonrpsee = { version = "0.21.0", default-features = false, features = [
     "macros",
 ] }
-chrono = "0.4"
 pin-project-lite = "0.2.13"
 zksync_types = { path = "../../lib/types" }
 
+[dev-dependencies]
+serde_json = "1.0"
+
 [features]
 default = ["server", "client"]
 server = ["jsonrpsee/server"]

core/lib/web3_decl/src/error.rs

@@ -16,6 +16,7 @@ use pin_project_lite::pin_project;
 use thiserror::Error;
 use zksync_types::{api::SerializationTransactionError, L1BatchNumber, MiniblockNumber};
 
+/// Server-side representation of the RPC error.
 #[derive(Debug, Error)]
 pub enum Web3Error {
     #[error("Block with such an ID doesn't exist yet")]
@@ -24,36 +25,27 @@ pub enum Web3Error {
     PrunedBlock(MiniblockNumber),
     #[error("L1 batch with such an ID is pruned; the first retained L1 batch is {0}")]
     PrunedL1Batch(L1BatchNumber),
-    #[error("Request timeout")]
-    RequestTimeout,
-    #[error("Internal error")]
-    InternalError,
-    #[error("RLP decoding error: {0}")]
-    RLPError(#[from] rlp::DecoderError),
-    #[error("No function with given signature found")]
-    NoSuchFunction,
-    #[error("Invalid transaction data: {0}")]
-    InvalidTransactionData(#[from] zksync_types::ethabi::Error),
+    #[error("{}", _0.as_ref())]
+    ProxyError(#[from] EnrichedClientError),
     #[error("{0}")]
     SubmitTransactionError(String, Vec<u8>),
     #[error("Failed to serialize transaction: {0}")]
     SerializationError(#[from] SerializationTransactionError),
-    #[error("Invalid fee parameters: {0}")]
-    InvalidFeeParams(String),
     #[error("More than four topics in filter")]
     TooManyTopics,
-    #[error("Your connection time exceeded the limit")]
-    PubSubTimeout,
     #[error("Filter not found")]
     FilterNotFound,
-    #[error("Not implemented")]
-    NotImplemented,
     #[error("Query returned more than {0} results. Try with this block range [{1:#x}, {2:#x}].")]
     LogsLimitExceeded(usize, u32, u32),
     #[error("invalid filter: if blockHash is supplied fromBlock and toBlock must not be")]
     InvalidFilterBlockHash,
+    #[error("Not implemented")]
+    NotImplemented,
+
     #[error("Tree API is not available")]
     TreeApiUnavailable,
+    #[error("Internal error")]
+    InternalError(#[from] anyhow::Error),
 }
 
 /// Client RPC error with additional details: the method name and arguments of the called method.

core/lib/web3_decl/src/types.rs

@@ -11,7 +11,6 @@ use core::{
     marker::PhantomData,
 };
 
-use itertools::unfold;
 use rlp::Rlp;
 use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
 pub use zksync_types::{
@@ -154,13 +153,15 @@ impl<'de, T: Deserialize<'de>> Deserialize<'de> for ValueOrArray<T> {
                     .map(|value| ValueOrArray(vec![value]))
             }
 
-            fn visit_seq<S>(self, visitor: S) -> Result<Self::Value, S::Error>
+            fn visit_seq<S>(self, mut visitor: S) -> Result<Self::Value, S::Error>
             where
                 S: de::SeqAccess<'de>,
             {
-                unfold(visitor, |vis| vis.next_element().transpose())
-                    .collect::<Result<_, _>>()
-                    .map(ValueOrArray)
+                let mut elements = Vec::with_capacity(visitor.size_hint().unwrap_or(1));
+                while let Some(element) = visitor.next_element()? {
+                    elements.push(element);
+                }
+                Ok(ValueOrArray(elements))
             }
         }
 

core/lib/zksync_core/Cargo.toml

@@ -64,11 +64,13 @@ rand = "0.8"
 
 tokio = { version = "1", features = ["time"] }
 futures = { version = "0.3", features = ["compat"] }
+pin-project-lite = "0.2.13"
 chrono = { version = "0.4", features = ["serde"] }
 anyhow = "1.0"
 thiserror = "1.0"
 async-trait = "0.1"
 bitflags = "1.3.2"
+thread_local = "1.1"
 
 reqwest = { version = "0.11", features = ["blocking", "json"] }
 hex = "0.4"

core/lib/zksync_core/src/api_server/tx_sender/proxy.rs

@@ -20,10 +20,7 @@ use zksync_web3_decl::{
 };
 
 use super::{tx_sink::TxSink, SubmitTxError};
-use crate::{
-    api_server::web3::backend_jsonrpsee::internal_error,
-    metrics::{TxStage, APP_METRICS},
-};
+use crate::metrics::{TxStage, APP_METRICS};
 
 #[derive(Debug, Clone, Default)]
 pub(crate) struct TxCache {
@@ -237,7 +234,6 @@ impl TxSink for TxProxy {
 
     async fn lookup_pending_nonce(
         &self,
-        _method_name: &'static str,
         account_address: Address,
         last_known_nonce: u32,
     ) -> Result<Option<Nonce>, Web3Error> {
@@ -251,30 +247,18 @@ impl TxSink for TxProxy {
         ))
     }
 
-    async fn lookup_tx(
-        &self,
-        method_name: &'static str,
-        id: TransactionId,
-    ) -> Result<Option<Transaction>, Web3Error> {
+    async fn lookup_tx(&self, id: TransactionId) -> Result<Option<Transaction>, Web3Error> {
         if let TransactionId::Hash(hash) = id {
             // If the transaction is not in the db, check the cache
             if let Some(tx) = self.find_tx(hash).await {
                 return Ok(Some(tx.into()));
             }
         }
         // If the transaction is not in the cache, query main node
-        self.request_tx(id)
-            .await
-            .map_err(|err| internal_error(method_name, err))
+        Ok(self.request_tx(id).await?)
     }
 
-    async fn lookup_tx_details(
-        &self,
-        method_name: &'static str,
-        hash: H256,
-    ) -> Result<Option<TransactionDetails>, Web3Error> {
-        self.request_tx_details(hash)
-            .await
-            .map_err(|err| internal_error(method_name, err))
+    async fn lookup_tx_details(&self, hash: H256) -> Result<Option<TransactionDetails>, Web3Error> {
+        Ok(self.request_tx_details(hash).await?)
     }
 }

core/lib/zksync_core/src/api_server/tx_sender/tx_sink.rs

@@ -34,7 +34,6 @@ pub trait TxSink: std::fmt::Debug + Send + Sync + 'static {
     /// By default, returns `Ok(None)`.
     async fn lookup_pending_nonce(
         &self,
-        _method_name: &'static str,
         _account_address: Address,
         _last_known_nonce: u32,
     ) -> Result<Option<Nonce>, Web3Error> {
@@ -43,19 +42,14 @@ pub trait TxSink: std::fmt::Debug + Send + Sync + 'static {
 
     /// Attempts to look up the transaction by its API ID in the sink-specific storage.
     /// By default, returns `Ok(None)`.
-    async fn lookup_tx(
-        &self,
-        _method_name: &'static str,
-        _id: TransactionId,
-    ) -> Result<Option<Transaction>, Web3Error> {
+    async fn lookup_tx(&self, _id: TransactionId) -> Result<Option<Transaction>, Web3Error> {
         Ok(None)
     }
 
     /// Attempts to look up the transaction details by its hash in the sink-specific storage.
     /// By default, returns `Ok(None)`.
     async fn lookup_tx_details(
         &self,
-        _method_name: &'static str,
         _hash: H256,
     ) -> Result<Option<TransactionDetails>, Web3Error> {
         Ok(None)