Merge pull request #2456 from eqlabs/chris/graceful-shutdown-2

Implement graceful shutdown
eqlabs · Dec 27, 2024 · a9e38b2 · a9e38b2
2 parents a054a03 + 4b8cbd4
commit a9e38b2
Show file tree

Hide file tree

Showing 89 changed files with 511 additions and 298 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+
+- Graceful shutdown upon SIGINT and SIGTERM with a default grace period of 10 seconds, configurable via `--shutdown.grace-period`.
+
 ### Removed
 
 - `storage_commitment` and `class_commitment` fields from the `pathfinder_subscribe_newHeads` method response.

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -10,7 +10,6 @@ members = [
     "crates/gateway-test-fixtures",
     "crates/gateway-test-utils",
     "crates/gateway-types",
-    "crates/make-stream",
     "crates/merkle-tree",
     "crates/p2p",
     "crates/p2p_proto",
@@ -23,6 +22,7 @@ members = [
     "crates/storage",
     "crates/tagged",
     "crates/tagged-debug-derive",
+    "crates/util",
 ]
 exclude = [
     "crates/load-test",
@@ -137,6 +137,7 @@ tokio = "1.37.0"
 tokio-retry = "0.3.0"
 tokio-stream = "0.1.14"
 tokio-tungstenite = "0.21"
+tokio-util = { version = "0.7.13", features = ["rt"] }
 tower = { version = "0.4.13", default-features = false }
 tower-http = { version = "0.5.2", default-features = false }
 tracing = "0.1.37"

diff --git a/crates/common/Cargo.toml b/crates/common/Cargo.toml
@@ -5,7 +5,6 @@ authors = { workspace = true }
 edition = { workspace = true }
 license = { workspace = true }
 rust-version = { workspace = true }
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
 full-serde = []

diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs
@@ -17,7 +17,6 @@ use serde::{Deserialize, Serialize};
 pub mod casm_class;
 pub mod class_definition;
 pub mod consts;
-pub mod error;
 pub mod event;
 pub mod hash;
 mod header;

diff --git a/crates/compiler/Cargo.toml b/crates/compiler/Cargo.toml
@@ -5,7 +5,6 @@ authors = { workspace = true }
 edition = { workspace = true }
 license = { 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 }

diff --git a/crates/crypto/Cargo.toml b/crates/crypto/Cargo.toml
@@ -5,7 +5,6 @@ authors = { workspace = true }
 edition = { workspace = true }
 license = { workspace = true }
 rust-version = { workspace = true }
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [lib]
 name = "pathfinder_crypto"

diff --git a/crates/ethereum/Cargo.toml b/crates/ethereum/Cargo.toml
@@ -26,4 +26,4 @@ reqwest = { workspace = true, features = ["json"] }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = ["macros"] }
 tracing = { workspace = true }
-
+util = { path = "../util" }
diff --git a/crates/ethereum/src/lib.rs b/crates/ethereum/src/lib.rs
@@ -153,7 +153,8 @@ impl EthereumApi for EthereumClient {
         let provider_clone = provider.clone();
         let (finalized_block_tx, mut finalized_block_rx) =
             tokio::sync::mpsc::channel::<L1BlockNumber>(1);
-        tokio::spawn(async move {
+
+        util::task::spawn(async move {
             let mut interval = tokio::time::interval(poll_interval);
             loop {
                 interval.tick().await;
@@ -165,6 +166,14 @@ impl EthereumApi for EthereumClient {
                     let _ = finalized_block_tx.send(block_number).await.unwrap();
                 }
             }
+            // This it to mitigate the warning: "this function depends on never type
+            // fallback being `()`" as we are unable to implement
+            // [`util::task::FutureOutputExt`] for the never type `!`. Consequently, when
+            // this warning becomes a hard error we should keep this workaround or by then
+            // the `!` type will not be experimental anymore.
+            // Warning related issue: https://github.com/rust-lang/rust/issues/123748
+            #[allow(unreachable_code)]
+            ()
         });
 
         // Process incoming events

diff --git a/crates/executor/Cargo.toml b/crates/executor/Cargo.toml
@@ -5,7 +5,6 @@ authors = { workspace = true }
 edition = { workspace = true }
 license = { 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 }

diff --git a/crates/gateway-client/Cargo.toml b/crates/gateway-client/Cargo.toml
@@ -5,7 +5,6 @@ authors = { workspace = true }
 edition = { workspace = true }
 license = { 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 }

diff --git a/crates/gateway-test-fixtures/Cargo.toml b/crates/gateway-test-fixtures/Cargo.toml
@@ -5,7 +5,6 @@ authors = { workspace = true }
 edition = { workspace = true }
 license = { workspace = true }
 rust-version = { workspace = true }
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 pathfinder-common = { path = "../common" }

diff --git a/crates/gateway-types/Cargo.toml b/crates/gateway-types/Cargo.toml
@@ -5,7 +5,6 @@ authors = { workspace = true }
 edition = { workspace = true }
 license = { 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 }

diff --git a/crates/load-test/Cargo.toml b/crates/load-test/Cargo.toml
@@ -6,7 +6,6 @@ authors = ["Equilibrium Labs <[email protected]>"]
 edition = "2021"
 license = "MIT OR Apache-2.0"
 rust-version = "1.73"
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 goose = "0.17.2"

diff --git a/crates/merkle-tree/Cargo.toml b/crates/merkle-tree/Cargo.toml
@@ -5,7 +5,6 @@ authors = { workspace = true }
 edition = { workspace = true }
 license = { 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 }

diff --git a/crates/p2p/Cargo.toml b/crates/p2p/Cargo.toml
@@ -6,7 +6,6 @@ authors = { workspace = true }
 edition = { workspace = true }
 license = { 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 }
@@ -34,7 +33,6 @@ libp2p = { workspace = true, features = [
     "tokio",
     "yamux",
 ] }
-make-stream = { path = "../make-stream" }
 p2p_proto = { path = "../p2p_proto" }
 p2p_stream = { path = "../p2p_stream" }
 pathfinder-common = { path = "../common" }
@@ -55,6 +53,7 @@ tokio-stream = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
 unsigned-varint = { workspace = true, features = ["futures"] }
+util = { path = "../util" }
 void = { workspace = true }
 zeroize = { workspace = true }
 

diff --git a/crates/p2p/src/client/peer_agnostic.rs b/crates/p2p/src/client/peer_agnostic.rs
@@ -644,7 +644,7 @@ mod header_stream {
 
         tracing::trace!(?start, ?stop, ?dir, "Streaming headers");
 
-        make_stream::from_future(move |tx| async move {
+        util::make_stream::from_future(move |tx| async move {
             // Loop which refreshes peer set once we exhaust it.
             loop {
                 'next_peer: for peer in get_peers().await {
@@ -778,7 +778,7 @@ mod transaction_stream {
     {
         tracing::trace!(?start, ?stop, "Streaming Transactions");
 
-        make_stream::from_future(move |tx| async move {
+        util::make_stream::from_future(move |tx| async move {
             let mut expected_transaction_counts_stream = Box::pin(counts_stream);
 
             let cnt = match try_next(&mut expected_transaction_counts_stream).await {
@@ -961,7 +961,7 @@ mod state_diff_stream {
     {
         tracing::trace!(?start, ?stop, "Streaming state diffs");
 
-        make_stream::from_future(move |tx| async move {
+        util::make_stream::from_future(move |tx| async move {
             let mut length_stream = Box::pin(length_stream);
 
             let cnt = match try_next(&mut length_stream).await {
@@ -1185,7 +1185,7 @@ mod class_definition_stream {
     {
         tracing::trace!(?start, ?stop, "Streaming classes");
 
-        make_stream::from_future(move |tx| async move {
+        util::make_stream::from_future(move |tx| async move {
             let mut declared_class_counts_stream = Box::pin(counts_stream);
 
             let cnt = match try_next(&mut declared_class_counts_stream).await {
@@ -1384,7 +1384,7 @@ mod event_stream {
     {
         tracing::trace!(?start, ?stop, "Streaming events");
 
-        make_stream::from_future(move |tx| async move {
+        util::make_stream::from_future(move |tx| async move {
             let mut counts_stream = Box::pin(counts_stream);
 
             let Some(Ok(cnt)) = counts_stream.next().await else {

diff --git a/crates/pathfinder/Cargo.toml b/crates/pathfinder/Cargo.toml
@@ -5,7 +5,6 @@ authors = { workspace = true }
 edition = { workspace = true }
 license = { workspace = true }
 rust-version = { workspace = true }
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [lib]
 name = "pathfinder_lib"
@@ -30,7 +29,6 @@ futures = { workspace = true, features = ["alloc"] }
 http = { workspace = true }
 ipnet = { workspace = true }
 jemallocator = { workspace = true }
-make-stream = { path = "../make-stream" }
 metrics = { workspace = true }
 metrics-exporter-prometheus = { workspace = true }
 p2p = { path = "../p2p" }
@@ -71,6 +69,7 @@ tracing-subscriber = { workspace = true, features = [
     "ansi",
 ] }
 url = { workspace = true }
+util = { path = "../util" }
 zeroize = { workspace = true }
 zstd = { workspace = true }