RPC456 temporarely removing the json validation and instead adding a counter to see how many json failures do we have

Author: dmitriy-helius

Cargo.lock

@@ -17,7 +17,7 @@ cadence = "0.29.0"
 cadence-macros = "0.29.0"
 dashmap = "5.5.3"
 queues = "1.1.0"
-jsonrpsee = { version = "0.20.1", features = [
+jsonrpsee = { version = "0.22.5", features = [
     "server",
     "http-client",
     "macros",
@@ -38,4 +38,4 @@ yellowstone-grpc-geyser = { git = "https://github.com/helius-labs/yellowstone-gr
 rand = "0.8.5"
 futures = "0.3.24"
 figment = { version = "0.10.6", features = ["env", "test"] }
-tower = { version = "0.4.13", features = ["full"] }
+tower = { version = "0.4.13", features = ["full"] }

Cargo.toml

@@ -5,18 +5,21 @@ use cadence::{BufferedUdpMetricSink, QueuingMetricSink, StatsdClient};
 use cadence_macros::set_global_default;
 use figment::{providers::Env, Figment};
 use grpc_geyser::GrpcGeyserImpl;
-use jsonrpsee::server::{middleware::ProxyGetRequestLayer, ServerBuilder};
+use jsonrpsee::server::{RpcServiceBuilder, ServerBuilder};
+use jsonrpsee::server::middleware::http::ProxyGetRequestLayer;
 use priority_fee::PriorityFeeTracker;
 use rpc_server::AtlasPriorityFeeEstimator;
 use serde::Deserialize;
 use tracing::{error, info};
+
 mod errors;
 mod grpc_consumer;
 mod grpc_geyser;
 mod priority_fee;
 mod rpc_server;
 mod slot_cache;
 mod solana;
+mod temp_validator;
 
 #[derive(Debug, Deserialize, Clone)]
 struct EstimatorEnv {
@@ -50,7 +53,8 @@ async fn main() {
 
     let port = env.port.unwrap_or(4141);
     let server = ServerBuilder::default()
-        .set_middleware(
+        .set_rpc_middleware(RpcServiceBuilder::new().layer(temp_validator::RpcValidatorLayer::new()))
+        .set_http_middleware(
             tower::ServiceBuilder::new()
                 // Proxy `GET /health` requests to internal `health` method.
                 .layer(

src/main.rs

@@ -44,8 +44,8 @@ impl fmt::Debug for AtlasPriorityFeeEstimator {
 #[derive(Serialize, Deserialize, Clone, Debug, Default)]
 #[serde(
     rename_all(serialize = "camelCase", deserialize = "camelCase"),
-    deny_unknown_fields
 )]
+// TODO: DKH - add deny_unknown_fields
 pub struct GetPriorityFeeEstimateRequest {
     pub transaction: Option<String>,       // estimate fee for a txn
     pub account_keys: Option<Vec<String>>, // estimate fee for a list of accounts
@@ -55,8 +55,8 @@ pub struct GetPriorityFeeEstimateRequest {
 #[derive(Serialize, Deserialize, Clone, Debug, Default)]
 #[serde(
     rename_all(serialize = "camelCase", deserialize = "camelCase"),
-    deny_unknown_fields
 )]
+// TODO: DKH - add deny_unknown_fields
 pub struct GetPriorityFeeEstimateOptions {
     // controls input txn encoding
     pub transaction_encoding: Option<UiTransactionEncoding>,
@@ -399,7 +399,6 @@ mod tests {
     use cadence::{NopMetricSink, StatsdClient};
     use jsonrpsee::core::Cow;
     use jsonrpsee::core::__reexports::serde_json;
-    use jsonrpsee::core::__reexports::serde_json::value::RawValue;
     use jsonrpsee::types::{Id, Request, TwoPointZero};
     use solana_sdk::clock::Slot;
     use solana_sdk::pubkey::Pubkey;
@@ -517,18 +516,21 @@ mod tests {
         assert_eq!(resp.priority_fee_estimate, Some(10500.0));
     }
 
-    #[test]
+    // #[test]
+    // TODO: DKH - add the test back after we readd the validation
     fn test_parsing_wrong_fields() {
         for (param, error) in bad_params() {
-            let json_val = format!("{{\"jsonrpc\": \"2.0\",\"id\": \"1\", \"method\": \"getPriorityFeeEstimate\", \"params\": {param} }}");
-            let res = serde_json::from_str::<jsonrpsee::types::Request>(json_val.as_str());
+            let json_val = format!("{{\"jsonrpc\": \"2.0\",\"id\": \"1\", \"method\": \"getPriorityFeeEstimate\", \"params\": [{param}] }}");
+            let res = serde_json::from_str::<Request>(json_val.as_str());
             let res = res.unwrap();
             assert_request(&res, Id::Str(Cow::const_str("1")), "getPriorityFeeEstimate");
 
-            let params: serde_json::error::Result<GetPriorityFeeEstimateRequest> =
-                serde_json::from_str(res.params.map(RawValue::get).unwrap());
-            assert!(params.is_err());
-            assert_eq!(params.err().unwrap().to_string(), error, "testing {param}");
+            if let Some(val) = res.params
+            {
+                let params: Result<Vec<GetPriorityFeeEstimateRequest>, _> = serde_json::from_str(val.get());
+                assert!(params.is_err());
+                assert_eq!(params.err().unwrap().to_string(), error, "testing {param}");
+            }
         }
     }
 

src/rpc_server.rs

@@ -0,0 +1,80 @@
+use cadence_macros::statsd_count;
+use crate::priority_fee::PriorityLevel;
+use jsonrpsee::core::__reexports::serde_json;
+use jsonrpsee::server::middleware::rpc::RpcServiceT;
+use jsonrpsee::types::Request;
+use serde::{Deserialize, Serialize};
+use solana_transaction_status::UiTransactionEncoding;
+use tracing::debug;
+
+#[derive(Serialize, Deserialize, Clone, Debug, Default)]
+#[serde(
+    rename_all(serialize = "camelCase", deserialize = "camelCase"),
+    deny_unknown_fields
+)]
+struct GetPriorityFeeEstimateOptionsFake {
+    // controls input txn encoding
+    pub transaction_encoding: Option<UiTransactionEncoding>,
+    // controls custom priority fee level response
+    pub priority_level: Option<PriorityLevel>, // Default to MEDIUM
+    pub include_all_priority_fee_levels: Option<bool>, // Include all priority level estimates in the response
+    #[serde()]
+    pub lookback_slots: Option<u32>, // how many slots to look back on, default 50, min 1, max 300
+    pub include_vote: Option<bool>, // include vote txns in the estimate
+    // returns recommended fee, incompatible with custom controls. Currently the recommended fee is the median fee excluding vote txns
+    pub recommended: Option<bool>, // return the recommended fee (median fee excluding vote txns)
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, Default)]
+#[serde(
+    rename_all(serialize = "camelCase", deserialize = "camelCase"),
+    deny_unknown_fields
+)]
+struct GetPriorityFeeEstimateRequestFake {
+    transaction: Option<String>,       // estimate fee for a txn
+    account_keys: Option<Vec<String>>, // estimate fee for a list of accounts
+    options: Option<GetPriorityFeeEstimateOptionsFake>,
+}
+
+/// RPC logger layer.
+#[derive(Copy, Clone, Debug)]
+pub struct RpcValidatorLayer;
+
+impl RpcValidatorLayer {
+    /// Create a new logging layer.
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl<S> tower::Layer<S> for RpcValidatorLayer {
+    type Service = RpcValidator<S>;
+
+    fn layer(&self, service: S) -> Self::Service {
+        RpcValidator { service }
+    }
+}
+
+/// A middleware that logs each RPC call and response.
+#[derive(Debug)]
+pub struct RpcValidator<S> {
+    service: S,
+}
+
+impl<'a, S> RpcServiceT<'a> for RpcValidator<S>
+where
+    S: RpcServiceT<'a> + Send + Sync,
+{
+    type Future = S::Future;
+
+    fn call(&self, req: Request<'a>) -> Self::Future {
+        if let Some(params) = &req.params {
+            if let Err(err_val) = serde_json::from_str::<Vec<GetPriorityFeeEstimateRequestFake>>(params.get()) {
+                statsd_count!("rpc_payload_parse_failed", 1);
+                debug!("RPC parse error: {}, {}", err_val, params);
+            }
+        }
+
+        self.service.call(req)
+    }
+}