create txn stream crate in this repo

Author: rtso

rust/Cargo.lock

@@ -1,7 +1,13 @@
 [workspace]
 resolver = "2"
 
-members = ["instrumented-channel", "sdk", "sdk-examples"]
+members = [
+    "instrumented-channel",
+    "moving-average",
+    "sdk",
+    "sdk-examples",
+    "transaction-stream",
+]
 
 [workspace.package]
 authors = ["Aptos Labs <[email protected]>"]
@@ -14,12 +20,13 @@ rust-version = "1.75"
 
 [workspace.dependencies]
 aptos-indexer-processor-sdk = { path = "sdk" }
+aptos-indexer-transaction-stream = { path = "transaction-stream" }
 instrumented-channel = { path = "instrumented-channel" }
+aptos-moving-average = { path = "moving-average" }
 sdk-server-framework = { path = 'sdk-server-framework' }
 
 ahash = { version = "0.8.7", features = ["serde"] }
 anyhow = "1.0.86"
-aptos-indexer-transaction-stream = { git = 'https://github.com/aptos-labs/aptos-indexer-processors.git', rev = '3c8896b489062ce0510c5e4162366db753b59499' }
 aptos-protos = { git = "https://github.com/aptos-labs/aptos-core.git", tag = "aptos-node-v1.12.1" }
 aptos-system-utils = { git = "https://github.com/aptos-labs/aptos-core.git", rev = "4541add3fd29826ec57f22658ca286d2d6134b93" }
 async-trait = "0.1.80"
@@ -54,6 +61,7 @@ field_count = "0.1.1"
 futures = "0.3.30"
 futures-util = "0.3.21"
 hex = "0.4.3"
