feat(node_framework): Add a task to handle sigint (#1471)

## What ❔

Adds a task that handles SIGINT (aka ctrl+c), propagating the signal to
the `ZkStackService` instead of shutting the node down immediately.

## Why ❔

- Graceful shutdown.
- Replicating the current behavior.

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [ ] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [ ] Tests for the changes have been added / updated.
- [ ] Documentation comments have been added / updated.
- [ ] Code has been formatted via `zk fmt` and `zk lint`.
- [ ] Spellcheck has been run via `zk spellcheck`.
- [ ] Linkcheck has been run via `zk linkcheck`.

checks-config/era.dic

@@ -920,3 +920,4 @@ oneshot
 p2p
 StorageProcessor
 StorageMarker
+SIGINT

core/node/node_framework/Cargo.toml

@@ -32,6 +32,7 @@ async-trait = "0.1"
 futures = "0.3"
 anyhow = "1"
 tokio = { version = "1", features = ["rt"] }
+ctrlc = { version = "3.1", features = ["termination"] }
 
 [dev-dependencies]
 zksync_env_config = { path = "../../lib/env_config" }

core/node/node_framework/examples/main_node.rs

@@ -37,6 +37,7 @@ use zksync_node_framework::{
         pools_layer::PoolsLayerBuilder,
         proof_data_handler::ProofDataHandlerLayer,
         query_eth_client::QueryEthClientLayer,
+        sigint::SigintHandlerLayer,
         state_keeper::{
             main_batch_executor::MainBatchExecutorLayer, mempool_io::MempoolIOLayer,
             StateKeeperLayer,
@@ -62,6 +63,11 @@ impl MainNodeBuilder {
         }
     }
 
+    fn add_sigint_handler_layer(mut self) -> anyhow::Result<Self> {
+        self.node.add_layer(SigintHandlerLayer);
+        Ok(self)
+    }
+
     fn add_pools_layer(mut self) -> anyhow::Result<Self> {
         let config = PostgresConfig::from_env()?;
         let pools_layer = PoolsLayerBuilder::empty(config)
@@ -316,6 +322,7 @@ fn main() -> anyhow::Result<()> {
         .build();
 
     MainNodeBuilder::new()
+        .add_sigint_handler_layer()?
         .add_pools_layer()?
         .add_query_eth_client_layer()?
         .add_sequencer_l1_gas_layer()?

core/node/node_framework/src/implementations/layers/mod.rs

@@ -12,5 +12,6 @@ pub mod pools_layer;
 pub mod prometheus_exporter;
 pub mod proof_data_handler;
 pub mod query_eth_client;
+pub mod sigint;
 pub mod state_keeper;
 pub mod web3_api;

core/node/node_framework/src/implementations/layers/sigint.rs

@@ -0,0 +1,60 @@
+use tokio::sync::oneshot;
+
+use crate::{
+    service::{ServiceContext, StopReceiver},
+    task::UnconstrainedTask,
+    wiring_layer::{WiringError, WiringLayer},
+};
+
+/// Layer that changes the handling of SIGINT signal, preventing an immediate shutdown.
+/// Instead, it would propagate the signal to the rest of the node, allowing it to shut down gracefully.
+#[derive(Debug)]
+pub struct SigintHandlerLayer;
+
+#[async_trait::async_trait]
+impl WiringLayer for SigintHandlerLayer {
+    fn layer_name(&self) -> &'static str {
+        "sigint_handler_layer"
+    }
+
+    async fn wire(self: Box<Self>, mut node: ServiceContext<'_>) -> Result<(), WiringError> {
+        // SIGINT may happen at any time, so we must handle it as soon as it happens.
+        node.add_unconstrained_task(Box::new(SigintHandlerTask));
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+struct SigintHandlerTask;
+
+#[async_trait::async_trait]
+impl UnconstrainedTask for SigintHandlerTask {
+    fn name(&self) -> &'static str {
+        "sigint_handler"
+    }
+
+    async fn run_unconstrained(
+        self: Box<Self>,
+        mut stop_receiver: StopReceiver,
+    ) -> anyhow::Result<()> {
+        let (sigint_sender, sigint_receiver) = oneshot::channel();
+        let mut sigint_sender = Some(sigint_sender); // Has to be done this way since `set_handler` requires `FnMut`.
+        ctrlc::set_handler(move || {
+            if let Some(sigint_sender) = sigint_sender.take() {
+                sigint_sender.send(()).ok();
+                // ^ The send fails if `sigint_receiver` is dropped. We're OK with this,
+                // since at this point the node should be stopping anyway, or is not interested
+                // in listening to interrupt signals.
+            }
+        })
+        .expect("Error setting Ctrl+C handler");
+
+        // Wait for either SIGINT or stop signal.
+        tokio::select! {
+            _ = sigint_receiver => {},
+            _ = stop_receiver.0.changed() => {},
+        };
+
+        Ok(())
+    }
+}