Capnp encoding for queries (#75)

* Initial impl of capnp serialization of query

* Refactor net types into separate modules

* Add field types

* Add tests for using all fields in schema

* Update field types and tests to use strum for ordering

* Compiling with new field types

* Move capnp logic relevant modules

* Use serialize packed for capnp

* Wip add capnp deserializer

* Wip add deserialization for individual filters

* Refactor and ensure filters get set

* Add test and benchmark for default query

* Passing base query

* Add tests for block serde

* Add test for transaction serde

* Fix transaction serde

* Fix api test

* Add test for logs serde

* Add test for trace serde

* Add benchmark for large payload and test with bincode

* Add moderate payload test

* Fix description of config

* Bump versions

* Add display and from_str traits to fields

* Change default serialization

* Change serialized query structure

* Working api_test

* Add summary

* Add new test

* Add commands

* Add benchmarks to presentation

* Add future improvements

* Backmerg main with stubs

* Implement new capnp structures

* Remove breaking bincode benchmarks

* Bump versions

* Include capnp generated code in src so that there is no dependency on capnpc

* Small serialization speed improvment

* Remove hack presentation MD file

* Run cargo fmt on generated file

* Ignore capnp api test

* Ignore clippy warnings for generated code

* Use full name flag for capnp compile output

* Testing compression

* Compression tests

* Use zstd compression on query

* Fix warnings in benchmark

* Remove Readme deps on capnp compiler

* Fix typo

* Apply code rabbit make file suggestion

* Fix bugs in return type

* Implement code rabbit suggestions

Author: JonoPrest

README.md

@@ -2,32 +2,10 @@
 
 [![CI](https://github.com/enviodev/hypersync-client-rust/actions/workflows/ci.yaml/badge.svg?branch=main)](https://github.com/enviodev/hypersync-client-rust/actions/workflows/ci.yaml)
 <a href="https://crates.io/crates/hypersync-client">
-    <img src="https://img.shields.io/crates/v/hypersync-client.svg?style=flat-square"
+<img src="https://img.shields.io/crates/v/hypersync-client.svg?style=flat-square"
     alt="Crates.io version" />
 </a>
 
 Rust crate for [Envio's](https://envio.dev/) HyperSync client.
 
 [Documentation Page](https://docs.envio.dev/docs/hypersync-clients)
-
-### Dependencies
-
-Need to install capnproto tool in order to build the library.
-
-#### Linux
-
-```bash
-apt-get install -y capnproto libcapnp-dev
-```
-
-#### Windows
-
-```bash
-choco install capnproto
-```
-
-#### MacOS
-
-```bash
-brew install capnp
-```

hypersync-client/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hypersync-client"
-version = "0.20.0-rc.1"
+version = "0.20.0-rc.2"
 edition = "2021"
 description = "client library for hypersync"
 license = "MPL-2.0"
@@ -47,7 +47,7 @@ nohash-hasher = "0.2.0"
 ethers = { version = "2.0.14", optional = true }
 alloy-primitives = "1.1"
 
-hypersync-net-types = { path = "../hypersync-net-types", version = "0.11.0-rc.1" }
+hypersync-net-types = { path = "../hypersync-net-types", version = "0.11.0-rc.2" }
 hypersync-format = { path = "../hypersync-format", version = "0.5" }
 hypersync-schema = { path = "../hypersync-schema", version = "0.3" }
 

hypersync-client/src/config.rs

@@ -21,6 +21,25 @@ pub struct ClientConfig {
     pub retry_base_ms: Option<u64>,
     /// Ceiling time for request backoff.
     pub retry_ceiling_ms: Option<u64>,
+    /// Query serialization format to use for HTTP requests.
+    #[serde(default)]
+    pub serialization_format: SerializationFormat,
+}
+
+/// Determines query serialization format for HTTP requests.
+#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+pub enum SerializationFormat {
+    /// Use JSON serialization (default)
+    Json,
+    /// Use Cap'n Proto binary serialization
+    CapnProto,
+}
+
+impl Default for SerializationFormat {
+    fn default() -> Self {
+        // Keep this the default until all hs instances are upgraded to use Cap'n Proto endpoint
+        Self::Json
+    }
 }
 
 /// Config for hypersync event streaming.

hypersync-client/src/lib.rs

@@ -35,7 +35,7 @@ use url::Url;
 
 pub use column_mapping::{ColumnMapping, DataType};
 pub use config::HexOutput;
-pub use config::{ClientConfig, StreamConfig};
+pub use config::{ClientConfig, SerializationFormat, StreamConfig};
 pub use decode::Decoder;
 pub use decode_call::CallDecoder;
 pub use types::{ArrowBatch, ArrowResponse, ArrowResponseData, QueryResponse};
@@ -61,6 +61,8 @@ pub struct Client {
     retry_base_ms: u64,
     /// Ceiling time for request backoff.
     retry_ceiling_ms: u64,
+    /// Query serialization format to use for HTTP requests.
+    serialization_format: SerializationFormat,
 }
 
 impl Client {
@@ -86,6 +88,7 @@ impl Client {
             retry_backoff_ms: cfg.retry_backoff_ms.unwrap_or(500),
             retry_base_ms: cfg.retry_base_ms.unwrap_or(200),
             retry_ceiling_ms: cfg.retry_ceiling_ms.unwrap_or(5_000),
+            serialization_format: cfg.serialization_format,
         })
     }
 
@@ -383,8 +386,8 @@ impl Client {
         ))
     }
 
-    /// Executes query once and returns the result in (Arrow, size) format.
-    async fn get_arrow_impl(&self, query: &Query) -> Result<(ArrowResponse, u64)> {
+    /// Executes query once and returns the result in (Arrow, size) format using JSON serialization.
+    async fn get_arrow_impl_json(&self, query: &Query) -> Result<(ArrowResponse, u64)> {
         let mut url = self.url.clone();
         let mut segments = url.path_segments_mut().ok().context("get path segments")?;
         segments.push("query");
@@ -418,6 +421,56 @@ impl Client {
         Ok((res, bytes.len().try_into().unwrap()))
     }
 
+    /// Executes query once and returns the result in (Arrow, size) format using Cap'n Proto serialization.
+    async fn get_arrow_impl_capnp(&self, query: &Query) -> Result<(ArrowResponse, u64)> {
+        let mut url = self.url.clone();
+        let mut segments = url.path_segments_mut().ok().context("get path segments")?;
+        segments.push("query");
+        segments.push("arrow-ipc");
+        segments.push("capnp");
+        std::mem::drop(segments);
+        let mut req = self.http_client.request(Method::POST, url);
+
+        if let Some(bearer_token) = &self.bearer_token {
+            req = req.bearer_auth(bearer_token);
+        }
+
+        let query_bytes = query.to_bytes().context("serialize query to bytes")?;
+        let res = req
+            .header("content-type", "application/x-capnp")
+            .body(query_bytes)
+            .send()
+            .await
+            .context("execute http req")?;
+
+        let status = res.status();
+        if !status.is_success() {
+            let text = res.text().await.context("read text to see error")?;
+
+            return Err(anyhow!(
+                "http response status code {}, err body: {}",
+                status,
+                text
+            ));
+        }
+
+        let bytes = res.bytes().await.context("read response body bytes")?;
+
+        let res = tokio::task::block_in_place(|| {
+            parse_query_response(&bytes).context("parse query response")
+        })?;
+
+        Ok((res, bytes.len().try_into().unwrap()))
+    }
+
+    /// Executes query once and returns the result in (Arrow, size) format.
+    async fn get_arrow_impl(&self, query: &Query) -> Result<(ArrowResponse, u64)> {
+        match self.serialization_format {
+            SerializationFormat::Json => self.get_arrow_impl_json(query).await,
+            SerializationFormat::CapnProto => self.get_arrow_impl_capnp(query).await,
+        }
+    }
+
     /// Executes query with retries and returns the response in Arrow format.
     pub async fn get_arrow(&self, query: &Query) -> Result<ArrowResponse> {
         self.get_arrow_with_size(query).await.map(|res| res.0)

hypersync-client/src/preset_query.rs

@@ -3,6 +3,9 @@ use std::collections::BTreeSet;
 
 use arrayvec::ArrayVec;
 use hypersync_format::{Address, LogArgument};
+use hypersync_net_types::block::BlockField;
+use hypersync_net_types::log::LogField;
+use hypersync_net_types::transaction::TransactionField;
 use hypersync_net_types::{
     FieldSelection, LogFilter, LogSelection, Query, TransactionFilter, TransactionSelection,
 };
@@ -12,17 +15,8 @@ use hypersync_net_types::{
 /// Note: this is only for quickstart purposes.  For the best performance, create a custom query
 /// that only includes the fields you'll use in `field_selection`.
 pub fn blocks_and_transactions(from_block: u64, to_block: Option<u64>) -> Query {
-    let all_block_fields: BTreeSet<String> = hypersync_schema::block_header()
-        .fields
-        .iter()
-        .map(|x| x.name.clone())
-        .collect();
-
-    let all_tx_fields: BTreeSet<String> = hypersync_schema::transaction()
-        .fields
-        .iter()
-        .map(|x| x.name.clone())
-        .collect();
+    let all_block_fields = BlockField::all();
+    let all_tx_fields = TransactionField::all();
 
     Query {
         from_block,
@@ -45,15 +39,11 @@ pub fn blocks_and_transactions(from_block: u64, to_block: Option<u64>) -> Query
 /// that only includes the fields you'll use in `field_selection`.
 pub fn blocks_and_transaction_hashes(from_block: u64, to_block: Option<u64>) -> Query {
     let mut tx_field_selection = BTreeSet::new();
-    tx_field_selection.insert("block_hash".to_owned());
-    tx_field_selection.insert("block_number".to_owned());
-    tx_field_selection.insert("hash".to_owned());
+    tx_field_selection.insert(TransactionField::BlockHash);
+    tx_field_selection.insert(TransactionField::BlockNumber);
+    tx_field_selection.insert(TransactionField::Hash);
 
-    let all_block_fields: BTreeSet<String> = hypersync_schema::block_header()
-        .fields
-        .iter()
-        .map(|x| x.name.clone())
-        .collect();
+    let all_block_fields = BlockField::all();
 
     Query {
         from_block,
@@ -74,11 +64,7 @@ pub fn blocks_and_transaction_hashes(from_block: u64, to_block: Option<u64>) ->
 /// Note: this is only for quickstart purposes.  For the best performance, create a custom query
 /// that only includes the fields you'll use in `field_selection`.
 pub fn logs(from_block: u64, to_block: Option<u64>, contract_address: Address) -> Query {
-    let all_log_fields: BTreeSet<String> = hypersync_schema::log()
-        .fields
-        .iter()
-        .map(|x| x.name.clone())
-        .collect();
+    let all_log_fields = LogField::all();
 
     Query {
         from_block,
@@ -109,11 +95,7 @@ pub fn logs_of_event(
     let mut topics = ArrayVec::<Vec<LogArgument>, 4>::new();
     topics.insert(0, vec![topic0]);
 
-    let all_log_fields: BTreeSet<String> = hypersync_schema::log()
-        .fields
-        .iter()
-        .map(|x| x.name.clone())
-        .collect();
+    let all_log_fields = LogField::all();
 
     Query {
         from_block,
@@ -136,11 +118,7 @@ pub fn logs_of_event(
 /// Note: this is only for quickstart purposes.  For the best performance, create a custom query
 /// that only includes the fields you'll use in `field_selection`.
 pub fn transactions(from_block: u64, to_block: Option<u64>) -> Query {
-    let all_txn_fields: BTreeSet<String> = hypersync_schema::transaction()
-        .fields
-        .iter()
-        .map(|x| x.name.clone())
-        .collect();
+    let all_txn_fields = TransactionField::all();
 
     Query {
         from_block,
@@ -163,11 +141,7 @@ pub fn transactions_from_address(
     to_block: Option<u64>,
     address: Address,
 ) -> Query {
-    let all_txn_fields: BTreeSet<String> = hypersync_schema::transaction()
-        .fields
-        .iter()
-        .map(|x| x.name.clone())
-        .collect();
+    let all_txn_fields = TransactionField::all();
 
     Query {
         from_block,

hypersync-client/src/simple_types.rs

@@ -6,7 +6,9 @@ use hypersync_format::{
     AccessList, Address, Authorization, BlockNumber, BloomFilter, Data, Hash, LogArgument,
     LogIndex, Nonce, Quantity, TransactionIndex, TransactionStatus, TransactionType, Withdrawal,
 };
-use hypersync_net_types::FieldSelection;
+use hypersync_net_types::{
+    block::BlockField, log::LogField, transaction::TransactionField, FieldSelection,
+};
 use nohash_hasher::IntMap;
 use serde::{Deserialize, Serialize};
 use xxhash_rust::xxh3::Xxh3Builder;
@@ -26,10 +28,10 @@ pub struct Event {
 
 // Field lists for implementing event based API, these fields are used for joining
 // so they should always be added to the field selection.
-const BLOCK_JOIN_FIELD: &str = "number";
-const TX_JOIN_FIELD: &str = "hash";
-const LOG_JOIN_FIELD_WITH_TX: &str = "transaction_hash";
-const LOG_JOIN_FIELD_WITH_BLOCK: &str = "block_number";
+const BLOCK_JOIN_FIELD: BlockField = BlockField::Number;
+const TX_JOIN_FIELD: TransactionField = TransactionField::Hash;
+const LOG_JOIN_FIELD_WITH_TX: LogField = LogField::TransactionHash;
+const LOG_JOIN_FIELD_WITH_BLOCK: LogField = LogField::BlockNumber;
 
 enum InternalJoinStrategy {
     NotSelected,
@@ -51,15 +53,15 @@ impl From<&FieldSelection> for InternalEventJoinStrategy {
         Self {
             block: if block_fields_num == 0 {
                 InternalJoinStrategy::NotSelected
-            } else if block_fields_num == 1 && field_selection.block.contains(BLOCK_JOIN_FIELD) {
+            } else if block_fields_num == 1 && field_selection.block.contains(&BLOCK_JOIN_FIELD) {
                 InternalJoinStrategy::OnlyLogJoinField
             } else {
                 InternalJoinStrategy::FullJoin
             },
             transaction: if transaction_fields_num == 0 {
                 InternalJoinStrategy::NotSelected
             } else if transaction_fields_num == 1
-                && field_selection.transaction.contains(TX_JOIN_FIELD)
+                && field_selection.transaction.contains(&TX_JOIN_FIELD)
             {
                 InternalJoinStrategy::OnlyLogJoinField
             } else {
@@ -75,34 +77,24 @@ impl InternalEventJoinStrategy {
         match self.block {
             InternalJoinStrategy::NotSelected => (),
             InternalJoinStrategy::OnlyLogJoinField => {
-                field_selection
-                    .log
-                    .insert(LOG_JOIN_FIELD_WITH_BLOCK.to_string());
-                field_selection.block.remove(BLOCK_JOIN_FIELD);
+                field_selection.log.insert(LOG_JOIN_FIELD_WITH_BLOCK);
+                field_selection.block.remove(&BLOCK_JOIN_FIELD);
             }
             InternalJoinStrategy::FullJoin => {
-                field_selection
-                    .log
-                    .insert(LOG_JOIN_FIELD_WITH_BLOCK.to_string());
-                field_selection.block.insert(BLOCK_JOIN_FIELD.to_string());
+                field_selection.log.insert(LOG_JOIN_FIELD_WITH_BLOCK);
+                field_selection.block.insert(BLOCK_JOIN_FIELD);
             }
         }
 
         match self.transaction {
             InternalJoinStrategy::NotSelected => (),
             InternalJoinStrategy::OnlyLogJoinField => {
-                field_selection
-                    .log
-                    .insert(LOG_JOIN_FIELD_WITH_TX.to_string());
-                field_selection.transaction.remove(TX_JOIN_FIELD);
+                field_selection.log.insert(LOG_JOIN_FIELD_WITH_TX);
+                field_selection.transaction.remove(&TX_JOIN_FIELD);
             }
             InternalJoinStrategy::FullJoin => {
-                field_selection
-                    .log
-                    .insert(LOG_JOIN_FIELD_WITH_TX.to_string());
-                field_selection
-                    .transaction
-                    .insert(TX_JOIN_FIELD.to_string());
+                field_selection.log.insert(LOG_JOIN_FIELD_WITH_TX);
+                field_selection.transaction.insert(TX_JOIN_FIELD);
             }
         }
     }