fix(core): pipeline doesn’t stop automatically after all registered datasources complete (#317)

* feat: add rpc-block-crawler example

* fix: complete pipeline run after all datasources finish

* fix(datasources): drop sender channel after complete crawling in rpc block crawler

* fix: sort Cargo.toml

Author: kalloc

Cargo.lock

@@ -10,34 +10,36 @@ license = "MIT"
 repository = "https://github.com/sevenlabs-hq/carbon"
 
 [workspace.dependencies]
-# main
-carbon-cli = { path = "crates/cli", version = "0.8.1" }
-carbon-core = { path = "crates/core", version = "0.8.1" }
-carbon-gql-server = { path = "crates/gql-server", version = "0.8.1" }
-carbon-macros = { path = "crates/macros", version = "0.8.1" }
-carbon-postgres-client = { path = "crates/postgres-client", version = "0.8.1" }
-carbon-proc-macros = { path = "crates/proc-macros", version = "0.8.1" }
-carbon-test-utils = { path = "crates/test-utils", version = "0.8.1" }
 
-# datasources
-carbon-helius-atlas-ws-datasource = { path = "datasources/helius-atlas-ws-datasource", version = "0.8.1" }
-carbon-jito-shredstream-grpc-datasource = { path = "datasources/jito-shredstream-grpc-datasource", version = "0.8.1" }
-carbon-rpc-block-crawler-datasource = { path = "datasources/rpc-block-crawler-datasource", version = "0.8.1" }
-carbon-rpc-block-subscribe-datasource = { path = "datasources/rpc-block-subscribe-datasource", version = "0.8.1" }
-carbon-rpc-program-subscribe-datasource = { path = "datasources/rpc-program-subscribe-datasource", version = "0.8.1" }
-carbon-rpc-transaction-crawler-datasource = { path = "datasources/rpc-transaction-crawler-datasource", version = "0.8.1" }
-carbon-yellowstone-grpc-datasource = { path = "datasources/yellowstone-grpc-datasource", version = "0.8.1" }
-
-# metrics
-carbon-log-metrics = { path = "metrics/log-metrics", version = "0.8.1" }
-carbon-prometheus-metrics = { path = "metrics/prometheus-metrics", version = "0.8.1" }
+# other
+anyhow = "1.0.96"
+askama = "0.12.1"
+async-stream = "0.3.6"
+async-trait = { version = "0.1.86" }
+axum = "0.8.4"
+base64 = "0.22.1"
+bincode = "1.3.3"
+borsh = { version = "1.5.1" }
+borsh-derive-internal = "0.10.3"
+bs58 = { version = "0.5.1", default-features = false }
 
 # decoders
 carbon-address-lookup-table-decoder = { path = "decoders/address-lookup-table-decoder", version = "0.8.1" }
 carbon-associated-token-account-decoder = { path = "decoders/associated-token-account-decoder", version = "0.8.1" }
 carbon-boop-decoder = { path = "decoders/boop-decoder", version = "0.8.1" }
+# main
+carbon-cli = { path = "crates/cli", version = "0.8.1" }
+carbon-core = { path = "crates/core", version = "0.8.1" }
 carbon-drift-v2-decoder = { path = "decoders/drift-v2-decoder", version = "0.8.1" }
 carbon-fluxbeam-decoder = { path = "decoders/fluxbeam-decoder", version = "0.8.1" }
+carbon-gql-server = { path = "crates/gql-server", version = "0.8.1" }
+
+# datasources
+carbon-helius-atlas-ws-datasource = { path = "datasources/helius-atlas-ws-datasource", version = "0.8.1" }
+
+# misc
+carbon-jito-protos = { path = "misc/jito-protos", version = "0.2.4" }
+carbon-jito-shredstream-grpc-datasource = { path = "datasources/jito-shredstream-grpc-datasource", version = "0.8.1" }
 carbon-jupiter-dca-decoder = { path = "decoders/jupiter-dca-decoder", version = "0.8.1" }
 carbon-jupiter-limit-order-2-decoder = { path = "decoders/jupiter-limit-order-2-decoder", version = "0.8.1" }
 carbon-jupiter-limit-order-decoder = { path = "decoders/jupiter-limit-order-decoder", version = "0.8.1" }
@@ -47,6 +49,10 @@ carbon-kamino-farms-decoder = { path = "decoders/kamino-farms-decoder", version
 carbon-kamino-lending-decoder = { path = "decoders/kamino-lending-decoder", version = "0.8.1" }
 carbon-kamino-vault-decoder = { path = "decoders/kamino-vault-decoder", version = "0.8.1" }
 carbon-lifinity-amm-v2-decoder = { path = "decoders/lifinity-amm-v2-decoder", version = "0.8.1" }
+
+# metrics
+carbon-log-metrics = { path = "metrics/log-metrics", version = "0.8.1" }
+carbon-macros = { path = "crates/macros", version = "0.8.1" }
 carbon-marginfi-v2-decoder = { path = "decoders/marginfi-v2-decoder", version = "0.8.1" }
 carbon-marinade-finance-decoder = { path = "decoders/marinade-finance-decoder", version = "0.8.1" }
 carbon-memo-program-decoder = { path = "decoders/memo-program-decoder", version = "0.8.1" }
@@ -60,61 +66,33 @@ carbon-okx-dex-decoder = { path = "decoders/okx-dex-decoder", version = "0.8.1"
 carbon-openbook-v2-decoder = { path = "decoders/openbook-v2-decoder", version = "0.8.1" }
 carbon-orca-whirlpool-decoder = { path = "decoders/orca-whirlpool-decoder", version = "0.8.1" }
 carbon-phoenix-v1-decoder = { path = "decoders/phoenix-v1-decoder", version = "0.8.1" }
+carbon-postgres-client = { path = "crates/postgres-client", version = "0.8.1" }
+carbon-proc-macros = { path = "crates/proc-macros", version = "0.8.1" }
+carbon-prometheus-metrics = { path = "metrics/prometheus-metrics", version = "0.8.1" }
 carbon-pump-swap-decoder = { path = "decoders/pump-swap-decoder", version = "0.8.1" }
 carbon-pumpfun-decoder = { path = "decoders/pumpfun-decoder", version = "0.8.1" }
 carbon-raydium-amm-v4-decoder = { path = "decoders/raydium-amm-v4-decoder", version = "0.8.1" }
 carbon-raydium-clmm-decoder = { path = "decoders/raydium-clmm-decoder", version = "0.8.1" }
 carbon-raydium-cpmm-decoder = { path = "decoders/raydium-cpmm-decoder", version = "0.8.1" }
 carbon-raydium-launchpad-decoder = { path = "decoders/raydium-launchpad-decoder", version = "0.8.1" }
 carbon-raydium-liquidity-locking-decoder = { path = "decoders/carbon-raydium-liquidity-locking-decoder", version = "0.8.1" }
+carbon-rpc-block-crawler-datasource = { path = "datasources/rpc-block-crawler-datasource", version = "0.8.1" }
+carbon-rpc-block-subscribe-datasource = { path = "datasources/rpc-block-subscribe-datasource", version = "0.8.1" }
+carbon-rpc-program-subscribe-datasource = { path = "datasources/rpc-program-subscribe-datasource", version = "0.8.1" }
+carbon-rpc-transaction-crawler-datasource = { path = "datasources/rpc-transaction-crawler-datasource", version = "0.8.1" }
 carbon-sharky-decoder = { path = "decoders/sharky-decoder", version = "0.8.1" }
 carbon-solayer-restaking-program-decoder = { path = "decoders/solayer-restaking-program-decoder", version = "0.8.1" }
 carbon-stabble-stable-swap-decoder = { path = "decoders/carbon-stabble-stable-swap-decoder", version = "0.8.1" }
 carbon-stabble-weighted-swap-decoder = { path = "decoders/carbon-stabble-weighted-swap-decoder", version = "0.8.1" }
 carbon-stake-program-decoder = { path = "decoders/carbon-stake-program-decoder", version = "0.8.1" }
 carbon-system-program-decoder = { path = "decoders/system-program-decoder", version = "0.8.1" }
+carbon-test-utils = { path = "crates/test-utils", version = "0.8.1" }
 carbon-token-2022-decoder = { path = "decoders/token-2022-decoder", version = "0.8.1" }
 carbon-token-program-decoder = { path = "decoders/token-program-decoder", version = "0.8.1" }
 carbon-virtual-curve-decoder = { path = "decoders/virtual-curve-decoder", version = "0.8.1" }
 carbon-virtuals-decoder = { path = "decoders/virtuals-decoder", version = "0.8.1" }
+carbon-yellowstone-grpc-datasource = { path = "datasources/yellowstone-grpc-datasource", version = "0.8.1" }
 carbon-zeta-decoder = { path = "decoders/zeta-decoder", version = "0.8.1" }
-
-# misc
-carbon-jito-protos = { path = "misc/jito-protos", version = "0.2.4" }
-
-# solana
-solana-account = "2.2"
-solana-account-decoder = "2.2"
-solana-account-decoder-client-types = "2.2"
-solana-client = "2.2"
-solana-clock = "2.2"
-solana-commitment-config = "2.2"
-solana-entry = "2.2"
-solana-hash = "2.2"
-solana-instruction = { version = "2.2", default-features = false }
-solana-message = "2.2"
-solana-native-token = "2.2"
-solana-program = "2.2"
-solana-program-pack = "2.2"
-solana-pubkey = { version = "2.2", features = ["serde"] }
-solana-signature = "2.2"
-solana-transaction = "2.2"
-solana-transaction-context = "2.2"
-solana-transaction-status = "2.2"
-spl-memo = "5.0.0"
-spl-token = "6.0.0"
-
-# other
-anyhow = "1.0.96"
-askama = "0.12.1"
-async-stream = "0.3.6"
-async-trait = { version = "0.1.86" }
-axum = "0.8.4"
-base64 = "0.22.1"
-bincode = "1.3.3"
-borsh = { version = "1.5.1" }
-borsh-derive-internal = "0.10.3"
-bs58 = { version = "0.5.1", default-features = false }
 chrono = { version = "0.4.40", features = ["serde"] }
 clap = { version = "4.5.30", features = ["derive"] }
 console = "0.15.8"
@@ -147,6 +125,28 @@ serde = { version = "1.0.208", features = ["derive"] }
 serde-big-array = "0.5.1"
 serde_json = "1.0.138"
 sha2 = "0.10.8"
+
+# solana
+solana-account = "2.2"
+solana-account-decoder = "2.2"
+solana-account-decoder-client-types = "2.2"
+solana-client = "2.2"
+solana-clock = "2.2"
+solana-commitment-config = "2.2"
+solana-entry = "2.2"
+solana-hash = "2.2"
+solana-instruction = { version = "2.2", default-features = false }
+solana-message = "2.2"
+solana-native-token = "2.2"
+solana-program = "2.2"
+solana-program-pack = "2.2"
+solana-pubkey = { version = "2.2", features = ["serde"] }
+solana-signature = "2.2"
+solana-transaction = "2.2"
+solana-transaction-context = "2.2"
+solana-transaction-status = "2.2"
+spl-memo = "5.0.0"
+spl-token = "6.0.0"
 sqlx = { version = "0.8.5", features = [
     "macros",
     "runtime-tokio-rustls",

Cargo.toml

@@ -102,7 +102,7 @@ use {
 pub trait Datasource: Send + Sync {
     async fn consume(
         &self,
-        sender: &tokio::sync::mpsc::Sender<Update>,
+        sender: tokio::sync::mpsc::Sender<Update>,
         cancellation_token: CancellationToken,
         metrics: Arc<MetricsCollection>,
     ) -> CarbonResult<()>;

crates/core/src/datasource.rs

@@ -354,7 +354,7 @@ impl Pipeline {
             tokio::spawn(async move {
                 if let Err(e) = datasource_clone
                     .consume(
-                        &sender_clone,
+                        sender_clone,
                         datasource_cancellation_token_clone,
                         metrics_collection,
                     )
@@ -365,6 +365,8 @@ impl Pipeline {
             });
         }
 
+        drop(update_sender);
+
         let mut interval = tokio::time::interval(time::Duration::from_secs(
             self.metrics_flush_interval.unwrap_or(5),
         ));

crates/core/src/pipeline.rs

@@ -94,7 +94,7 @@ impl HeliusWebsocket {
 impl Datasource for HeliusWebsocket {
     async fn consume(
         &self,
-        sender: &Sender<Update>,
+        sender: Sender<Update>,
         cancellation_token: CancellationToken,
         metrics: Arc<MetricsCollection>,
     ) -> CarbonResult<()> {

datasources/helius-atlas-ws-datasource/src/lib.rs

@@ -34,11 +34,10 @@ impl JitoShredstreamGrpcClient {
 impl Datasource for JitoShredstreamGrpcClient {
     async fn consume(
         &self,
-        sender: &Sender<Update>,
+        sender: Sender<Update>,
         cancellation_token: CancellationToken,
         metrics: Arc<MetricsCollection>,
     ) -> CarbonResult<()> {
-        let sender = sender.clone();
         let endpoint = self.0.clone();
 
         let mut client = ShredstreamProxyClient::connect(endpoint)