This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch update-all-deps-workspace in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 5a31ee5b61ce4e207467bc673126e6e0f632269f Author: Hubert Gruszecki <[email protected]> AuthorDate: Wed Sep 2 11:04:00 2026 +0200 fix(server): log a backtrace and shut down on any shard panic compio's `spawn` catches task panics, and nothing observed them while the server ran: the pump's handle is awaited only at shutdown, and `track_background` reaps listener and connection handles without reading their result. A panicking pump left its partitions stalled until an operator noticed. A panicking listener or connection task left the server running without it and the eventual Ctrl-C exited 0. The thread-level path (`ShutdownOnDrop`) covered only panics that unwound out of `run_shard_thread`, and no path logged a backtrace. Install a process-wide panic hook in `bootstrap` before the first shard thread spawns. It logs the panic with thread, location and a forced backtrace through tracing, records the first one, and flips the shared shutdown flag so every shard drains the same way a thread-level failure already does. `join_all` fails the exit with `ServerError::Panicked` when every thread exited `Ok` but a panic was recorded, so an orchestrator never reads a panic as a clean stop. Graceful drain rather than abort keeps the pump's final journal flush on the non-pump panic paths. A pump panic still surfaces as `ShardPumpDied` through the existing drain verdict. --- core/server/src/boot/mod.rs | 9 +++- core/server/src/boot/threads.rs | 113 ++++++++++++++++++++++++++++++++++++---- core/server/src/server_error.rs | 7 +++ 3 files changed, 117 insertions(+), 12 deletions(-) diff --git a/core/server/src/boot/mod.rs b/core/server/src/boot/mod.rs index b8a5ed9a3..841b3e3ab 100644 --- a/core/server/src/boot/mod.rs +++ b/core/server/src/boot/mod.rs @@ -50,8 +50,9 @@ use crate::boot::recovery::{ RecoveredOwnerState, build_shard_for_thread, restore_metadata_consensus, }; use crate::boot::threads::{ - StopSignals, await_pump_drain, join_partial_shard_survivors, resolve_shard_assignments, - run_shard_thread, spawn_shutdown_watchdog, validate_sharding_runtime_knobs, + StopSignals, await_pump_drain, install_panic_hook, join_partial_shard_survivors, + resolve_shard_assignments, run_shard_thread, spawn_shutdown_watchdog, + validate_sharding_runtime_knobs, }; use crate::boot::topology::{RosterCells, resolve_tcp_topology}; use crate::dispatch::partition::make_partition_read_handler; @@ -272,6 +273,9 @@ pub fn bootstrap( let (senders, mut inboxes, mut reply_inboxes) = shard_mesh_channels(total_shards, inbox_capacity, reply_inbox_capacity); let shutdown_flag = Arc::new(AtomicBool::new(false)); + // Before the first shard thread exists, so no panic on a shard, in the + // thread body or in a task compio's `spawn` would swallow, escapes it. + let first_panic = install_panic_hook(Arc::clone(&shutdown_flag)); let config = Arc::new(config); // One owner table per server process, Arc-cloned into every shard's bus so // any shard's bus reads the same atomic slots that the owning @@ -403,6 +407,7 @@ pub fn bootstrap( shutdown_flag, shard_threads, join_timeout: config.system.sharding.shutdown_join_timeout.get_duration(), + first_panic, }) } diff --git a/core/server/src/boot/threads.rs b/core/server/src/boot/threads.rs index 95de1d5b2..28a209a4c 100644 --- a/core/server/src/boot/threads.rs +++ b/core/server/src/boot/threads.rs @@ -32,23 +32,26 @@ use partitions::FatalCommit; use server_common::executor::create_shard_executor; use shard::metrics::ShardMetrics; use shard::{Receiver as ShardReceiver, Sender, ShardFrame, TaggedSender}; +use std::backtrace::Backtrace; use std::rc::Rc; -use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, OnceLock}; use std::time::{Duration, Instant}; use std::{panic, thread}; use tracing::{error, info, warn}; /// Result of a multi-shard bootstrap. /// -/// Carries the cross-thread shutdown flag and one OS-thread `JoinHandle` -/// per shard. The caller flips the flag via [`Self::install_ctrlc_handler`] -/// and then drains every shard via [`Self::join_all`], bounded by -/// `join_timeout` (`system.sharding.shutdown_join_timeout`). +/// Carries the cross-thread shutdown flag, one OS-thread `JoinHandle` +/// per shard, and the first panic [`install_panic_hook`] recorded. The +/// caller flips the flag via [`Self::install_ctrlc_handler`] and then +/// drains every shard via [`Self::join_all`], bounded by `join_timeout` +/// (`system.sharding.shutdown_join_timeout`). pub struct ShardHandles { pub(in crate::boot) shutdown_flag: Arc<AtomicBool>, pub(in crate::boot) shard_threads: Vec<(u16, thread::JoinHandle<Result<(), ServerError>>)>, pub(in crate::boot) join_timeout: Duration, + pub(in crate::boot) first_panic: Arc<OnceLock<String>>, } impl ShardHandles { @@ -98,7 +101,9 @@ impl ShardHandles { /// returned a `Result::Err`, panicked, or wedged past the deadline. /// The variant carries every per-shard failure in shard-id order so /// the caller does not need to read the trace log to discover - /// late-failing shards. + /// late-failing shards. Returns [`ServerError::Panicked`] when every + /// thread exited `Ok` but the panic hook recorded a panic: a task + /// compio's `spawn` caught, which no thread result can carry. pub fn join_all(self) -> Result<(), ServerError> { let mut failures: Vec<ShardJoinFailure> = Vec::new(); // Armed on the first poll that observes the shutdown flag, shared @@ -157,14 +162,55 @@ impl ShardHandles { } } } - if failures.is_empty() { - Ok(()) - } else { - Err(ServerError::ShardJoinFailures { failures }) + if !failures.is_empty() { + return Err(ServerError::ShardJoinFailures { failures }); } + self.first_panic.get().map_or_else( + || Ok(()), + |description| { + Err(ServerError::Panicked { + description: description.clone(), + }) + }, + ) } } +/// Install the process-wide panic hook and return the slot it records +/// the first panic into. +/// +/// The hook runs on the panicking thread before the unwind, so it sees +/// every panic on a shard: the thread body, whose unwind +/// [`run_shard_thread`] already turns into a join failure, and the tasks +/// compio's `spawn` catches, which nothing observes while the server +/// runs (a dead listener or connection task leaves every thread exiting +/// `Ok`). It logs the panic with its backtrace, records the first one for +/// [`ShardHandles::join_all`] to fail the exit on, and flips the shutdown +/// flag so every shard drains instead of the half-alive state one dead +/// task leaves behind. The previous hook still runs after it, so stderr +/// keeps the standard panic line. +pub(in crate::boot) fn install_panic_hook(shutdown_flag: Arc<AtomicBool>) -> Arc<OnceLock<String>> { + let first_panic = Arc::new(OnceLock::new()); + let recorded = Arc::clone(&first_panic); + let previous_hook = panic::take_hook(); + panic::set_hook(Box::new(move |info| { + let current = thread::current(); + let thread_name = current.name().unwrap_or("<unnamed>"); + let location = info + .location() + .map_or_else(|| "<unknown>".to_string(), ToString::to_string); + let message = panic_payload_to_string(info.payload()); + let backtrace = Backtrace::force_capture(); + error!(thread = thread_name, location = %location, backtrace = %backtrace, "{message}"); + let _ = recorded.set(format!( + "thread '{thread_name}' panicked at {location}: {message}" + )); + shutdown_flag.store(true, Ordering::Relaxed); + previous_hook(info); + })); + first_panic +} + /// Poll cadence for the bounded shard joins. Coarse enough to cost /// nothing during a normal drain, fine enough that exit latency past /// the last shard's return stays imperceptible. @@ -682,6 +728,53 @@ mod tests { ); } + #[test] + fn join_all_fails_the_exit_on_a_panic_no_thread_surfaced() { + // A listener or connection task panic leaves every shard thread + // exiting Ok; only the hook's record can keep that from reading + // as a clean shutdown to an orchestrator. + let first_panic = Arc::new(OnceLock::new()); + first_panic + .set("thread 'shard-0' panicked at listener.rs:1:1: boom".to_string()) + .expect("a fresh slot accepts the first record"); + let handle = thread::spawn(|| -> Result<(), ServerError> { Ok(()) }); + let handles = ShardHandles { + shutdown_flag: Arc::new(AtomicBool::new(true)), + shard_threads: vec![(0, handle)], + join_timeout: Duration::from_secs(1), + first_panic, + }; + let error = handles + .join_all() + .expect_err("a recorded panic must fail the exit even when every thread exited Ok"); + assert!( + matches!(&error, ServerError::Panicked { description } if description.contains("boom")), + "unexpected error: {error}" + ); + } + + #[test] + fn panic_hook_records_the_panic_and_flips_the_shutdown_flag() { + let shutdown_flag = Arc::new(AtomicBool::new(false)); + let first_panic = install_panic_hook(Arc::clone(&shutdown_flag)); + let joined = thread::Builder::new() + .name("shard-7".to_string()) + .spawn(|| panic!("injected task panic")) + .expect("spawn") + .join(); + assert!(joined.is_err(), "the thread must have panicked"); + assert!( + shutdown_flag.load(Ordering::Relaxed), + "a panic anywhere must drive the whole server down" + ); + let description = first_panic.get().expect("the hook records the first panic"); + assert!( + description.starts_with("thread 'shard-7' panicked at ") + && description.ends_with(": injected task panic"), + "unexpected record: {description}" + ); + } + #[test] fn join_abandons_a_wedged_shard_after_the_shutdown_deadline() { let shutdown_flag = AtomicBool::new(true); diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs index 12438e710..7f0f86c5e 100644 --- a/core/server/src/server_error.rs +++ b/core/server/src/server_error.rs @@ -292,6 +292,13 @@ pub enum ServerError { ShardConstruction(#[source] ShardCtorError), #[error("{} shard thread(s) failed: {}", failures.len(), format_shard_failures(failures))] ShardJoinFailures { failures: Vec<ShardJoinFailure> }, + /// A panic no shard thread could surface: compio's `spawn` catches task + /// panics, so a dead listener or connection task leaves every thread + /// exiting `Ok`. The panic hook records the first one and the join path + /// fails the exit on it, so an orchestrator does not read the shutdown + /// as clean. + #[error("server shut down after a panic: {description}")] + Panicked { description: String }, } /// Why a partition's recovered segments cannot be served.
