all: Use a workspace package for tokio

lutter · lutter · commit 939391c4d889 · 2025-10-22T18:47:26.000+02:00
Instead of reexporting from the graph crate, declare tokio once in the
workspace and use it as needed
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -93,6 +93,9 @@ syn = { version = "2.0.106", features = ["full"] }
 test-store = { path = "./store/test-store" }
 thiserror = "2.0.16"
 tokio = { version = "1.45.1", features = ["full"] }
+tokio-stream = { version = "0.1.15", features = ["sync"] }
+tokio-retry = "0.3.0"
+
 tonic = { version = "0.12.3", features = ["tls-roots", "gzip"] }
 tonic-build = { version = "0.12.3", features = ["prost"] }
 tower-http = { version = "0.6.6", features = ["cors"] }
diff --git a/chain/ethereum/Cargo.toml b/chain/ethereum/Cargo.toml
@@ -16,6 +16,8 @@ tiny-keccak = "1.5.0"
 hex = "0.4.3"
 semver = "1.0.27"
 thiserror = { workspace = true }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
 
 itertools = "0.14.0"
 
diff --git a/chain/ethereum/src/chain.rs b/chain/ethereum/src/chain.rs
@@ -1268,8 +1268,10 @@ impl FirehoseMapperTrait<Chain> for FirehoseMapper {
 
 #[cfg(test)]
 mod tests {
+    use tokio;
+
     use graph::blockchain::mock::MockChainStore;
-    use graph::{slog, tokio};
+    use graph::slog;
 
     use super::*;
     use std::sync::Arc;
diff --git a/chain/ethereum/src/ethereum_adapter.rs b/chain/ethereum/src/ethereum_adapter.rs
@@ -1,5 +1,8 @@
 use async_trait::async_trait;
 use futures03::{future::BoxFuture, stream::FuturesUnordered};
+use tokio::sync::RwLock;
+use tokio::time::timeout;
+
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::BlockHash;
 use graph::blockchain::ChainIdentifier;
@@ -23,8 +26,6 @@ use graph::prelude::ethabi::Token;
 use graph::prelude::tokio::try_join;
 use graph::prelude::web3::types::U256;
 use graph::slog::o;
-use graph::tokio::sync::RwLock;
-use graph::tokio::time::timeout;
 use graph::{
     blockchain::{block_stream::BlockWithTriggers, BlockPtr, IngestorError},
     prelude::{
@@ -2368,7 +2369,7 @@ async fn fetch_individual_receipts_with_retry(
     }
 
     // Use a stream to fetch receipts individually
-    let hash_stream = graph::tokio_stream::iter(hashes);
+    let hash_stream = tokio_stream::iter(hashes);
     let receipt_stream = hash_stream
         .map(move |tx_hash| {
             fetch_transaction_receipt_with_retry(
@@ -2380,7 +2381,7 @@ async fn fetch_individual_receipts_with_retry(
         })
         .buffered(ENV_VARS.block_ingestor_max_concurrent_json_rpc_calls);
 
-    graph::tokio_stream::StreamExt::collect::<Result<Vec<Arc<TransactionReceipt>>, IngestorError>>(
+    tokio_stream::StreamExt::collect::<Result<Vec<Arc<TransactionReceipt>>, IngestorError>>(
         receipt_stream,
     )
     .await
diff --git a/chain/ethereum/src/network.rs b/chain/ethereum/src/network.rs
@@ -309,6 +309,8 @@ impl EthereumNetworkAdapters {
 
 #[cfg(test)]
 mod tests {
+    use tokio;
+
     use graph::cheap_clone::CheapClone;
     use graph::components::network_provider::ProviderCheckStrategy;
     use graph::components::network_provider::ProviderManager;
@@ -320,7 +322,6 @@ mod tests {
         firehose::SubgraphLimit,
         prelude::MetricsRegistry,
         slog::{o, Discard, Logger},
-        tokio,
         url::Url,
     };
     use std::sync::Arc;
diff --git a/chain/ethereum/src/polling_block_stream.rs b/chain/ethereum/src/polling_block_stream.rs
@@ -1,11 +1,11 @@
 use anyhow::{anyhow, Error};
-use graph::tokio;
 use std::cmp;
 use std::collections::VecDeque;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 use std::time::Duration;
+use tokio;
 
 use graph::blockchain::block_stream::{
     BlockStream, BlockStreamError, BlockStreamEvent, BlockWithTriggers, ChainHeadUpdateStream,
diff --git a/chain/near/Cargo.toml b/chain/near/Cargo.toml
@@ -20,3 +20,4 @@ graph-runtime-derive = { path = "../../runtime/derive" }
 [dev-dependencies]
 diesel = { workspace = true }
 trigger-filters.path = "../../substreams/trigger-filters"
+tokio = { workspace = true }
diff --git a/chain/near/src/trigger.rs b/chain/near/src/trigger.rs
@@ -154,6 +154,8 @@ pub struct ReceiptWithOutcome {
 mod tests {
     use std::convert::TryFrom;
 
+    use tokio;
+
     use super::*;
 
     use graph::{
@@ -162,7 +164,6 @@ mod tests {
         data::subgraph::API_VERSION_0_0_5,
         prelude::{hex, BigInt},
         runtime::{gas::GasCounter, DeterministicHostError, HostExportError},
-        tokio,
         util::mem::init_slice,
     };
 
diff --git a/chain/substreams/Cargo.toml b/chain/substreams/Cargo.toml
@@ -18,6 +18,7 @@ anyhow = "1.0"
 hex = "0.4.3"
 semver = "1.0.27"
 base64 = "0.22.1"
+tokio-stream = { workspace = true }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full"] }
diff --git a/chain/substreams/examples/substreams.rs b/chain/substreams/examples/substreams.rs
@@ -5,12 +5,12 @@ use graph::blockchain::substreams_block_stream::SubstreamsBlockStream;
 use graph::endpoint::EndpointMetrics;
 use graph::firehose::{FirehoseEndpoints, SubgraphLimit};
 use graph::prelude::{info, tokio, DeploymentHash, MetricsRegistry, Registry};
-use graph::tokio_stream::StreamExt;
 use graph::{env::env_var, firehose::FirehoseEndpoint, log::logger, substreams};
 use graph_chain_substreams::mapper::Mapper;
 use prost::Message;
 use std::env;
 use std::sync::Arc;
+use tokio_stream::StreamExt;
 
 #[tokio::main]
 async fn main() -> Result<(), Error> {
diff --git a/chain/substreams/src/block_ingestor.rs b/chain/substreams/src/block_ingestor.rs
@@ -13,14 +13,14 @@ use graph::components::store::ChainHeadStore;
 use graph::prelude::MetricsRegistry;
 use graph::slog::trace;
 use graph::substreams::Package;
-use graph::tokio_stream::StreamExt;
 use graph::{
     blockchain::block_stream::BlockStreamEvent,
     cheap_clone::CheapClone,
     prelude::{error, info, DeploymentHash, Logger},
     util::backoff::ExponentialBackoff,
 };
 use prost::Message;
+use tokio_stream::StreamExt;
 
 const SUBSTREAMS_HEAD_TRACKER_BYTES: &[u8; 89935] = include_bytes!(
     "../../../substreams/substreams-head-tracker/substreams-head-tracker-v1.0.0.spkg"
diff --git a/core/Cargo.toml b/core/Cargo.toml
@@ -13,6 +13,8 @@ graph-chain-near = { path = "../chain/near" }
 graph-chain-substreams = { path = "../chain/substreams" }
 graph-runtime-wasm = { path = "../runtime/wasm" }
 serde_yaml = { workspace = true }
+tokio = { workspace = true }
+tokio-retry = { workspace = true }
 # Switch to crates.io once tower 0.5 is released
 tower = { git = "https://github.com/tower-rs/tower.git", features = ["full"] }
 thiserror = { workspace = true }
diff --git a/core/src/polling_monitor/ipfs_service.rs b/core/src/polling_monitor/ipfs_service.rs
@@ -109,7 +109,7 @@ mod test {
     use graph::ipfs::test_utils::add_files_to_local_ipfs_node_for_testing;
     use graph::ipfs::{IpfsContext, IpfsMetrics, IpfsRpcClient, ServerAddress};
     use graph::log::discard;
-    use graph::tokio;
+    use tokio;
     use tower::ServiceExt;
     use wiremock::matchers as m;
     use wiremock::Mock;
diff --git a/core/src/subgraph/context/mod.rs b/core/src/subgraph/context/mod.rs
@@ -25,10 +25,10 @@ use graph::{
         TriggerProcessor,
     },
     slog::Logger,
-    tokio::sync::mpsc,
 };
 use std::sync::{Arc, RwLock};
 use std::{collections::HashMap, time::Instant};
+use tokio::sync::mpsc;
 
 use self::instance::SubgraphInstance;
 use super::Decoder;
@@ -281,7 +281,7 @@ impl OffchainMonitor {
     }
 
     pub fn ready_offchain_events(&mut self) -> Result<Vec<offchain::TriggerData>, Error> {
-        use graph::tokio::sync::mpsc::error::TryRecvError;
+        use tokio::sync::mpsc::error::TryRecvError;
 
         let mut triggers = vec![];
         loop {
diff --git a/core/src/subgraph/registrar.rs b/core/src/subgraph/registrar.rs
@@ -18,9 +18,9 @@ use graph::prelude::{
     CreateSubgraphResult, SubgraphAssignmentProvider as SubgraphAssignmentProviderTrait,
     SubgraphRegistrar as SubgraphRegistrarTrait, *,
 };
-use graph::tokio_retry::Retry;
 use graph::util::futures::retry_strategy;
 use graph::util::futures::RETRY_DEFAULT_LIMIT;
+use tokio_retry::Retry;
 
 pub struct SubgraphRegistrar<P, S, SM> {
     logger: Logger,
diff --git a/gnd/src/main.rs b/gnd/src/main.rs
@@ -9,11 +9,11 @@ use graph::{
     log::logger,
     prelude::{CheapClone, DeploymentHash, LinkResolver, SubgraphName},
     slog::{error, info, Logger},
-    tokio::{self, sync::mpsc},
 };
 use graph_core::polling_monitor::ipfs_service;
 use graph_node::{launcher, opt::Opt};
 use lazy_static::lazy_static;
+use tokio::{self, sync::mpsc};
 
 use gnd::watcher::{deploy_all_subgraphs, parse_manifest_args, watch_subgraphs};
 
diff --git a/gnd/src/watcher.rs b/gnd/src/watcher.rs
@@ -2,12 +2,12 @@ use anyhow::{anyhow, Context, Result};
 use globset::{Glob, GlobSet, GlobSetBuilder};
 use graph::prelude::{DeploymentHash, SubgraphName};
 use graph::slog::{self, error, info, Logger};
-use graph::tokio::sync::mpsc::Sender;
 use notify::{recommended_watcher, Event, RecursiveMode, Watcher};
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::sync::mpsc;
 use std::time::Duration;
+use tokio::sync::mpsc::Sender;
 
 const WATCH_DELAY: Duration = Duration::from_secs(5);
 const DEFAULT_BUILD_DIR: &str = "build";
diff --git a/graph/Cargo.toml b/graph/Cargo.toml
@@ -68,16 +68,9 @@ slog-envlogger = "2.1.0"
 slog-term = "2.7.0"
 petgraph = "0.8.2"
 tiny-keccak = "1.5.0"
-tokio = { version = "1.45.1", features = [
-    "time",
-    "sync",
-    "macros",
-    "test-util",
-    "rt-multi-thread",
-    "parking_lot",
-] }
-tokio-stream = { version = "0.1.15", features = ["sync"] }
-tokio-retry = "0.3.0"
+tokio = { workspace = true}
+tokio-stream = { workspace = true }
+tokio-retry = { workspace = true }
 toml = "0.9.7"
 url = "2.5.7"
 prometheus = "0.14.0"
diff --git a/graph/src/components/server/server.rs b/graph/src/components/server/server.rs
@@ -5,15 +5,15 @@ use std::sync::Arc;
 
 use hyper::body::Incoming;
 use hyper::Request;
+use tokio::net::TcpListener;
+use tokio::task::JoinHandle;
 
+use crate::anyhow;
 use crate::cheap_clone::CheapClone;
 use crate::hyper::server::conn::http1;
 use crate::hyper::service::service_fn;
 use crate::hyper_util::rt::TokioIo;
 use crate::slog::error;
-use crate::tokio::net::TcpListener;
-use crate::tokio::task::JoinHandle;
-use crate::{anyhow, tokio};
 
 use crate::prelude::Logger;
 
diff --git a/graph/src/lib.rs b/graph/src/lib.rs
@@ -62,9 +62,6 @@ pub use slog;
 pub use sqlparser;
 pub use stable_hash;
 pub use stable_hash_legacy;
-pub use tokio;
-pub use tokio_retry;
-pub use tokio_stream;
 pub use url;
 
 /// A prelude that makes all system component traits and data types available.
diff --git a/node/Cargo.toml b/node/Cargo.toml
@@ -36,6 +36,7 @@ graphman = { workspace = true }
 serde = { workspace = true }
 shellexpand = "3.1.1"
 termcolor = "1.4.1"
+tokio = { workspace = true }
 diesel = { workspace = true }
 diesel-async = { workspace = true }
 prometheus = { version = "0.14.0", features = ["push"] }
diff --git a/node/src/chain.rs b/node/src/chain.rs
@@ -26,13 +26,13 @@ use graph::log::factory::LoggerFactory;
 use graph::prelude::anyhow;
 use graph::prelude::MetricsRegistry;
 use graph::slog::{debug, info, o, warn, Logger};
-use graph::tokio::time::timeout;
 use graph::url::Url;
 use graph_chain_ethereum::{self as ethereum, Transport};
 use graph_store_postgres::{BlockStore, ChainHeadUpdateListener};
 use std::cmp::Ordering;
 use std::collections::BTreeMap;
 use std::sync::Arc;
+use tokio::time::timeout;
 
 // The status of a provider that we learned from connecting to it
 #[derive(PartialEq)]
diff --git a/node/src/helpers.rs b/node/src/helpers.rs
@@ -5,12 +5,12 @@ use graph::prelude::{
     BlockPtr, DeploymentHash, NodeId, SubgraphRegistrarError, SubgraphStore as SubgraphStoreTrait,
 };
 use graph::slog::{error, info, Logger};
-use graph::tokio::sync::mpsc::Receiver;
 use graph::{
     components::store::DeploymentLocator,
     prelude::{SubgraphName, SubgraphRegistrar},
 };
 use graph_store_postgres::SubgraphStore;
+use tokio::sync::mpsc::Receiver;
 
 /// Cleanup a subgraph
 /// This is used to remove a subgraph before redeploying it when using the watch flag
diff --git a/node/src/manager/commands/assign.rs b/node/src/manager/commands/assign.rs
@@ -1,8 +1,8 @@
 use graph::components::store::DeploymentLocator;
 use graph::prelude::{anyhow::anyhow, Error, NodeId, StoreEvent};
-use graph::tokio;
 use graph_store_postgres::{command_support::catalog, ConnectionPool, NotificationSender};
 use std::time::Duration;
+use tokio;
 
 use crate::manager::deployment::DeploymentSearch;
 
diff --git a/store/postgres/Cargo.toml b/store/postgres/Cargo.toml
@@ -28,6 +28,8 @@ rand.workspace = true
 serde = { workspace = true }
 serde_json = { workspace = true }
 stable-hash_legacy = { git = "https://github.com/graphprotocol/stable-hash", branch = "old", package = "stable-hash" }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
 anyhow = "1.0.100"
 git-testament = "0.2.6"
 itertools = "0.14.0"
diff --git a/store/postgres/src/copy.rs b/store/postgres/src/copy.rs
diff --git a/store/postgres/src/deployment_store.rs b/store/postgres/src/deployment_store.rs
diff --git a/store/postgres/src/relational.rs b/store/postgres/src/relational.rs
diff --git a/store/postgres/src/store_events.rs b/store/postgres/src/store_events.rs
diff --git a/store/postgres/src/vid_batcher.rs b/store/postgres/src/vid_batcher.rs
diff --git a/store/postgres/src/writable.rs b/store/postgres/src/writable.rs
diff --git a/store/test-store/Cargo.toml b/store/test-store/Cargo.toml
diff --git a/store/test-store/tests/postgres/relational_bytes.rs b/store/test-store/tests/postgres/relational_bytes.rs
