diff --git a/Cargo.lock b/Cargo.lock index f05a302..b7a2fb9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -388,6 +388,17 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "async-channel" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81953c529336010edd6d8e358f886d9581267795c61b19475b71314bffa46d35" +dependencies = [ + "concurrent-queue", + "event-listener", + "futures-core", +] + [[package]] name = "async-compression" version = "0.4.5" @@ -406,6 +417,28 @@ dependencies = [ "zstd-safe", ] +[[package]] +name = "async-stream" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd56dd203fef61ac097dd65721a419ddccb106b2d2b70ba60a6b529f03961a51" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16e62a023e7c117e27523144c5d2459f4397fcc3cab0085af8e2224f643a0193" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.60", +] + [[package]] name = "async-trait" version = "0.1.74" @@ -432,6 +465,51 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa" +[[package]] +name = "axum" +version = "0.6.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b829e4e32b91e643de6eafe82b1d90675f5874230191a4ffbc1b336dec4d6bf" +dependencies = [ + "async-trait", + "axum-core", + "bitflags 1.3.2", + "bytes", + "futures-util", + "http", + "http-body", + "hyper", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "rustversion", + "serde", + "sync_wrapper", + "tower", + "tower-layer", + "tower-service", +] + +[[package]] +name = "axum-core" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "759fa577a247914fd3f7f76d62972792636412fbfd634cd452f6a385a74d2d2c" +dependencies = [ + "async-trait", + "bytes", + "futures-util", + "http", + "http-body", + "mime", + "rustversion", + "tower-layer", + "tower-service", +] + [[package]] name = "backoff" version = "0.4.0" @@ -473,6 +551,12 @@ version = "0.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ba43ea6f343b788c8764558649e08df62f86c6ef251fdaeb1ffd010a9ae50a2" +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + [[package]] name = "bigdecimal" version = "0.4.2" @@ -743,6 +827,15 @@ dependencies = [ "unicode-width", ] +[[package]] +name = "concurrent-queue" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "const-random" version = "0.1.17" @@ -1110,7 +1203,7 @@ checksum = "c2e66c9d817f1720209181c316d28635c050fa304f9c79e47a520882661b7308" [[package]] name = "datafast-runtime" -version = "0.2.0" +version = "0.2.1" dependencies = [ "apollo-parser", "async-trait", @@ -1122,9 +1215,11 @@ dependencies = [ "ethabi", "figment", "futures-util", + "google-cloud-pubsub", "hex", "kanal", "log", + "lz4", "mongodb", "num-bigint", "num-traits", @@ -1692,6 +1787,12 @@ dependencies = [ "uint", ] +[[package]] +name = "event-listener" +version = "2.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" + [[package]] name = "fallible-iterator" version = "0.2.0" @@ -1941,8 +2042,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be4136b2a15dd319360be1c07d9933517ccf0be8f16bf62a3bee4f0d618df427" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -1968,6 +2071,94 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b" +[[package]] +name = "google-cloud-auth" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe54dd8b6eb2bcd5390998238bcc39d1daed4dbb70df2845832532540384fc41" +dependencies = [ + "async-trait", + "base64 0.21.4", + "google-cloud-metadata", + "google-cloud-token", + "home", + "jsonwebtoken", + "reqwest", + "serde", + "serde_json", + "thiserror", + "time", + "tokio", + "tracing", + "urlencoding", +] + +[[package]] +name = "google-cloud-gax" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8cb60314136e37de9e2a05ddb427b9c5a39c3d188de2e2f026c6af74425eef44" +dependencies = [ + "google-cloud-token", + "http", + "thiserror", + "tokio", + "tokio-retry", + "tonic", + "tower", + "tracing", +] + +[[package]] +name = "google-cloud-googleapis" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db8a478015d079296167e3f08e096dc99cffc2cb50fa203dd38aaa9dd37f8354" +dependencies = [ + "prost", + "prost-types", + "tonic", +] + +[[package]] +name = "google-cloud-metadata" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc279bfb50487d7bcd900e8688406475fc750fe474a835b2ab9ade9eb1fc90e2" +dependencies = [ + "reqwest", + "thiserror", + "tokio", +] + +[[package]] +name = "google-cloud-pubsub" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b2184a5c70b994e6d77eb1c140e193e7f5fe6015e9115322fac24f7e33f003c" +dependencies = [ + "async-channel", + "async-stream", + "google-cloud-auth", + "google-cloud-gax", + "google-cloud-googleapis", + "google-cloud-token", + "prost-types", + "thiserror", + "tokio", + "tokio-util", + "tracing", +] + +[[package]] +name = "google-cloud-token" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f49c12ba8b21d128a2ce8585955246977fbce4415f680ebf9199b6f9d6d725f" +dependencies = [ + "async-trait", +] + [[package]] name = "h2" version = "0.3.21" @@ -2093,6 +2284,15 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "home" +version = "0.5.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3d1354bf6b7235cb4a0576c2619fd4ed18183f689b12b006a0ee7329eeff9a5" +dependencies = [ + "windows-sys 0.52.0", +] + [[package]] name = "hostname" version = "0.3.1" @@ -2199,6 +2399,18 @@ dependencies = [ "tokio-rustls 0.24.1", ] +[[package]] +name = "hyper-timeout" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbb958482e8c7be4bc3cf272a766a2b0bf1a6755e7a6ae777f017a31d11b13b1" +dependencies = [ + "hyper", + "pin-project-lite", + "tokio", + "tokio-io-timeout", +] + [[package]] name = "hyper-tls" version = "0.5.0" @@ -2416,6 +2628,21 @@ dependencies = [ "serde_json", ] +[[package]] +name = "jsonwebtoken" +version = "9.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ae10193d25051e74945f1ea2d0b42e03cc3b890f7e4cc5faa44997d808193f" +dependencies = [ + "base64 0.21.4", + "js-sys", + "pem", + "ring 0.17.5", + "serde", + "serde_json", + "simple_asn1", +] + [[package]] name = "kanal" version = "0.1.0-pre8" @@ -2571,6 +2798,26 @@ dependencies = [ "linked-hash-map", ] +[[package]] +name = "lz4" +version = "1.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e9e2dd86df36ce760a60f6ff6ad526f7ba1f14ba0356f8254fb6905e6494df1" +dependencies = [ + "libc", + "lz4-sys", +] + +[[package]] +name = "lz4-sys" +version = "1.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57d27b317e207b10f69f5e75494119e391a96f48861ae870d1da6edac98ca900" +dependencies = [ + "cc", + "libc", +] + [[package]] name = "lz4_flex" version = "0.11.1" @@ -2618,6 +2865,12 @@ version = "0.1.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2532096657941c2fea9c289d370a250971c689d4f143798ff67113ec042024a5" +[[package]] +name = "matchit" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" + [[package]] name = "md-5" version = "0.9.1" @@ -3139,6 +3392,16 @@ dependencies = [ "syn 2.0.60", ] +[[package]] +name = "pem" +version = "3.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e459365e590736a54c3fa561947c84837534b8e9af6fc5bf781307e82658fae" +dependencies = [ + "base64 0.22.1", + "serde", +] + [[package]] name = "percent-encoding" version = "2.3.0" @@ -3350,6 +3613,15 @@ dependencies = [ "syn 2.0.60", ] +[[package]] +name = "prost-types" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "193898f59edcf43c26227dcd4c8427f00d99d61e95dcde58dabd49fa291d470e" +dependencies = [ + "prost", +] + [[package]] name = "protobuf" version = "2.28.0" @@ -3711,9 +3983,9 @@ dependencies = [ [[package]] name = "rstest" -version = "0.18.2" +version = "0.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97eeab2f3c0a199bc4be135c36c924b6590b88c377d416494288c14f2db30199" +checksum = "9d5316d2a1479eeef1ea21e7f9ddc67c191d497abc8fc3ba2467857abbb68330" dependencies = [ "futures", "futures-timer", @@ -3723,9 +3995,9 @@ dependencies = [ [[package]] name = "rstest_macros" -version = "0.18.2" +version = "0.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d428f8247852f894ee1be110b375111b586d4fa431f6c46e64ba5a0dcccbe605" +checksum = "04a9df72cc1f67020b0d63ad9bfe4a323e459ea7eb68e03bd9824db49f9a4c25" dependencies = [ "cfg-if", "glob", @@ -4378,6 +4650,18 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f27f6278552951f1f2b8cf9da965d10969b2efdea95a6ec47987ab46edfe263a" +[[package]] +name = "simple_asn1" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adc4e5204eb1910f40f9cfa375f6f05b68c3abac4b6fd879c8ff5e7ae8a0a085" +dependencies = [ + "num-bigint", + "num-traits", + "thiserror", + "time", +] + [[package]] name = "siphasher" version = "0.3.11" @@ -4599,6 +4883,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2047c6ded9c721764247e62cd3b03c09ffc529b2ba5b10ec482ae507a4a70160" + [[package]] name = "system-configuration" version = "0.5.1" @@ -4760,6 +5050,16 @@ dependencies = [ "windows-sys 0.48.0", ] +[[package]] +name = "tokio-io-timeout" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30b74022ada614a1b4834de765f9bb43877f910cc8ce4be40e89042c9223a8bf" +dependencies = [ + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-macros" version = "2.1.0" @@ -4896,6 +5196,64 @@ dependencies = [ "winnow", ] +[[package]] +name = "tonic" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d560933a0de61cf715926b9cac824d4c883c2c43142f787595e48280c40a1d0e" +dependencies = [ + "async-stream", + "async-trait", + "axum", + "base64 0.21.4", + "bytes", + "flate2", + "h2", + "http", + "http-body", + "hyper", + "hyper-timeout", + "percent-encoding", + "pin-project", + "prost", + "rustls 0.21.8", + "rustls-pemfile", + "tokio", + "tokio-rustls 0.24.1", + "tokio-stream", + "tower", + "tower-layer", + "tower-service", + "tracing", + "webpki-roots", +] + +[[package]] +name = "tower" +version = "0.4.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c" +dependencies = [ + "futures-core", + "futures-util", + "indexmap 1.9.3", + "pin-project", + "pin-project-lite", + "rand", + "slab", + "tokio", + "tokio-util", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-layer" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c20c8dbed6283a09604c3e69b4b7eeb54e298b8a600d4d5ecb5ad39de609f1d0" + [[package]] name = "tower-service" version = "0.3.2" @@ -5123,6 +5481,12 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "urlencoding" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" + [[package]] name = "utf-8" version = "0.7.6" diff --git a/Cargo.toml b/Cargo.toml index e3f3e65..324f4df 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,9 +1,13 @@ [package] name = "datafast-runtime" -version = "0.2.0" +version = "0.2.1" edition = "2021" +homepage = "https://runtime.datafast.network/" +repository = "https://github.com/datafast-network/datafast-runtime" +description = "DataFast Runtime" +license = "Apache-2.0" +keywords = ["indexer", "subgraph", "blockchain", "ethereum", "subgraph-runtime"] -# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] async-trait = "0.1.74" @@ -26,24 +30,29 @@ thiserror = "1.0.49" tokio = { version = "1.33.0", features = ["full"] } web3 = "0.19.0" tiny-keccak = "2.0.2" -scylla = { version = "0.12.0", optional = true } -mongodb = { version = "2.7.1", optional = true} apollo-parser = "0.7.3" prometheus = "0.13.3" warp = "0.3.6" tokio-retry = "0.3.0" -deltalake = { version = "0.17.1", features = ["s3", "datafusion"], optional = true } rayon = "1.8.0" prost = "0.12.3" wasmer = "4.2.4" regex = "1.10.2" +deltalake = { version = "0.17.1", features = ["s3", "datafusion"], optional = true } +scylla = { version = "0.12.0", optional = true } +mongodb = { version = "2.7.1", optional = true } +google-cloud-pubsub = { version = "0.24.0", optional = true } +lz4 = { version = "1.24.0", optional = true } + [features] default = ["mongo", "deltalake"] scylla = ["dep:scylla"] mongo = ["dep:mongodb"] deltalake = ["dep:deltalake"] +pubsub = ["dep:google-cloud-pubsub", "dep:lz4"] +mongsub = ["mongo", "pubsub"] [dev-dependencies] convert_case = "0.6.0" -rstest = "0.18.2" +rstest = "0.19.0" diff --git a/src/components/block_source/mod.rs b/src/components/block_source/mod.rs index 547fa77..4c93ae7 100644 --- a/src/components/block_source/mod.rs +++ b/src/components/block_source/mod.rs @@ -1,4 +1,5 @@ mod metrics; + use super::Valve; use crate::common::BlockDataMessage; use crate::common::Chain; @@ -15,9 +16,16 @@ use delta::DeltaClient; #[cfg(feature = "deltalake")] use delta::DeltaEthereumBlocks; +#[cfg(feature = "pubsub")] +mod pubsub; +#[cfg(feature = "pubsub")] +use pubsub::PubSubSource; + enum Source { #[cfg(feature = "deltalake")] Delta(DeltaClient), + #[cfg(feature = "pubsub")] + PubSub(PubSubSource), } pub struct BlockSource { @@ -36,6 +44,11 @@ impl BlockSource { SourceTypes::Delta(delta_cfg) => { Source::Delta(DeltaClient::new(delta_cfg.to_owned(), start_block, registry).await?) } + #[cfg(feature = "pubsub")] + SourceTypes::PubSub { + sub_id, + compression, + } => Source::PubSub(PubSubSource::new(sub_id.clone(), compression.clone()).await?), }; Ok(Self { source, @@ -58,6 +71,13 @@ impl BlockSource { }; query_blocks.await? } + #[cfg(feature = "pubsub")] + Source::PubSub(source) => { + let query_blocks = match self.chain { + Chain::Ethereum => source.get_block_stream::(sender), + }; + query_blocks.await? + } }; Ok(()) diff --git a/src/components/block_source/pubsub/mod.rs b/src/components/block_source/pubsub/mod.rs new file mode 100644 index 0000000..7f983cb --- /dev/null +++ b/src/components/block_source/pubsub/mod.rs @@ -0,0 +1,69 @@ +use crate::common::BlockDataMessage; +use crate::components::block_source::delta::DeltaBlockTrait; +use crate::errors::SourceError; +use crate::proto::ethereum::Block; +use futures_util::StreamExt; +use google_cloud_pubsub::client::{Client, ClientConfig}; +use google_cloud_pubsub::subscription::Subscription; +use kanal::AsyncSender; +use prost::Message; + +#[derive(Clone)] +pub struct PubSubSource { + sub: Subscription, + compression: bool, +} + +impl PubSubSource { + pub async fn new(sub_id: String, compression: bool) -> Result { + let cfg = ClientConfig::default() + .with_auth() + .await + .map_err(|e| SourceError::PubSubError(format!("Failed to auth pubsub: {:?}", e)))?; + let client = Client::new(cfg).await.map_err(|e| { + SourceError::PubSubError(format!("Failed to create pubsub client: {:?}", e)) + })?; + let sub = client.subscription(&sub_id); + Ok(Self { sub, compression }) + } + + pub async fn get_block_stream( + &self, + sender: AsyncSender>, + ) -> Result<(), SourceError> { + let mut stream = self.sub.subscribe(None).await?; + while let Some(message) = stream.next().await { + let block = if self.compression { + let block_compressed = lz4::block::decompress(&message.message.data, None) + .map_err(|e| { + SourceError::DecodeMessageError(format!( + "Failed to decompress block: {:?}", + e + )) + })?; + Block::decode(&block_compressed) + .map_err(|e| { + SourceError::DecodeMessageError(format!("Failed to decode block: {:?}", e)) + }) + .map(BlockDataMessage::from)? + } else { + Block::decode(&message.message.data) + .map_err(|e| { + SourceError::DecodeMessageError(format!("Failed to decode block: {:?}", e)) + }) + .map(BlockDataMessage::from)? + }; + sender + .send(vec![block]) + .await + .map_err(|e| SourceError::ChannelSendFail(e))?; + if let Err(e) = message.ack().await { + return Err(SourceError::PubSubError(format!( + "Failed to ack message: {:?}", + e + )))?; + } + } + Ok(()) + } +} diff --git a/src/config.rs b/src/config.rs index c31863d..b0ecd0f 100644 --- a/src/config.rs +++ b/src/config.rs @@ -5,6 +5,7 @@ use figment::providers::Toml; use figment::Figment; use serde::Deserialize; +#[cfg(feature = "deltalake")] #[derive(Clone, Debug, Deserialize)] pub struct DeltaConfig { pub table_path: String, @@ -15,7 +16,10 @@ pub struct DeltaConfig { #[derive(Deserialize, Clone, Debug)] #[serde(rename_all = "lowercase")] pub enum SourceTypes { + #[cfg(feature = "deltalake")] Delta(DeltaConfig), + #[cfg(feature = "pubsub")] + PubSub { sub_id: String, compression: bool }, } #[derive(Deserialize, Clone, Debug)] diff --git a/src/errors.rs b/src/errors.rs index cf96076..0d74428 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -186,6 +186,12 @@ pub enum SourceError { DeltaSerializationError, #[error("No blocks found from Delta")] DeltaEmptyData, + #[cfg(feature = "pubsub")] + #[error("PubSub error: {0}")] + PubSubError(String), + #[cfg(feature = "pubsub")] + #[error("Decode message error: {0}")] + DecodeMessageError(String), } #[derive(Debug, Error)]