+itertools = "0.13.0"
 jemallocator = { version = "0.5.0", features = [
     "profiling",
     "unprefixed_malloc_on_supported_platforms",
@@ -66,6 +74,7 @@ once_cell = { version = "1.19.0" }
 petgraph = "0.6.5"
 prometheus = "0.13.4"
 prometheus-client = "0.22.2"
+prost = { version = "0.12.3", features = ["no-recursion-limit"] }
 rayon = "1.10.0"
 serde = { version = "1.0.193", features = ["derive", "rc"] }
 serde_json = { version = "1.0.81", features = ["preserve_order"] }
@@ -78,6 +87,15 @@ tiny-keccak = { version = "2.0.2", features = ["keccak", "sha3"] }
 tracing = "0.1.34"
 tokio = { version = "1.37.0", features = ["full"] }
 toml = "0.7.4"
+tonic = { version = "0.11.0", features = [
+    "tls",
+    "tls-roots",
+    "transport",
+    "prost",
+    "gzip",
+    "codegen",
+    "zstd",
+] }
 tracing-subscriber = { version = "0.3.17", features = ["json", "env-filter"] }
 url = { version = "2.5.1", features = ["serde"] }
 warp = { version = "0.3.5", features = ["tls"] }

rust/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "aptos-moving-average"
+description = "Utility to calculate moving average such as TPS"
+version = "0.1.0"
+
+# Workspace inherited keys
+authors = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+repository = { workspace = true }
+rust-version = { workspace = true }
+
+[dependencies]
+chrono = { workspace = true }

rust/moving-average/Cargo.toml

@@ -0,0 +1,84 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+#![forbid(unsafe_code)]
+
+use std::collections::VecDeque;
+
+// TPS data
+pub struct MovingAverage {
+    window_millis: u64,
+    // (timestamp_millis, value)
+    values: VecDeque<(u64, u64)>,
+    sum: u64,
+}
+
+impl MovingAverage {
+    pub fn new(window_millis: u64) -> Self {
+        let now = chrono::Utc::now().naive_utc().timestamp_millis() as u64;
+        let mut queue = VecDeque::new();
+        queue.push_back((now, 0));
+        Self {
+            window_millis,
+            values: queue,
+            sum: 0,
+        }
+    }
+
+    pub fn tick_now(&mut self, value: u64) {
+        let now = chrono::Utc::now().naive_utc().timestamp_millis() as u64;
+        self.tick(now, value);
+    }
+
+    pub fn tick(&mut self, timestamp_millis: u64, value: u64) -> f64 {
+        self.values.push_back((timestamp_millis, value));
+        self.sum += value;
+        while self.values.len() > 2 {
+            match self.values.front() {
+                None => break,
+                Some((ts, val)) => {
+                    if timestamp_millis - ts > self.window_millis {
+                        self.sum -= val;
+                        self.values.pop_front();
+                    } else {
+                        break;
+                    }
+                },
+            }
+        }
+        self.avg()
+    }
+
+    // Only be called after tick_now/tick is called.
+    pub fn avg(&self) -> f64 {
+        if self.values.len() < 2 {
+            0.0
+        } else {
+            let elapsed = self.values.back().unwrap().0 - self.values.front().unwrap().0;
+            (self.sum * 1000) as f64 / elapsed as f64
+        }
+    }
+
+    pub fn sum(&self) -> u64 {
+        self.sum
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn test_moving_average() {
+        // 10 Second window.
+        let mut ma = MovingAverage::new(10_000);
+        // 9 seconds spent at 100 TPS.
+        for _ in 0..9 {
+            ma.tick_now(100);
+            std::thread::sleep(std::time::Duration::from_secs(1));
+        }
+        // No matter what algorithm we use, the average should be 99 at least.
+        let avg = ma.avg();
+        assert!(avg >= 99.0, "Average is too low: {}", avg);
+    }
+}

rust/moving-average/src/lib.rs

@@ -31,7 +31,7 @@ async fn run_processor() -> Result<()> {
     // let (input_sender, input_receiver) = kanal::bounded_async(1);
     let transaction_stream_config = TransactionStreamConfig {
         indexer_grpc_data_service_address: Url::parse("https://grpc.devnet.aptoslabs.com:443")?,
-        starting_version: 0,
+        starting_version: Some(0),
         request_ending_version: None,
         auth_token: String::from("aptoslabs_TJs4NQU8Xf5_EJMNnZFPXRH6YNpWM7bCcurMBEUtZtRb6"),
         request_name_header: String::from("sdk_processor"),

rust/sdk/src/main.rs

@@ -1,4 +1,4 @@
-use aptos_indexer_transaction_stream::utils::util::timestamp_to_unixtime;
+use aptos_indexer_transaction_stream::utils::timestamp::timestamp_to_unixtime;
 
 /// TransactionContext is a struct that holds data processed from a set of transactions
 /// and includes metadata about the transactions that the data is associated with.

rust/sdk/src/types/transaction_context.rs

@@ -0,0 +1,31 @@
+[package]
+name = "aptos-indexer-transaction-stream"
+version = "0.1.0"
+
+# Workspace inherited keys
+authors = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+repository = { workspace = true }
+rust-version = { workspace = true }
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+anyhow = { workspace = true }
+aptos-moving-average = { workspace = true }
+aptos-protos = { workspace = true }
+bigdecimal = { workspace = true }
+chrono = { workspace = true }
+futures-util = { workspace = true }
+itertools = { workspace = true }
+kanal = { workspace = true }
+once_cell = { workspace = true }
+prometheus = { workspace = true }
+prost = { workspace = true }
+serde = { workspace = true }
+tokio = { workspace = true }
+tonic = { workspace = true }
+tracing = { workspace = true }
+url = { workspace = true }

rust/transaction-stream/Cargo.toml

@@ -0,0 +1,60 @@
+use serde::{Deserialize, Serialize};
+use std::time::Duration;
+use url::Url;
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct TransactionStreamConfig {
+    pub indexer_grpc_data_service_address: Url,
+    pub starting_version: Option<u64>,
+    pub request_ending_version: Option<u64>,
+    pub auth_token: String,
+    pub request_name_header: String,
+    #[serde(default = "TransactionStreamConfig::default_indexer_grpc_http2_ping_interval")]
+    pub indexer_grpc_http2_ping_interval_secs: u64,
+    #[serde(default = "TransactionStreamConfig::default_indexer_grpc_http2_ping_timeout")]
+    pub indexer_grpc_http2_ping_timeout_secs: u64,
+    #[serde(default = "TransactionStreamConfig::default_indexer_grpc_reconnection_timeout")]
+    pub indexer_grpc_reconnection_timeout_secs: u64,
+    #[serde(default = "TransactionStreamConfig::default_indexer_grpc_response_item_timeout")]
+    pub indexer_grpc_response_item_timeout_secs: u64,
+}
+
+impl TransactionStreamConfig {
+    pub const fn indexer_grpc_http2_ping_interval(&self) -> Duration {
+        Duration::from_secs(self.indexer_grpc_http2_ping_interval_secs)
+    }
+
+    pub const fn indexer_grpc_http2_ping_timeout(&self) -> Duration {
+        Duration::from_secs(self.indexer_grpc_http2_ping_timeout_secs)
+    }
+
+    pub const fn indexer_grpc_reconnection_timeout(&self) -> Duration {
+        Duration::from_secs(self.indexer_grpc_reconnection_timeout_secs)
+    }
+
+    pub const fn indexer_grpc_response_item_timeout(&self) -> Duration {
+        Duration::from_secs(self.indexer_grpc_response_item_timeout_secs)
+    }
+
+    /// Indexer GRPC http2 ping interval in seconds. Defaults to 30.
+    /// Tonic ref: https://docs.rs/tonic/latest/tonic/transport/channel/struct.Endpoint.html#method.http2_keep_alive_interval
+    pub const fn default_indexer_grpc_http2_ping_interval() -> u64 {
+        30
+    }
+
+    /// Indexer GRPC http2 ping timeout in seconds. Defaults to 10.
+    pub const fn default_indexer_grpc_http2_ping_timeout() -> u64 {
+        10
+    }
+
+    /// Default timeout for establishing a grpc connection. Defaults to 5 seconds.
+    pub const fn default_indexer_grpc_reconnection_timeout() -> u64 {
+        5
+    }
+
+    /// Default timeout for receiving an item from grpc stream. Defaults to 60 seconds.
+    pub const fn default_indexer_grpc_response_item_timeout() -> u64 {
+        60
+    }
+}

rust/transaction-stream/src/config.rs

@@ -0,0 +1,6 @@
+pub mod config;
+pub mod transaction_stream;
+pub mod utils;
+
+pub use config::TransactionStreamConfig;
+pub use transaction_stream::{TransactionStream, TransactionsPBResponse};