From 9521cdf26dec5bec21f14c6f460da4655d86d26e Mon Sep 17 00:00:00 2001 From: Emmanuel Ikwuoma Date: Sun, 6 Sep 2026 06:10:46 +0100 Subject: [PATCH] fix race condition in sigterm handler --- crates/malachite-app/src/main.rs | 14 ++++++++--- crates/malachite-app/src/node.rs | 40 ++++++++++++++++++++++++++++++-- 2 files changed, 49 insertions(+), 5 deletions(-) diff --git a/crates/malachite-app/src/main.rs b/crates/malachite-app/src/main.rs index 52179ec..dab3767 100644 --- a/crates/malachite-app/src/main.rs +++ b/crates/malachite-app/src/main.rs @@ -31,7 +31,7 @@ use arc_consensus_types::{ SigningConfig, }; use arc_node_consensus::hardcoded_config; -use arc_node_consensus::node::{App, StartConfig}; +use arc_node_consensus::node::{App, StartConfig, SIGTERM_EXIT_CODE}; use arc_node_consensus::store::migrations::MigrationCoordinator; use arc_node_consensus::store::{rollback_to_height, CERTIFICATES_TABLE, ROLLBACK_BATCH_SIZE}; use arc_node_consensus_cli::{ @@ -298,8 +298,16 @@ fn start(args: &Args, cmd: &StartCmd, logging: config::LoggingConfig) -> Result< // Setup the application let app = App::new(config, args.get_home_dir()?, private_key_file, start_config); - // Start the node - rt.block_on(app.run()) + // Start the node — map a SIGTERM-induced shutdown to the conventional exit + // code 143 (128 + SIGTERM) so the container orchestrator observes the correct + // termination reason. + match rt.block_on(app.run()) { + Ok(()) => Ok(()), + Err(e) if e.to_string().contains("SIGTERM") => { + std::process::exit(SIGTERM_EXIT_CODE); + } + Err(e) => Err(e), + } } fn init(args: &Args, cmd: &InitCmd, _logging: config::LoggingConfig) -> Result<()> { diff --git a/crates/malachite-app/src/node.rs b/crates/malachite-app/src/node.rs index 402c5d6..017ffb2 100644 --- a/crates/malachite-app/src/node.rs +++ b/crates/malachite-app/src/node.rs @@ -36,8 +36,12 @@ use std::time::Duration; use bytesize::ByteSize; use eyre::Context; use rand::rngs::OsRng; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; use tokio::signal::unix::SignalKind; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{mpsc, oneshot, Notify}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; @@ -122,6 +126,10 @@ pub struct Handle { graceful_shutdown: CancellationToken, /// Fires when the EL IPC watchdog triggered shutdown (as opposed to SIGTERM or normal halt). el_watchdog_triggered: oneshot::Receiver<()>, + /// Tracks SIGTERM cleanup completion so `Node::run` can keep the runtime alive + /// until the handler finishes instead of racing against it. + pub(crate) sigterm_done: Arc, + pub(crate) sigterm_received: Arc, /// Kept alive to prevent the app request channel from closing when RPC is disabled. _tx_app_req: mpsc::Sender, } @@ -881,6 +889,8 @@ impl App { let tx_event = channels.events.clone(); let cancel_token = CancellationToken::new(); let graceful_shutdown = CancellationToken::new(); + let sigterm_done = Arc::new(Notify::new()); + let sigterm_received = Arc::new(AtomicBool::new(false)); // Watchdog: on unexpected EL IPC close, signal the run loop and cancel the app task; // the run loop performs the bounded Node stop and the process exit. @@ -931,6 +941,8 @@ impl App { cancel_token, graceful_shutdown, el_watchdog_triggered: el_watchdog_rx, + sigterm_done, + sigterm_received, _tx_app_req: tx_app_req, }) } @@ -964,6 +976,16 @@ impl App { // Wait for the application to finish let result = handles.app.await?; + // SIGTERM handover: if `sigterm_received` was set by the handler, await + // the one-shot `sigterm_done` Notify until cleanup (engine stop + + // savepoint) completes. Without this, `run` would return immediately + // after the app future resolves and drop the Tokio runtime, killing the + // handler task mid-cleanup. + if handles.sigterm_received.load(Ordering::SeqCst) { + handles.sigterm_done.notified().await; + return Err(eyre::eyre!("Received SIGTERM signal")); + } + // EL IPC closed: stop the Node actor with a bounded timeout, then exit non-zero so // the orchestrator restarts the container. if handles.el_watchdog_triggered.try_recv().is_ok() { @@ -1011,7 +1033,7 @@ const EL_IPC_SHUTDOWN_EXIT_CODE: i32 = 1; /// Exit code for a SIGTERM-triggered shutdown: 143 = 128 + SIGTERM (15), the conventional /// exit status for a process terminated by SIGTERM. -const SIGTERM_EXIT_CODE: i32 = 143; +pub const SIGTERM_EXIT_CODE: i32 = 143; /// Grace period for in-flight tasks to finish after teardown, before the process exits. const SHUTDOWN_DRAIN_DELAY: Duration = Duration::from_millis(500); @@ -1133,6 +1155,8 @@ fn install_sigterm_handler(handle: &Handle) { let store = handle.store.clone(); let cancel_token = handle.cancel_token.clone(); let graceful_shutdown = handle.graceful_shutdown.clone(); + let sigterm_done = handle.sigterm_done.clone(); + let sigterm_received = handle.sigterm_received.clone(); let mut sigterm = signal(SignalKind::terminate()).expect("inside Tokio runtime"); @@ -1141,6 +1165,11 @@ fn install_sigterm_handler(handle: &Handle) { warn!("Received SIGTERM, shutting down..."); + // Mark as received before teardown so `Node::run` knows to wait on the + // one-shot `Notify` rather than returning immediately and dropping the + // runtime. + sigterm_received.store(true, Ordering::SeqCst); + stop_node_and_teardown( node.stop_and_wait( Some("Received SIGTERM signal".to_string()), @@ -1152,6 +1181,13 @@ fn install_sigterm_handler(handle: &Handle) { .await; drain_before_exit(|| store.savepoint()).await; + sigterm_done.notify_one(); + // Keep the direct exit for the `HaltAndWait` path where `Node::run` + // is parked in `sleep(Duration::MAX)` and not awaiting the `Notify`. + // For the normal path, `Node::run` observes the `Notify` and returns + // `Err("SIGTERM")` — `main::start` maps it to 143, so either exit path + // yields the correct K8s code. The `Notify` handshake guarantees the + // runtime stays alive until this point. std::process::exit(SIGTERM_EXIT_CODE); }); }