diff --git a/docs/backlogs/000174-atomic-index-metadata-publication-and-panic-safe-shutdown.md b/docs/backlogs/closed/000174-atomic-index-metadata-publication-and-panic-safe-shutdown.md similarity index 95% rename from docs/backlogs/000174-atomic-index-metadata-publication-and-panic-safe-shutdown.md rename to docs/backlogs/closed/000174-atomic-index-metadata-publication-and-panic-safe-shutdown.md index d72caf06..06a4ef7d 100644 --- a/docs/backlogs/000174-atomic-index-metadata-publication-and-panic-safe-shutdown.md +++ b/docs/backlogs/closed/000174-atomic-index-metadata-publication-and-panic-safe-shutdown.md @@ -126,3 +126,11 @@ When a backlog item is moved to `docs/backlogs/closed/`, append: - Reference: - Closed At: ``` + +## Close Reason + +- Type: implemented +- Detail: Implemented by tasks 000250 and 000256: task 000250 completed atomic index metadata publication, and task 000256 completed panic-contained component shutdown. +- Closed By: backlog close +- Reference: User decision +- Closed At: 2026-08-05 diff --git a/docs/engine-component-lifetime.md b/docs/engine-component-lifetime.md index 3668d514..2c6ef426 100644 --- a/docs/engine-component-lifetime.md +++ b/docs/engine-component-lifetime.md @@ -347,8 +347,79 @@ Normal shutdown is: its local event, and repeat from the first current blocker 6. remove idle registry-owned sessions 7. call `ComponentRegistry::shutdown_all()` in reverse registration order; - redo stops before internal mandatory admission drains, and purge stops last -8. mark lifecycle state as `Shutdown` + redo stops before internal mandatory admission drains, purge stops last, and + each hook is independently panic-contained so later hooks still run +8. mark lifecycle state as `Shutdown`, release the owner shutdown mutex, report + the aggregate outcome, and only then propagate or suppress its first payload + +### Panic-contained shutdown + +Component shutdown has a narrow terminal panic-containment contract. The +registry catches each hook with `catch_unwind(AssertUnwindSafe(...))`, reports +every panic, marks that exact owner suspect, retains the first original payload, +and continues in exact reverse registration order. Every hook is invoked at +most once. A repeated registry or engine shutdown returns an empty/already +complete outcome and never replays a payload. + +`AssertUnwindSafe` is justified only because the graph becomes terminal and is +never exposed for recovery or reuse. It does not state that the storage engine, +transaction system, or worker mutation bodies implement `UnwindSafe` or +`RefUnwindSafe`. An active hook must close ingress and signal its workers before +a deliberate catchable panic point. A multi-worker hook must attempt every +join and required infallible release before exposing the first payload. +Bootstrap rollback establishes its own local preconditions because the normal +engine foreground drain does not yet exist. + +After dispatch, the engine publishes `Shutdown` and releases the shutdown mutex +before applying the aggregate policy. An explicit caller on a non-unwinding +thread receives the first original payload through `resume_unwind`. If owner +drop is already running during another unwind, the payload is reported and +forgotten so teardown does not introduce a second panic. Later payloads are +reported and forgotten without running arbitrary payload destructors. Either +case is terminal: callers must not recover or reuse the in-memory component +graph after any contained hook panic. + +The complete reverse-order shutdown audit is: + +| Reverse order | Component | Shutdown authority and panic caveat | +| ---: | --- | --- | +| 1 | `TransactionRedoWorkers` | Closes group commit, queues the shutdown marker, joins the log thread, releases the active log file, then exposes a captured join payload. Arbitrary redo-body unwind is not repaired. | +| 2 | `MandatoryRuntimeWorkers` | Closes caller/internal admission, records caller-drain validation, drains internal work, signals stop, joins every runner, validates the executor, then exposes the first invariant or join payload. Accepted task bodies retain their domain supervision. | +| 3 | `TransactionPurgeWorkers` | Sends `Purge::Stop`, joins the dispatcher and every executor, and retains an explicit transaction-system guard so degraded leakage pins the dependency closure. Arbitrary mid-purge unwind remains unsupported. | +| 4 | `TransactionSystem` | Passive hook. Redo, mandatory-runtime, and purge worker owners hold active shutdown authority; transaction state is terminal after a worker panic. | +| 5 | `Catalog` | Passive hook. Purge stops before owner release, and foreground catalog users were drained before component dispatch. | +| 6 | `LockManager` | Passive hook. The session/operation drain removes its users. | +| 7 | `SharedPoolEvictorWorkers` | Sets the shutdown flag, signals every pool, wakes the worker, and then joins. Join propagation follows all stop signalling; arbitrary eviction-body unwind is not repaired. | +| 8 | `FileSystemWorkers` | Closes every I/O ingress lane, drains the worker, and then joins. Arbitrary I/O-body unwind is not repaired. | +| 9 | `MemPool` | Passive hook. Shared evictor and I/O worker components own active shutdown. | +| 10 | `IndexPool` | Passive hook with the same split authority as `MemPool`. | +| 11 | `MetaPool` | Passive owner with no worker; release follows catalog and transaction guard teardown. | +| 12 | `DiskPool` | Passive hook. The shared evictor stops earlier in reverse order. | +| 13 | `FileSystem` | Passive hook. `FileSystemWorkers` owns active I/O shutdown and retains this dependency. | +| 14 | `MandatoryRuntime` | Passive hook. `MandatoryRuntimeWorkers` owns admission drain, stop, and joins. | +| 15 | `EnginePoisoner` | Passive hook. It remains available through components that may report fatal state. | +| 16 | `StorageRootLease` | Takes and drops the lock file last, so root ownership brackets subordinate storage activity even after a contained earlier panic. | + +Any new production component, dependency edge, or panic-capable shutdown +operation must update this inventory and its adjacent `Panic safety:` comment. + +The purge position also closes the CTS/STS boundary used by containment. +Foreground sessions and operations are gone before component hooks. Redo joins +before purge stop, so no later ordered commit producer can hand off a committed +payload, and mandatory internal work drains before purge. `Purge::Stop` is a +terminal queue barrier: already observed messages are absorbed, while pending +committed payloads may remain owned by GC buckets without requiring physical +reclamation during shutdown. After purge joins, no later hook reads CTS, STS, +GC buckets, row undo, retained roots, metadata history, or dropped-table state. + +`published_gc_horizon` records a fresh active-bucket scan and does not claim +physical purge. `global_visible_sts` advances only after all selected bucket, +retirement, retained-root, metadata-history, and dropped-table work for a +complete cycle succeeds. A failed join proves worker termination, not unwind +safety for an arbitrary in-progress purge mutation. The deterministic shutdown +fault occurs at the named-worker `Finished("Purge-Dispatcher")` observer after +the body returns; it does not exercise a mid-`purge_trx_list_inner` unwind or +the `CommittedTrx`/raw-`RowUndoRef` ownership limitation. `Engine::try_shutdown()` uses the same first-blocker traversal without installing an event or listener. It queues at most that blocker's cleanup hint @@ -440,6 +511,25 @@ Dropping `EngineInner` first releases `EngineCore` and its runtime-held quiescent guards before registry-owned component owners start their final `QuiescentBox` drains. +Panic-free registry drop keeps the strict behavior: it clears published access +handles and drops owners in reverse order, allowing `QuiescentBox` to wait and +therefore expose hidden guard-lifetime defects. If any hook panicked, registry +drop uses a separate degraded policy after owner-side reachability is gone. A +suspect owner is intentionally leaked. An independent non-suspect owner with a +zero sampled guard count is dropped normally. A non-suspect owner with +outstanding guards is also leaked, allowing retained dependency guards to +produce a bounded closure rather than a hang or use-after-free. Each leak is +reported with component name, reason (`shutdown_panic` or +`outstanding_guards`), and the acquire-ordered guard-count sample. + +The bounded unit is the suspect component plus owners pinned through its +quiescent dependency closure for one failed engine. Independent owners, +including the root lease owner after its active release hook, remain +reclaimable. Builder rollback clears transient shelf provisions before this +policy or payload propagation because provisions may retain quiescent guards. +This protects teardown-owned allocations only; it cannot restore memory that an +arbitrary worker body already released while unwinding. + `Engine::drop` invokes the same synchronous drain as `Engine::shutdown()`. An unintended owner drop can therefore block indefinitely while caller-retained foreground operations, observers, mandatory work, or @@ -459,10 +549,14 @@ for `QuiescentGuard`. The contract is: - owner allocation address stays stable for the full guard lifetime - guard acquisition is one atomic increment - guard release is one atomic decrement -- owner drop blocks until the outstanding guard count reaches zero - -The current contract is still purely blocking owner drop. There is no local -timeout or diagnostic hook in the runtime. +- normal owner drop blocks until the outstanding guard count reaches zero +- terminal degraded registry release may sample the count with acquire + ordering and intentionally leak an owner instead of entering that wait + +The sample is valid only after registry access handles, engine-core handles, and +builder shelf provisions are gone. At that point zero cannot increase because +no guard remains from which another guard can be cloned. Ordinary runtime code +does not use the observation and there is no forced timeout or cancellation. ## Pool Guard Provenance diff --git a/docs/tasks/000256-component-shutdown-panic-containment.md b/docs/tasks/000256-component-shutdown-panic-containment.md new file mode 100644 index 00000000..1b86e98c --- /dev/null +++ b/docs/tasks/000256-component-shutdown-panic-containment.md @@ -0,0 +1,271 @@ +--- +id: 000256 +title: Contain Component Shutdown Panics +status: implemented # proposal | implemented | superseded +created: 2026-08-05 +github_issue: 942 +--- + +# Task: Contain Component Shutdown Panics + +## Summary + +Component shutdown now contains each catchable hook panic independently, +continues exact reverse-order teardown, and makes the engine terminal before +propagating the first original payload. Repeated shutdown is a no-op, and owner +drop during an existing unwind suppresses the retained payload instead of +causing a double-panic abort. + +Purge, mandatory-runtime, and redo worker owners complete their safe terminal +work before exposing a captured join panic. If a hook panics, registry owner +release avoids an indefinite quiescent wait by leaking only the suspect owner +and owners still pinned by its guard closure; independent owners continue to be +reclaimed. + +This is a narrow panic-contained shutdown contract. It does not make arbitrary +redo, purge, eviction, buffer, or I/O mutation bodies unwind-safe, and an engine +that encountered a shutdown panic cannot be recovered or reused. + +## Context + +`Issue Labels:` +`- type:task` +`- priority:high` +`- codex` + +`Source Backlogs:` +`- docs/backlogs/closed/000174-atomic-index-metadata-publication-and-panic-safe-shutdown.md` + +`Related RFCs:` +`- docs/rfcs/0026-engine-owned-mandatory-background-runtime.md` + +`Related Tasks:` +`- docs/tasks/000212-introduce-observability-logging.md` +`- docs/tasks/000250-runtime-owned-index-ddl.md` +`- docs/tasks/000252-mandatory-runtime-lifecycle-fairness-evolution-readiness.md` +`- docs/tasks/000254-remove-engine-runtime-reference-accounting.md` + +Backlog 000174 combined an index-metadata publication race with shutdown panic +containment. Task 000250 completed the publication half; this task completed +the remaining shutdown half. + +Before this change, the first panicking component hook aborted registry +dispatch after the registry-wide once flag had been set. Later active hooks +could therefore remain unexecuted, and registry owner drop could block forever +on quiescent guards retained by their workers. Purge and mandatory-runtime +owners also stopped at the first failed join, while redo could skip final +log-file release. + +Normal engine shutdown already provides the containment boundary: it closes +foreground and mandatory admissions, drains sessions and accepted work, then +stops redo, mandatory workers, purge, eviction, and shared I/O in dependency +order. That terminal boundary permits registry-level `AssertUnwindSafe`; it is +not a proof that component domain mutations are generally `UnwindSafe`. + +The concrete unsupported case remains purge mutation. A local +`Vec` owns boxed row undo while reachable row-version chains hold +non-owning `RowUndoRef` links. An arbitrary mid-mutation unwind can invalidate +that ownership relationship, so this task injects panics only at named-worker +finish after the worker body has returned. + +## Goals + +1. Invoke every registered shutdown hook at most once and in exact reverse + registration order, even when earlier hooks panic. +2. Report every hook or worker-join panic while retaining only the first + original payload for propagation. +3. Publish terminal engine lifecycle state and finish required cleanup before + propagation, or suppress propagation during an existing unwind. +4. Join all purge and mandatory-runtime workers and release redo resources + before exposing safe post-stop failures. +5. Preserve live mandatory runners when accepted callers have not drained. +6. Prevent degraded owner release from hanging or reclaiming allocations still + reachable through quiescent guards. +7. Preserve normal panic-free shutdown, full owner reclamation, component + order, CTS/STS semantics, and root-lease-last behavior. +8. Keep the complete production component audit and arbitrary-worker-unwind + limitation in durable lifecycle documentation and adjacent code comments. + +## Non-Goals + +1. General `UnwindSafe` or `RefUnwindSafe` support for the storage engine. +2. Recovery, restart, or reuse of a component graph after a shutdown panic. +3. Repair of arbitrary mid-body panics in redo, purge, eviction, buffer, or + kernel I/O state machines. +4. Forced cancellation, worker termination, watchdogs, timeouts, deadlock + recovery, or process-abort policy. +5. Changes to public shutdown APIs, component registration order, CTS/STS + semantics, persisted formats, recovery, or transaction visibility. +6. Rework of the index-publication half already completed by task 000250. + +## Plan + +### Aggregate once-only component shutdown + +`ComponentRegistry::shutdown_all` returns a must-use +`ComponentShutdownOutcome`. The existing atomic remains the once-only dispatch +gate. Each erased hook runs through `catch_unwind(AssertUnwindSafe(...))`; +success and panic are logged per component, the exact owner is marked suspect, +and dispatch continues. + +`FirstPanic` retains the first original payload. Later payloads are described +without consuming them and then forgotten so an arbitrary payload destructor +cannot start another unwind on the teardown path. A repeated registry shutdown +returns an empty completed outcome and neither reruns hooks nor replays a +payload. + +### Terminal engine propagation policy + +Explicit shutdown, try-shutdown, and owner drop all receive the aggregate +outcome after idle session state and component dispatch finish. They publish +`EngineLifecycleState::Shutdown`, release the shutdown mutex, and log the final +result before applying the payload policy. + +On a non-unwinding thread, the first payload is resumed unchanged. During an +existing unwind it is reported and forgotten. Builder rollback uses the same +policy only after clearing shelf provisions that may retain guards into +registered owners. + +### Active worker invariants + +- Purge sends `Purge::Stop`, takes the handle vector once, attempts every + dispatcher/executor join, and then resumes the first payload. Its owner holds + an explicit transaction-system guard so degraded leakage pins the required + dependency graph. +- Mandatory-runtime first closes caller admission and checks that accepted + callers are drained. If callers remain, it reports the invariant and leaves + internal admission and runners live. Once drained, it closes and drains + internal admission, signals stop, joins every runner, validates the executor, + and then resumes the first collected join or terminal-invariant payload. +- Redo closes group-commit admission, queues shutdown, joins the log thread, + releases the active log file, and only then resumes a captured join payload. +- Shared evictor and filesystem hooks retain their established signal/ingress + closure before join ordering. Their join payloads are contained by the + registry only after those terminal actions. +- Startup rollback joins every started worker while preserving the typed + startup error as the primary diagnostic. + +### Guard-aware degraded owner release + +Panic-free registry drop remains strict: clear published access handles and +drop owners in reverse order, allowing `QuiescentBox` to wait and expose hidden +guard-lifetime defects. + +After any hook panic, the registry clears published access handles, then +samples each owner's quiescent guard count with acquire ordering. A suspect +owner is leaked regardless of count. A non-suspect owner with outstanding +guards is also leaked, allowing retained dependency guards to form a bounded +leak closure. A zero-guard independent owner is dropped normally. Every leak +records component name, reason, and observed count. + +The zero sample is valid only after registry handles, engine-core handles, and +builder shelf provisions are gone; no remaining guard exists from which the +count can increase. This protects teardown-owned allocations but cannot restore +memory already released by an arbitrary worker-body unwind. + +### Shutdown order and transaction boundary + +The reverse production order remains: redo workers, mandatory-runtime workers, +purge workers, transaction system, catalog, lock manager, evictor workers, +filesystem workers, memory/index/metadata/disk pools, filesystem, +mandatory runtime, poisoner, and storage root lease. + +Foreground work drains before dispatch. Redo joins before purge stop, mandatory +internal work drains before purge, and `Purge::Stop` is a terminal queue +barrier. `published_gc_horizon` remains scan progress rather than physical +purge; `global_visible_sts` advances only after a complete successful cycle. +After purge joins, no later hook reads transaction purge state. The full +sixteen-component authority and panic audit lives in +`docs/engine-component-lifetime.md`. + +## Implementation Notes + +Implemented panic-contained component shutdown across the registry, engine +lifecycle, active worker owners, quiescent owner release, observability, and +lifecycle documentation. The first payload now reaches an explicit caller only +after all hooks ran, lifecycle state became terminal, and the shutdown lock was +released; shutdown during an outer panic completes without aborting. + +The degraded release implementation records per-owner shutdown failure and a +registry-wide degraded bit. Tests proved that the suspect owner and its +outstanding-guard dependency are leaked while an independent zero-guard owner +is reclaimed. Builder rollback clears shelf-held guards before degraded drop +or payload propagation. + +Purge now joins all worker handles and retains a transaction-system dependency. +Redo releases its active log file after a failed join. Mandatory-runtime joins +all runners only after caller admission is confirmed drained, then performs +executor validation before propagation. + +Review found that the original proposal would stop mandatory runners even when +accepted callers remained. Commit `a001a71` corrected the implementation and +added regression coverage: active callers now produce the existing invariant +panic without closing internal admission, signalling stop, or taking runner +handles. No unresolved, current PR review threads remained after that fix. + +Two documentation details differed from the proposal: + +- the CTS/STS ordering and arbitrary purge-unwind limitation were consolidated + in `docs/engine-component-lifetime.md` and the purge ownership comment rather + than duplicating them in `docs/transaction-system.md`; +- evictor and filesystem coverage uses the end-to-end engine shutdown test, + while the hooks retain their existing local signal-before-join behavior. + +Verification on PR 943 at current head `a001a71` passed workspace nextest +coverage, default and libaio Clippy, libaio nextest, Codecov project/patch +checks, and the aggregate CI verification job. The resolve-time branch-diff +style gate also passed formatting, Clippy, and repository style checks for all +13 changed Rust files. Follow-up automated review reported no blocking issue. + +Source backlog 000174 is fully implemented by tasks 000250 and 000256. No new +bounded deferred work was discovered; the broader arbitrary-worker-body unwind +problem remains an explicit non-goal rather than an underspecified backlog. + +## Impacts + +- Internal component shutdown now has an aggregate outcome, per-hook + containment, first-payload retention, and degraded owner-release state. +- Engine lifecycle reporting distinguishes successful and panic-degraded + shutdown and applies propagation only after terminal publication. +- Purge, mandatory-runtime, redo, eviction, filesystem, passive component, and + root-lease shutdown authority is documented next to the implementation. +- Quiescent owners expose a narrow acquire-ordered guard-count observation used + only for terminal degraded release. +- Structured logs identify each hook panic, worker join panic, propagation or + suppression decision, final engine result, and intentionally leaked owner. +- Lifecycle documentation records the full component audit, CTS/STS shutdown + boundary, terminal/no-reuse rule, and bounded leak policy. +- No public API, configuration, persisted format, recovery compatibility, + registration order, or benchmark interface changed. + +## Test Cases + +Completed coverage includes: + +1. Exact reverse hook order and reverse owner release on normal shutdown. +2. Independent containment of multiple hook panics, first-payload preservation, + secondary-payload forgetting, and once-only repeated shutdown. +3. Payload suppression during an existing unwind without double panic. +4. Degraded release of a suspect owner, retained-guard dependency closure, and + independent zero-guard owner; shelf guards are cleared before builder drop. +5. Purge and mandatory-runtime multi-worker failure paths attempt every safe + join and preserve the first payload. +6. Mandatory-runtime leaves internal admission, stop state, and runner handles + intact while accepted callers remain. +7. Redo releases the active log file before propagating a finish panic. +8. End-to-end purge-finish injection proves redo and mandatory workers finish + first, purge executors join, evictor and I/O hooks still run, lifecycle is + terminal, repeated shutdown is inert, and a replacement engine reacquires + the same storage root. +9. Engine owner drop during an outer panic suppresses the shutdown payload. +10. Evictor and I/O finish injections prove their stop signals precede visible + join panics. +11. Acquire-ordered quiescent guard-count observation tracks clone and drop. +12. Full workspace and alternate-libaio test and lint jobs pass. + +## Open Questions + +None for the implemented scope. Arbitrary worker-body unwind support would +require separate domain-specific ownership, retention, quarantine, and +recovery proofs; no concrete bounded follow-up was identified during +implementation or resolution. diff --git a/docs/tasks/next-id b/docs/tasks/next-id index bc96b9d5..0f04a302 100644 --- a/docs/tasks/next-id +++ b/docs/tasks/next-id @@ -1 +1 @@ -000256 +000257 diff --git a/doradb-storage/src/buffer/evictor.rs b/doradb-storage/src/buffer/evictor.rs index 980571e9..71328294 100644 --- a/doradb-storage/src/buffer/evictor.rs +++ b/doradb-storage/src/buffer/evictor.rs @@ -839,6 +839,8 @@ impl Component for SharedPoolEvictorWorkers { #[inline] fn shutdown(component: &Self::Owned) { + // Panic safety: close every pool domain and wake the evictor before the + // only deliberate propagation point, its joined worker payload. component.shutdown_flag.store(true, Ordering::SeqCst); component.disk_pool.signal_shutdown(); component.index_pool.signal_shutdown(); @@ -854,7 +856,8 @@ impl Component for SharedPoolEvictorWorkers { Err(payload) => { // Eviction errors are handled through pool state machines. // A worker panic is an invariant failure in shared pool - // eviction. + // eviction. Arbitrary eviction-body unwind remains outside + // registry-level shutdown containment. resume_unwind(payload); } } @@ -1323,7 +1326,9 @@ mod tests { impl Drop for StartedSharedEvictorRuntime { fn drop(&mut self) { - self.registry.shutdown_all(); + self.registry + .shutdown_all() + .propagate_or_suppress("started_shared_evictor_drop"); } } diff --git a/doradb-storage/src/buffer/mod.rs b/doradb-storage/src/buffer/mod.rs index 7733f065..c2c41986 100644 --- a/doradb-storage/src/buffer/mod.rs +++ b/doradb-storage/src/buffer/mod.rs @@ -305,7 +305,10 @@ impl Component for MetaPool { } #[inline] - fn shutdown(_component: &Self::Owned) {} + fn shutdown(_component: &Self::Owned) { + // Panic safety: this fixed metadata pool owns no worker. Catalog and + // transaction guards are drained before passive owner release. + } } impl Component for IndexPool { @@ -344,7 +347,10 @@ impl Component for IndexPool { } #[inline] - fn shutdown(_component: &Self::Owned) {} + fn shutdown(_component: &Self::Owned) { + // Panic safety: shared I/O and evictor worker components own active + // index-pool shutdown and run earlier in reverse order. + } } impl Component for MemPool { @@ -379,7 +385,10 @@ impl Component for MemPool { } #[inline] - fn shutdown(_component: &Self::Owned) {} + fn shutdown(_component: &Self::Owned) { + // Panic safety: shared I/O and evictor worker components own active + // memory-pool shutdown and run earlier in reverse order. + } } impl Component for DiskPool { @@ -410,7 +419,10 @@ impl Component for DiskPool { } #[inline] - fn shutdown(_component: &Self::Owned) {} + fn shutdown(_component: &Self::Owned) { + // Panic safety: `SharedPoolEvictorWorkers` stops the readonly domain + // before this passive pool-owner hook runs. + } } #[cfg(test)] diff --git a/doradb-storage/src/catalog/mod.rs b/doradb-storage/src/catalog/mod.rs index 279ac04a..2c0c8b01 100644 --- a/doradb-storage/src/catalog/mod.rs +++ b/doradb-storage/src/catalog/mod.rs @@ -893,7 +893,10 @@ impl Component for Catalog { } #[inline] - fn shutdown(_component: &Self::Owned) {} + fn shutdown(_component: &Self::Owned) { + // Panic safety: foreground catalog users are drained before component + // dispatch, and purge workers stop before catalog owner release. + } } /// Dropped table runtime detached from the catalog map for purge destruction. diff --git a/doradb-storage/src/component.rs b/doradb-storage/src/component.rs index 2654c227..f4fb6986 100644 --- a/doradb-storage/src/component.rs +++ b/doradb-storage/src/component.rs @@ -8,10 +8,13 @@ use std::any::{Any, TypeId}; use std::fmt::Display; use std::future::Future; use std::marker::PhantomData; +use std::mem::forget; use std::ops::Deref; +use std::panic::{AssertUnwindSafe, catch_unwind, resume_unwind}; use std::path::PathBuf; use std::result::Result as StdResult; use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread::panicking; /// One lifecycle-managed engine subsystem. /// @@ -33,6 +36,18 @@ use std::sync::atomic::{AtomicBool, Ordering}; /// drop. Because shutdown does not receive the registry, components that need /// other objects during teardown must retain those dependencies in their own /// owned state. +/// +/// Panic safety: +/// - shutdown must close ingress and signal owned workers before a deliberate +/// catchable panic point; +/// - multi-worker owners must attempt every join and required infallible +/// release before propagating the first captured payload; +/// - registry containment uses [`AssertUnwindSafe`] only because the component +/// graph becomes terminal and is never exposed for recovery or reuse; +/// - that containment does not imply that arbitrary component mutation bodies +/// implement `UnwindSafe` or `RefUnwindSafe`; and +/// - normal engine shutdown supplies the documented foreground drain, while +/// bootstrap rollback must establish its own local shutdown preconditions. pub(crate) trait Component: Sized + 'static { type Config; type Owned: Send + Sync + 'static; @@ -70,10 +85,17 @@ trait ErasedComponentBox: Send + Sync { fn name(&self) -> &'static str; fn shutdown(&self); + + fn mark_shutdown_panicked(&self); + + fn shutdown_panicked(&self) -> bool; + + fn outstanding_guard_count(&self) -> usize; } struct TypedComponentBox { owner: QuiescentBox, + shutdown_panicked: AtomicBool, } impl ErasedComponentBox for TypedComponentBox { @@ -86,6 +108,138 @@ impl ErasedComponentBox for TypedComponentBox { fn shutdown(&self) { C::shutdown(&self.owner); } + + #[inline] + fn mark_shutdown_panicked(&self) { + self.shutdown_panicked.store(true, Ordering::Release); + } + + #[inline] + fn shutdown_panicked(&self) -> bool { + self.shutdown_panicked.load(Ordering::Acquire) + } + + #[inline] + fn outstanding_guard_count(&self) -> usize { + self.owner.outstanding_guard_count() + } +} + +/// Original payload produced by a catchable Rust panic. +pub(crate) type PanicPayload = Box; + +/// Retains the first panic payload while safely discarding later payloads. +#[derive(Default)] +pub(crate) struct FirstPanic { + first_payload: Option, + panic_count: usize, +} + +impl FirstPanic { + /// Captures `payload`, retaining the first and forgetting all later ones. + /// + /// Forgetting secondary payloads avoids running an arbitrary payload + /// destructor on an already panic-sensitive teardown path. + #[inline] + pub(crate) fn capture(&mut self, payload: PanicPayload) { + self.panic_count += 1; + if self.first_payload.is_none() { + self.first_payload = Some(payload); + } else { + forget(payload); + } + } + + /// Resumes the first captured panic after local cleanup is complete. + #[inline] + pub(crate) fn resume(mut self) { + if let Some(payload) = self.first_payload.take() { + resume_unwind(payload); + } + } + + #[inline] + fn into_shutdown_outcome(mut self) -> ComponentShutdownOutcome { + ComponentShutdownOutcome { + first_payload: self.first_payload.take(), + panic_count: self.panic_count, + } + } +} + +impl Drop for FirstPanic { + #[inline] + fn drop(&mut self) { + if let Some(payload) = self.first_payload.take() { + forget(payload); + } + } +} + +/// Aggregate result of one once-only component shutdown dispatch. +#[must_use = "component shutdown panic outcomes must be propagated or suppressed"] +pub(crate) struct ComponentShutdownOutcome { + first_payload: Option, + panic_count: usize, +} + +impl ComponentShutdownOutcome { + #[inline] + fn complete() -> Self { + Self { + first_payload: None, + panic_count: 0, + } + } + + /// Returns whether one or more component hooks panicked. + #[inline] + pub(crate) fn is_degraded(&self) -> bool { + self.panic_count != 0 + } + + /// Propagates the first payload, or suppresses it during an existing unwind. + /// + /// All component hooks and required owner-side terminal transitions must be + /// complete before this policy is applied. + #[inline] + pub(crate) fn propagate_or_suppress(mut self, context: &'static str) { + let Some(payload) = self.first_payload.take() else { + return; + }; + if panicking() { + obs::error!( + "event=component_shutdown component=engine action=suppress result=panic context={} panic_count={} reason=thread_already_panicking payload={}", + context, + self.panic_count, + panic_payload_description(payload.as_ref()) + ); + forget(payload); + } else { + obs::error!( + "event=component_shutdown component=engine action=propagate result=panic context={} panic_count={} payload={}", + context, + self.panic_count, + panic_payload_description(payload.as_ref()) + ); + resume_unwind(payload); + } + } + + #[cfg(test)] + #[inline] + fn panic_count(&self) -> usize { + self.panic_count + } +} + +impl Drop for ComponentShutdownOutcome { + #[inline] + fn drop(&mut self) { + if let Some(payload) = self.first_payload.take() { + forget(payload); + } + } } /// Ordered registry for engine components and their dependency handles. @@ -144,6 +298,7 @@ pub(crate) struct ComponentRegistry { access_map: FastHashMap>, boxed_vec: Vec>, shutdown_started: AtomicBool, + shutdown_degraded: AtomicBool, } impl Default for ComponentRegistry { @@ -161,6 +316,7 @@ impl ComponentRegistry { access_map: FastHashMap::default(), boxed_vec: Vec::new(), shutdown_started: AtomicBool::new(false), + shutdown_degraded: AtomicBool::new(false), } } @@ -190,8 +346,10 @@ impl ComponentRegistry { let owner = QuiescentBox::new(owned); let access = C::access(&owner); self.access_map.insert(tid, Box::new(access)); - self.boxed_vec - .push(Box::new(TypedComponentBox:: { owner })); + self.boxed_vec.push(Box::new(TypedComponentBox:: { + owner, + shutdown_panicked: AtomicBool::new(false), + })); } /// Return the cloned dependency handle for a previously registered @@ -224,24 +382,44 @@ impl ComponentRegistry { /// Run explicit component shutdown in reverse registration order. /// /// This is idempotent at the registry level and is intended to stop worker - /// activity before owner drop starts waiting on quiescent guards. + /// activity before owner drop starts waiting on quiescent guards. Catchable + /// hook panics are contained independently so every later hook still runs + /// once. The returned outcome retains the first original payload. #[inline] - pub(crate) fn shutdown_all(&self) { + pub(crate) fn shutdown_all(&self) -> ComponentShutdownOutcome { if self.shutdown_started.swap(true, Ordering::AcqRel) { - return; + return ComponentShutdownOutcome::complete(); } + let mut panics = FirstPanic::default(); for component in self.boxed_vec.iter().rev() { let component_name = component.name(); obs::info!( "event=component_lifecycle component=engine storage_component={} action=shutdown_start result=ok", component_name ); - component.shutdown(); - obs::info!( - "event=component_lifecycle component=engine storage_component={} action=shutdown_finish result=ok", - component_name - ); + // The graph is terminal after shutdown starts. This assertion is + // scoped to hook dispatch and is not a claim that arbitrary + // component-domain mutation is unwind-safe. + match catch_unwind(AssertUnwindSafe(|| component.shutdown())) { + Ok(()) => { + obs::info!( + "event=component_lifecycle component=engine storage_component={} action=shutdown_finish result=ok", + component_name + ); + } + Err(payload) => { + component.mark_shutdown_panicked(); + self.shutdown_degraded.store(true, Ordering::Release); + obs::error!( + "event=component_lifecycle component=engine storage_component={} action=shutdown_finish result=panic payload={}", + component_name, + panic_payload_description(payload.as_ref()) + ); + panics.capture(payload); + } + } } + panics.into_shutdown_outcome() } } @@ -249,8 +427,33 @@ impl Drop for ComponentRegistry { #[inline] fn drop(&mut self) { self.access_map.clear(); + if !self.shutdown_degraded.load(Ordering::Acquire) { + while let Some(owner) = self.boxed_vec.pop() { + drop(owner); + } + return; + } + while let Some(owner) = self.boxed_vec.pop() { - drop(owner); + let guard_count = owner.outstanding_guard_count(); + let reason = if owner.shutdown_panicked() { + Some("shutdown_panic") + } else if guard_count != 0 { + Some("outstanding_guards") + } else { + None + }; + if let Some(reason) = reason { + obs::error!( + "event=component_owner component=engine storage_component={} action=leak result=degraded reason={} guard_count={}", + owner.name(), + reason, + guard_count + ); + forget(owner); + } else { + drop(owner); + } } } } @@ -444,12 +647,13 @@ impl RegistryBuilder { impl Drop for RegistryBuilder { #[inline] fn drop(&mut self) { - if let Some(registry) = self.registry.as_ref() { - registry.shutdown_all(); - } + let outcome = self.registry.as_ref().map(ComponentRegistry::shutdown_all); // Provisions can retain quiescent guards into registered owners, so the // shelf must be drained before owner drop starts. self.shelf.clear(); + if let Some(outcome) = outcome { + outcome.propagate_or_suppress("registry_builder_drop"); + } } } @@ -604,6 +808,18 @@ impl DiskPoolConfig { } } +/// Renders a panic payload without consuming it. +#[inline] +pub(crate) fn panic_payload_description(payload: &(dyn Any + Send)) -> &str { + if let Some(message) = payload.downcast_ref::() { + message.as_str() + } else if let Some(message) = payload.downcast_ref::<&'static str>() { + message + } else { + "" + } +} + #[cfg(test)] mod tests { use super::*; @@ -611,7 +827,9 @@ mod tests { use error_stack::Report; use parking_lot::Mutex; use std::convert::Infallible; + use std::panic::{self, AssertUnwindSafe}; use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; struct ValueComponent; @@ -680,6 +898,14 @@ mod tests { struct ShutdownProbe { events: Arc>>, + panic_message: Option<&'static str>, + drop_event: &'static str, + } + + impl Drop for ShutdownProbe { + fn drop(&mut self) { + self.events.lock().push(self.drop_event); + } } struct ShutdownA; @@ -711,6 +937,9 @@ mod tests { #[inline] fn shutdown(component: &Self::Owned) { component.events.lock().push($name); + if let Some(message) = component.panic_message { + panic::panic_any(message); + } } } }; @@ -815,18 +1044,125 @@ mod tests { let mut registry = ComponentRegistry::new(); registry.register::(ShutdownProbe { events: Arc::clone(&events), + panic_message: None, + drop_event: "drop-a", }); registry.register::(ShutdownProbe { events: Arc::clone(&events), + panic_message: None, + drop_event: "drop-b", }); registry.register::(ShutdownProbe { events: Arc::clone(&events), + panic_message: None, + drop_event: "drop-c", }); - registry.shutdown_all(); - registry.shutdown_all(); + let outcome = registry.shutdown_all(); + assert!(!outcome.is_degraded()); + outcome.propagate_or_suppress("component_registry_test"); + let repeated = registry.shutdown_all(); + assert!(!repeated.is_degraded()); + repeated.propagate_or_suppress("component_registry_test"); + + assert_eq!(events.lock().as_slice(), &["c", "b", "a"]); + drop(registry); + assert_eq!( + events.lock().as_slice(), + &["c", "b", "a", "drop-c", "drop-b", "drop-a"] + ); + } + + #[test] + fn test_component_registry_contains_all_hook_panics_and_resumes_first_payload() { + let events = Arc::new(Mutex::new(Vec::new())); + let mut registry = ComponentRegistry::new(); + registry.register::(ShutdownProbe { + events: Arc::clone(&events), + panic_message: None, + drop_event: "drop-a", + }); + registry.register::(ShutdownProbe { + events: Arc::clone(&events), + panic_message: Some("second panic"), + drop_event: "drop-b", + }); + registry.register::(ShutdownProbe { + events: Arc::clone(&events), + panic_message: Some("first panic"), + drop_event: "drop-c", + }); + let outcome = registry.shutdown_all(); + assert!(outcome.is_degraded()); + assert_eq!(outcome.panic_count(), 2); assert_eq!(events.lock().as_slice(), &["c", "b", "a"]); + + let repeated = registry.shutdown_all(); + assert!(!repeated.is_degraded()); + repeated.propagate_or_suppress("component_registry_repeated_test"); + + let payload = panic::catch_unwind(AssertUnwindSafe(|| { + outcome.propagate_or_suppress("component_registry_test"); + })) + .unwrap_err(); + assert_eq!( + payload.downcast_ref::<&'static str>().copied(), + Some("first panic") + ); + } + + #[test] + fn test_first_panic_forgets_secondary_payload_without_running_destructor() { + struct PanicOnDrop; + + impl Drop for PanicOnDrop { + fn drop(&mut self) { + panic!("secondary panic payload destructor must not run"); + } + } + + let mut panics = FirstPanic::default(); + panics.capture(Box::new("first panic")); + panics.capture(Box::new(PanicOnDrop)); + let payload = panic::catch_unwind(AssertUnwindSafe(|| panics.resume())).unwrap_err(); + assert_eq!( + payload.downcast_ref::<&'static str>().copied(), + Some("first panic") + ); + } + + #[test] + fn test_component_shutdown_payload_is_suppressed_during_existing_unwind() { + struct ApplyOutcomeOnDrop(Option); + + impl Drop for ApplyOutcomeOnDrop { + fn drop(&mut self) { + self.0 + .take() + .expect("test shutdown outcome") + .propagate_or_suppress("existing_unwind_test"); + } + } + + let events = Arc::new(Mutex::new(Vec::new())); + let mut registry = ComponentRegistry::new(); + registry.register::(ShutdownProbe { + events, + panic_message: Some("shutdown panic"), + drop_event: "drop-a", + }); + let outcome = registry.shutdown_all(); + + let payload = panic::catch_unwind(AssertUnwindSafe(|| { + let _apply = ApplyOutcomeOnDrop(Some(outcome)); + panic::panic_any("outer panic"); + })) + .unwrap_err(); + assert_eq!( + payload.downcast_ref::<&'static str>().copied(), + Some("outer panic") + ); } #[test] @@ -842,6 +1178,196 @@ mod tests { assert_eq!(events.lock().as_slice(), &["access", "owner"]); } + struct CountedOwner { + dropped: Arc, + } + + impl Drop for CountedOwner { + fn drop(&mut self) { + self.dropped.fetch_add(1, AtomicOrdering::Relaxed); + } + } + + struct DependencyComponent; + struct IndependentComponent; + struct ShelfGuardConsumer; + struct ShelfPanicComponent; + struct SuspectComponent; + + struct SuspectOwned { + _dependency: QuiescentGuard, + _owner: CountedOwner, + } + + macro_rules! counted_component { + ($component:ident, $access:ty, $access_expr:expr, $shutdown:block) => { + impl Component for $component { + type Config = (); + type Owned = CountedOwner; + type Access = $access; + type Error = Infallible; + + const NAME: &'static str = stringify!($component); + + async fn build( + _config: Self::Config, + _registry: &mut ComponentRegistry, + _shelf: ShelfScope<'_, Self>, + ) -> StdResult<(), Self::Error> { + unreachable!("test-only component") + } + + fn access(owner: &QuiescentBox) -> Self::Access { + ($access_expr)(owner) + } + + fn shutdown(_component: &Self::Owned) $shutdown + } + }; + } + + counted_component!( + DependencyComponent, + QuiescentGuard, + |owner: &QuiescentBox| owner.guard(), + {} + ); + counted_component!( + IndependentComponent, + (), + |_owner: &QuiescentBox| (), + {} + ); + + counted_component!( + ShelfPanicComponent, + (), + |_owner: &QuiescentBox| (), + { + panic!("builder shelf shutdown panic"); + } + ); + + impl Component for ShelfGuardConsumer { + type Config = (); + type Owned = (); + type Access = (); + type Error = Infallible; + + const NAME: &'static str = "ShelfGuardConsumer"; + + async fn build( + _config: Self::Config, + _registry: &mut ComponentRegistry, + _shelf: ShelfScope<'_, Self>, + ) -> StdResult<(), Self::Error> { + unreachable!("test-only component") + } + + fn access(_owner: &QuiescentBox) -> Self::Access {} + + fn shutdown(_component: &Self::Owned) {} + } + + impl Supplier for DependencyComponent { + type Provision = QuiescentGuard; + } + + impl Component for SuspectComponent { + type Config = (); + type Owned = SuspectOwned; + type Access = (); + type Error = Infallible; + + const NAME: &'static str = "SuspectComponent"; + + async fn build( + _config: Self::Config, + _registry: &mut ComponentRegistry, + _shelf: ShelfScope<'_, Self>, + ) -> StdResult<(), Self::Error> { + unreachable!("test-only component") + } + + fn access(_owner: &QuiescentBox) -> Self::Access {} + + fn shutdown(_component: &Self::Owned) { + panic!("suspect shutdown"); + } + } + + #[test] + fn test_degraded_registry_drop_leaks_only_suspect_guard_closure() { + let dependency_dropped = Arc::new(AtomicUsize::new(0)); + let independent_dropped = Arc::new(AtomicUsize::new(0)); + let suspect_dropped = Arc::new(AtomicUsize::new(0)); + let mut registry = ComponentRegistry::new(); + registry.register::(CountedOwner { + dropped: Arc::clone(&dependency_dropped), + }); + let external_guard = registry.dependency::(); + registry.register::(CountedOwner { + dropped: Arc::clone(&independent_dropped), + }); + registry.register::(SuspectOwned { + _dependency: external_guard.clone(), + _owner: CountedOwner { + dropped: Arc::clone(&suspect_dropped), + }, + }); + + let outcome = registry.shutdown_all(); + assert!(outcome.is_degraded()); + drop(outcome); + drop(registry); + + assert_eq!(independent_dropped.load(AtomicOrdering::Relaxed), 1); + assert_eq!(suspect_dropped.load(AtomicOrdering::Relaxed), 0); + assert_eq!(dependency_dropped.load(AtomicOrdering::Relaxed), 0); + // The dependency allocation was intentionally leaked, so a guard + // retained past registry destruction can still release safely. + drop(external_guard); + } + + #[test] + fn test_builder_clears_shelf_guards_before_degraded_registry_drop() { + let dependency_dropped = Arc::new(AtomicUsize::new(0)); + let suspect_dropped = Arc::new(AtomicUsize::new(0)); + let payload = panic::catch_unwind(AssertUnwindSafe(|| { + let mut builder = RegistryBuilder::new(); + let shelf_guard = { + let registry = builder + .registry + .as_mut() + .expect("test builder retains registry"); + registry.register::(CountedOwner { + dropped: Arc::clone(&dependency_dropped), + }); + registry.dependency::() + }; + builder + .shelf + .scope::() + .put::(shelf_guard); + builder + .registry + .as_mut() + .expect("test builder retains registry") + .register::(CountedOwner { + dropped: Arc::clone(&suspect_dropped), + }); + drop(builder); + })) + .unwrap_err(); + + assert_eq!( + payload.downcast_ref::<&'static str>().copied(), + Some("builder shelf shutdown panic") + ); + assert_eq!(dependency_dropped.load(AtomicOrdering::Relaxed), 1); + assert_eq!(suspect_dropped.load(AtomicOrdering::Relaxed), 0); + } + struct Upstream; struct Downstream; diff --git a/doradb-storage/src/engine.rs b/doradb-storage/src/engine.rs index aa1ee3f1..4d43d510 100644 --- a/doradb-storage/src/engine.rs +++ b/doradb-storage/src/engine.rs @@ -12,8 +12,8 @@ use crate::catalog::index::tests::IndexDdlTestController; use crate::catalog::table::tests::TableDdlTestController; use crate::catalog::{Catalog, CatalogConfig}; use crate::component::{ - ComponentRegistry, DiskPoolConfig, EnginePools, IndexPoolConfig, MetaPoolConfig, - RegistryBuilder, + ComponentRegistry, ComponentShutdownOutcome, DiskPoolConfig, EnginePools, IndexPoolConfig, + MetaPoolConfig, RegistryBuilder, }; use crate::conf::EngineConfig; use crate::error::{ @@ -386,10 +386,18 @@ impl Engine { "origin=explicit, session_blocker={session_blocker}, operation_state={operation_state}, observer_count={observer_count}, cleanup_queued={cleanup_queued}, mandatory_callers={mandatory_callers}, mandatory_internal={mandatory_internal}" ))); } - self.finish_shutdown_locked(inner); - obs::info!( - "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=try origin=explicit" - ); + let outcome = self.finish_shutdown_locked(inner); + drop(_shutdown); + if outcome.is_degraded() { + obs::error!( + "event=engine_lifecycle component=engine action=shutdown_finish result=panic mode=try origin=explicit" + ); + } else { + obs::info!( + "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=try origin=explicit" + ); + } + outcome.propagate_or_suppress("engine_try_shutdown"); Ok(()) } @@ -431,11 +439,20 @@ impl Engine { let shutdown_wait = inner.session_registry.first_shutdown_wait(); if shutdown_wait.is_none() { - self.finish_shutdown_locked(inner); - obs::info!( - "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=wait origin={}", - origin.label(), - ); + let outcome = self.finish_shutdown_locked(inner); + drop(_shutdown); + if outcome.is_degraded() { + obs::error!( + "event=engine_lifecycle component=engine action=shutdown_finish result=panic mode=wait origin={}", + origin.label(), + ); + } else { + obs::info!( + "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=wait origin={}", + origin.label(), + ); + } + outcome.propagate_or_suppress("engine_shutdown"); return; } drop(_shutdown); @@ -448,13 +465,14 @@ impl Engine { } #[inline] - fn finish_shutdown_locked(&self, inner: &Arc) { + fn finish_shutdown_locked(&self, inner: &Arc) -> ComponentShutdownOutcome { // Once no registered operation or observer remains, idle session state // can release its registry-owned guards before component shutdown. inner.session_registry.shutdown_idle(); - self.components().shutdown_all(); + let outcome = self.components().shutdown_all(); inner.lifecycle.mark_shutdown(); + outcome } /// Queues rollback for one shutdown-discovered abandoned transaction. @@ -811,6 +829,7 @@ mod tests { use std::future::pending; use std::io::Error as StdIoError; use std::os::unix::fs::symlink; + use std::panic::{self, AssertUnwindSafe}; use std::path::{Path, PathBuf}; use std::sync::atomic::AtomicBool; use std::sync::mpsc; @@ -1760,6 +1779,151 @@ mod tests { }); } + #[test] + fn test_engine_shutdown_contains_purge_finish_panic_and_releases_root() { + let root = TempDir::new().unwrap(); + let events = Arc::new(Mutex::new(Vec::new())); + let injected = Arc::new(AtomicBool::new(false)); + let observed_events = Arc::clone(&events); + let observed_injected = Arc::clone(&injected); + let observer = observe_spawn_named(move |event| { + observed_events.lock().push(event.clone()); + if event == SpawnTestEvent::Finished("Purge-Dispatcher".to_owned()) + && !observed_injected.swap(true, Ordering::AcqRel) + { + panic::panic_any("injected purge dispatcher finish panic"); + } + }); + let engine = + smol::block_on(Engine::bootstrap(test_engine_config_for(root.path()))).unwrap(); + + let payload = panic::catch_unwind(AssertUnwindSafe(|| engine.shutdown())).unwrap_err(); + assert_eq!( + payload.downcast_ref::<&'static str>().copied(), + Some("injected purge dispatcher finish panic") + ); + assert!(injected.load(Ordering::Acquire)); + assert_eq!( + engine.inner().lifecycle.inspect_state(), + EngineLifecycleState::Shutdown + ); + + // A contained payload is consumed once; neither repeated explicit + // shutdown nor eventual owner drop may replay it. + engine.shutdown(); + engine.try_shutdown().unwrap(); + + let events = events.lock(); + let finish_position = |worker: &str| { + events + .iter() + .position(|event| event == &SpawnTestEvent::Finished(worker.to_owned())) + .unwrap_or_else(|| panic!("worker did not finish after contained panic: {worker}")) + }; + let redo_finished = finish_position("Log-Thread"); + let mandatory_1_finished = finish_position("Mandatory-Runtime-1"); + let mandatory_2_finished = finish_position("Mandatory-Runtime-2"); + let purge_dispatcher_finished = finish_position("Purge-Dispatcher"); + let purge_executor_finished = finish_position("Purge-Executor-1"); + let evictor_finished = finish_position("Shared-Pool-Evictor"); + let io_finished = finish_position("IO-Thread"); + assert!(redo_finished < mandatory_1_finished); + assert!(redo_finished < mandatory_2_finished); + assert!(mandatory_1_finished < purge_dispatcher_finished); + assert!(mandatory_2_finished < purge_dispatcher_finished); + assert!(purge_dispatcher_finished < evictor_finished); + assert!(purge_executor_finished < evictor_finished); + assert!(evictor_finished < io_finished); + drop(events); + drop(observer); + + // Root-lease shutdown is an active hook, so a replacement engine can + // start while the degraded terminal owner remains allocated. + let replacement = + smol::block_on(Engine::bootstrap(test_engine_config_for(root.path()))).unwrap(); + replacement.shutdown(); + drop(replacement); + drop(engine); + } + + #[test] + fn test_engine_owner_drop_suppresses_shutdown_panic_during_outer_unwind() { + let root = TempDir::new().unwrap(); + let injected = Arc::new(AtomicBool::new(false)); + let observed_injected = Arc::clone(&injected); + let observer = observe_spawn_named(move |event| { + if event == SpawnTestEvent::Finished("Purge-Dispatcher".to_owned()) + && !observed_injected.swap(true, Ordering::AcqRel) + { + panic::panic_any("injected owner-drop purge finish panic"); + } + }); + let engine = + smol::block_on(Engine::bootstrap(test_engine_config_for(root.path()))).unwrap(); + + let payload = panic::catch_unwind(AssertUnwindSafe(move || { + let _engine = engine; + panic::panic_any("outer engine owner panic"); + })) + .unwrap_err(); + assert_eq!( + payload.downcast_ref::<&'static str>().copied(), + Some("outer engine owner panic") + ); + assert!(injected.load(Ordering::Acquire)); + drop(observer); + + let replacement = + smol::block_on(Engine::bootstrap(test_engine_config_for(root.path()))).unwrap(); + replacement.shutdown(); + } + + #[test] + fn test_engine_contains_evictor_and_io_finish_panics_after_stop_signals() { + for target in ["Shared-Pool-Evictor", "IO-Thread"] { + let root = TempDir::new().unwrap(); + let events = Arc::new(Mutex::new(Vec::new())); + let observed_events = Arc::clone(&events); + let observer = observe_spawn_named(move |event| { + observed_events.lock().push(event.clone()); + if event == SpawnTestEvent::Finished(target.to_owned()) { + panic::panic_any(format!("injected {target} finish panic")); + } + }); + let engine = + smol::block_on(Engine::bootstrap(test_engine_config_for(root.path()))).unwrap(); + + let payload = panic::catch_unwind(AssertUnwindSafe(|| engine.shutdown())).unwrap_err(); + assert_eq!( + payload.downcast_ref::().map(String::as_str), + Some(format!("injected {target} finish panic").as_str()) + ); + assert_eq!( + engine.inner().lifecycle.inspect_state(), + EngineLifecycleState::Shutdown + ); + let events = events.lock(); + assert!( + events.contains(&SpawnTestEvent::Finished(target.to_owned())), + "target worker did not finish: {target}" + ); + if target == "Shared-Pool-Evictor" { + assert!( + events.contains(&SpawnTestEvent::Finished("IO-Thread".to_owned())), + "I/O teardown did not continue after evictor join panic" + ); + } + drop(events); + drop(observer); + + let replacement = + smol::block_on(Engine::bootstrap(test_engine_config_for(root.path()))).unwrap(); + replacement.shutdown(); + drop(replacement); + drop(engine); + } + } + #[test] fn test_engine_shutdown_ignores_live_idle_session_handle() { smol::block_on(async { diff --git a/doradb-storage/src/file/fs.rs b/doradb-storage/src/file/fs.rs index 05fabbe2..d15605b3 100644 --- a/doradb-storage/src/file/fs.rs +++ b/doradb-storage/src/file/fs.rs @@ -1698,6 +1698,8 @@ impl Component for FileSystemWorkers { /// Stop ingress, then join the worker thread after all queued work drains. #[inline] fn shutdown(component: &Self::Owned) { + // Panic safety: all table/index/memory ingress lanes close before the + // only deliberate propagation point, the joined I/O worker payload. component.fs.shutdown_io_clients(); if let Some(handle) = component.handle.lock().take() { match handle.join().inspect_err(|_| { @@ -1709,7 +1711,8 @@ impl Component for FileSystemWorkers { Err(payload) => { // IO request failures are reported through their // completions. A worker panic indicates broken IO-thread - // invariants. + // invariants; arbitrary I/O-body unwind is not repaired by + // registry-level shutdown containment. resume_unwind(payload); } } @@ -2062,7 +2065,10 @@ impl Component for FileSystem { } #[inline] - fn shutdown(_component: &Self::Owned) {} + fn shutdown(_component: &Self::Owned) { + // Panic safety: active I/O shutdown belongs to `FileSystemWorkers`, + // which retains this filesystem dependency and runs earlier. + } } /// Build the filesystem facade together with the deferred shared worker state. @@ -2182,7 +2188,9 @@ pub(crate) mod tests { impl TestFileSystem { #[inline] pub(crate) fn shutdown(&self) { - self.registry.shutdown_all(); + self.registry + .shutdown_all() + .propagate_or_suppress("test_filesystem_shutdown"); } #[inline] @@ -2222,7 +2230,9 @@ pub(crate) mod tests { #[inline] fn drop(&mut self) { self.fs.take(); - self.registry.shutdown_all(); + self.registry + .shutdown_all() + .propagate_or_suppress("test_filesystem_drop"); } } diff --git a/doradb-storage/src/lock/mod.rs b/doradb-storage/src/lock/mod.rs index 74a6a7c8..f84fee96 100644 --- a/doradb-storage/src/lock/mod.rs +++ b/doradb-storage/src/lock/mod.rs @@ -611,7 +611,10 @@ impl Component for LockManager { } #[inline] - fn shutdown(_component: &Self::Owned) {} + fn shutdown(_component: &Self::Owned) { + // Panic safety: the engine session/operation drain removes every lock + // manager user before this passive hook is dispatched. + } } #[derive(Default)] diff --git a/doradb-storage/src/poison.rs b/doradb-storage/src/poison.rs index f14324c9..bb3d93f0 100644 --- a/doradb-storage/src/poison.rs +++ b/doradb-storage/src/poison.rs @@ -120,7 +120,10 @@ impl Component for EnginePoisoner { } #[inline] - fn shutdown(_component: &Self::Owned) {} + fn shutdown(_component: &Self::Owned) { + // Panic safety: this passive owner remains available through every + // earlier component hook that can report fatal runtime state. + } } #[cfg(test)] diff --git a/doradb-storage/src/quiescent.rs b/doradb-storage/src/quiescent.rs index 75a76e26..9b5dfa9d 100644 --- a/doradb-storage/src/quiescent.rs +++ b/doradb-storage/src/quiescent.rs @@ -75,6 +75,11 @@ impl QuiescentGuardCount { wait_for_guard_count_zero(&self.0); } + #[inline] + fn acquire_load(&self) -> usize { + self.0.load(Ordering::Acquire) + } + #[cfg(test)] #[inline] fn load(&self, ordering: Ordering) -> usize { @@ -129,6 +134,16 @@ impl QuiescentBox { pub(crate) fn guard(&self) -> QuiescentGuard { QuiescentGuard::new(self.inner_ptr()) } + + /// Samples the number of outstanding guards for degraded owner release. + /// + /// The registry uses this only after clearing every published access handle + /// and all other owner-side reachability. At that terminal boundary, a zero + /// count cannot increase because no guard remains from which to clone. + #[inline] + pub(crate) fn outstanding_guard_count(&self) -> usize { + self.inner.as_ref().get_ref().guard_count.acquire_load() + } } impl Deref for QuiescentBox { @@ -315,13 +330,20 @@ mod tests { #[test] fn test_quiescent_guard_clone_keeps_same_pointer() { let owner = QuiescentBox::new(vec![1u64, 2, 3, 4]); + assert_eq!(owner.outstanding_guard_count(), 0); let guard = owner.guard(); + assert_eq!(owner.outstanding_guard_count(), 1); let guard_clone = guard.clone(); + assert_eq!(owner.outstanding_guard_count(), 2); let owner_ptr = from_ref::>(&owner); assert_eq!(guard.as_ptr(), owner_ptr); assert_eq!(guard_clone.as_ptr(), owner_ptr); assert_eq!(guard.iter().sum::(), 10); assert_eq!(guard_clone.iter().sum::(), 10); + drop(guard); + assert_eq!(owner.outstanding_guard_count(), 1); + drop(guard_clone); + assert_eq!(owner.outstanding_guard_count(), 0); } #[test] diff --git a/doradb-storage/src/root.rs b/doradb-storage/src/root.rs index 9720429a..593d8c29 100644 --- a/doradb-storage/src/root.rs +++ b/doradb-storage/src/root.rs @@ -1071,6 +1071,9 @@ impl Component for StorageRootLease { #[inline] fn shutdown(component: &Self::Owned) { + // Panic safety: this component is registered first, so taking and + // dropping the lock file happens after every subordinate shutdown hook, + // including hooks reached after a contained panic. drop(component.file.lock().take()); } } diff --git a/doradb-storage/src/runtime/mandatory.rs b/doradb-storage/src/runtime/mandatory.rs index 1ccd72ca..ba6e9968 100644 --- a/doradb-storage/src/runtime/mandatory.rs +++ b/doradb-storage/src/runtime/mandatory.rs @@ -1,5 +1,7 @@ use crate::completion::{Completion, CompletionTake}; -use crate::component::{Component, ComponentRegistry, ShelfScope, Supplier}; +use crate::component::{ + Component, ComponentRegistry, FirstPanic, ShelfScope, Supplier, panic_payload_description, +}; use crate::conf::MandatoryRuntimeConfig; use crate::error::{ CompletionErrorBridge, CompletionResult, ConfigError, ConfigResult, DiscloseError, FatalError, @@ -20,7 +22,7 @@ use std::any::Any; use std::fmt::{self, Display, Formatter}; use std::future::Future; use std::mem::take; -use std::panic::{AssertUnwindSafe, resume_unwind}; +use std::panic::AssertUnwindSafe; use std::result::Result as StdResult; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; @@ -780,7 +782,10 @@ impl Component for MandatoryRuntime { } #[inline] - fn shutdown(_component: &Self::Owned) {} + fn shutdown(_component: &Self::Owned) { + // Panic safety: caller/internal admission, stop signalling, and runner + // joins belong to the later `MandatoryRuntimeWorkers` component. + } } impl Supplier for MandatoryRuntime { @@ -856,9 +861,21 @@ impl Drop for PendingMandatoryRuntimeWorkers { self.runtime.admission.close(); self.runtime.internal_admission.close(); self.runtime.signal_stop(); + let mut panics = FirstPanic::default(); for handle in take(&mut self.handles) { - let _ = handle.join(); + let worker = handle.thread().name().unwrap_or("unknown").to_owned(); + if let Err(payload) = handle.join() { + obs::error!( + "event=worker_startup_rollback component=mandatory_runtime worker={} action=join result=panic payload={}", + worker, + panic_payload_description(payload.as_ref()) + ); + panics.capture(payload); + } } + // A component-startup error remains the primary diagnostic; dropping + // the accumulator forgets captured rollback payloads after all joins. + drop(panics); } } @@ -910,24 +927,52 @@ impl MandatoryRuntimeWorkersOwned { // teardown. Bootstrap rollback may reach this owner before the engine // lifecycle exists, so close it defensively while still requiring every // accepted caller to have drained. + let mut panics = FirstPanic::default(); self.runtime.admission.close(); let (_, callers) = self.runtime.admission.inspect(); - assert!( - callers == 0, - "mandatory runner shutdown requires drained caller admission: callers={callers}" - ); + if callers != 0 { + let message = format!( + "mandatory runner shutdown requires drained caller admission: callers={callers}" + ); + obs::error!( + "event=worker_shutdown component=mandatory_runtime worker=all action=validate_callers result=panic payload={}", + message + ); + panics.capture(Box::new(message)); + // Accepted callers can still depend on the runners and internal + // admission remaining live. Report the invariant without starting + // worker teardown. + panics.resume(); + return; + } self.runtime.internal_admission.close(); runtime::block_on(self.runtime.internal_admission.drain()); self.runtime.signal_stop(); for handle in take(&mut *self.handles.lock()) { - if let Err(panic) = handle.join() { - resume_unwind(panic); + let worker = handle.thread().name().unwrap_or("unknown").to_owned(); + if let Err(payload) = handle.join() { + obs::error!( + "event=worker_shutdown component=mandatory_runtime worker={} action=join result=panic payload={}", + worker, + panic_payload_description(payload.as_ref()) + ); + panics.capture(payload); } } - assert!( - self.runtime.executor.is_empty(), - "mandatory executor must be empty after admission drain and runner join" - ); + if !self.runtime.executor.is_empty() { + let message = + "mandatory executor must be empty after admission drain and runner join".to_owned(); + obs::error!( + "event=worker_shutdown component=mandatory_runtime worker=all action=validate_executor result=panic payload={}", + message + ); + panics.capture(Box::new(message)); + } + // Panic safety after caller admission is confirmed drained: both + // admissions are closed, internal work is drained, every runner is + // signalled and joined, and terminal validation is complete before the + // first join or executor-invariant payload is resumed. + panics.resume(); } } @@ -1145,7 +1190,9 @@ mod tests { use crate::component::RegistryBuilder; use crate::conf::MandatoryRuntimeConfig; use crate::error::{ErrorKind, OperationError}; - use crate::thread::fail_spawn_named; + use crate::thread::{SpawnTestEvent, fail_spawn_named, observe_spawn_named}; + use std::panic::{self, AssertUnwindSafe}; + use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; #[test] @@ -1598,7 +1645,116 @@ mod tests { }); mandatory.close_admission(); runtime::block_on(mandatory.drain_callers()); - registry.shutdown_all(); + registry + .shutdown_all() + .propagate_or_suppress("mandatory_runtime_test"); + } + + #[test] + fn mandatory_shutdown_joins_every_runner_before_resuming_first_panic() { + let events = Arc::new(Mutex::new(Vec::new())); + let observed_events = Arc::clone(&events); + let observer = observe_spawn_named(move |event| { + observed_events.lock().push(event.clone()); + match event { + SpawnTestEvent::Finished(name) if name == "Mandatory-Runtime-1" => { + panic::panic_any("first mandatory runner panic"); + } + SpawnTestEvent::Finished(name) if name == "Mandatory-Runtime-2" => { + panic::panic_any("second mandatory runner panic"); + } + _ => {} + } + }); + let (registry, mandatory) = runtime::block_on(async { + let mut builder = RegistryBuilder::new(); + builder.build::(()).await.unwrap(); + builder + .build::( + MandatoryRuntimeConfig::default() + .worker_threads(2) + .concurrency_limit(1), + ) + .await + .unwrap(); + builder.build::(()).await.unwrap(); + let registry = builder.finish(); + let mandatory = registry.dependency::(); + (registry, mandatory) + }); + mandatory.close_admission(); + runtime::block_on(mandatory.drain_callers()); + + let outcome = registry.shutdown_all(); + assert!(outcome.is_degraded()); + let events = events.lock(); + assert!(events.contains(&SpawnTestEvent::Finished("Mandatory-Runtime-1".to_owned()))); + assert!(events.contains(&SpawnTestEvent::Finished("Mandatory-Runtime-2".to_owned()))); + drop(events); + drop(observer); + + let payload = panic::catch_unwind(AssertUnwindSafe(|| { + outcome.propagate_or_suppress("mandatory_multi_runner_test"); + })) + .unwrap_err(); + assert_eq!( + payload.downcast_ref::<&'static str>().copied(), + Some("first mandatory runner panic") + ); + } + + #[test] + fn mandatory_shutdown_preserves_runners_until_callers_drain() { + let (registry, mandatory) = runtime::block_on(async { + let mut builder = RegistryBuilder::new(); + builder.build::(()).await.unwrap(); + builder + .build::( + MandatoryRuntimeConfig::default() + .worker_threads(1) + .concurrency_limit(1), + ) + .await + .unwrap(); + builder.build::(()).await.unwrap(); + let registry = builder.finish(); + let mandatory = registry.dependency::(); + (registry, mandatory) + }); + let caller = runtime::block_on(mandatory.admission.acquire(mandatory.clone())).unwrap(); + let workers = MandatoryRuntimeWorkersOwned { + runtime: mandatory.clone(), + handles: Mutex::new(vec![std::thread::spawn(|| {})]), + shutdown_started: AtomicBool::new(false), + }; + + let payload = panic::catch_unwind(AssertUnwindSafe(|| workers.shutdown())).unwrap_err(); + let caller_admission = mandatory.admission.inspect(); + let internal_admission = mandatory.internal_admission.inspect(); + let stopping = mandatory.stopping.load(Ordering::Acquire); + let handle_count = workers.handles.lock().len(); + + // Always restore the normal shutdown preconditions before asserting so + // a regression cannot leave mandatory-runtime runners detached. + drop(caller); + for handle in take(&mut *workers.handles.lock()) { + handle.join().unwrap(); + } + drop(workers); + mandatory.close_admission(); + runtime::block_on(mandatory.drain_callers()); + registry + .shutdown_all() + .propagate_or_suppress("mandatory_active_caller_shutdown_test"); + + assert_eq!( + payload.downcast_ref::().map(String::as_str), + Some("mandatory runner shutdown requires drained caller admission: callers=1") + ); + assert_eq!(caller_admission, (true, 1)); + assert_eq!(internal_admission, (false, 0)); + assert!(!stopping); + assert_eq!(handle_count, 1); } #[test] @@ -1663,7 +1819,9 @@ mod tests { }); mandatory.close_admission(); runtime::block_on(mandatory.drain_callers()); - registry.shutdown_all(); + registry + .shutdown_all() + .propagate_or_suppress("mandatory_runtime_test"); } #[test] @@ -1742,7 +1900,9 @@ mod tests { }); mandatory.close_admission(); runtime::block_on(mandatory.drain_callers()); - registry.shutdown_all(); + registry + .shutdown_all() + .propagate_or_suppress("mandatory_runtime_test"); } #[test] @@ -1790,7 +1950,9 @@ mod tests { }); mandatory.close_admission(); runtime::block_on(mandatory.drain_callers()); - registry.shutdown_all(); + registry + .shutdown_all() + .propagate_or_suppress("mandatory_runtime_test"); } #[test] @@ -1847,7 +2009,9 @@ mod tests { }); mandatory.close_admission(); runtime::block_on(mandatory.drain_callers()); - registry.shutdown_all(); + registry + .shutdown_all() + .propagate_or_suppress("mandatory_runtime_test"); assert_eq!(dropped.load(Ordering::Relaxed), 1); } @@ -1901,7 +2065,9 @@ mod tests { }); mandatory.close_admission(); runtime::block_on(mandatory.drain_callers()); - registry.shutdown_all(); + registry + .shutdown_all() + .propagate_or_suppress("mandatory_runtime_test"); assert_eq!(dropped.load(Ordering::Relaxed), 1); } } diff --git a/doradb-storage/src/trx/purge.rs b/doradb-storage/src/trx/purge.rs index f1db0555..45767db2 100644 --- a/doradb-storage/src/trx/purge.rs +++ b/doradb-storage/src/trx/purge.rs @@ -21,6 +21,7 @@ use error_stack::Report; use flume::{Receiver, Sender}; use parking_lot::Mutex; use std::collections::VecDeque; +use std::mem::forget; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::thread::JoinHandle; @@ -328,6 +329,13 @@ impl TransactionSystem { /// secondary-index cleanup fails, purge cannot replay the same in-memory /// mutation safely, so it poisons runtime admission before returning the /// fatal purge error. + /// + /// Panic containment does not cover an arbitrary unwind from the mutation + /// body below. The owned `Vec` can contain boxed row undo + /// backing non-owning `RowUndoRef` links that are still reachable from row + /// version chains. A worker join panic proves thread termination only; + /// domain-level unwind support would first need explicit retention or + /// quarantine for that ownership graph. #[inline] pub(super) async fn purge_trx_list( &self, @@ -1316,8 +1324,18 @@ fn reclaim_partial_purge_workers( ) -> Report { let mut join_panics = 0usize; for handle in handles { - if handle.join().is_err() { + let worker = handle.thread().name().unwrap_or("unknown").to_owned(); + if let Err(payload) = handle.join() { join_panics += 1; + obs::error!( + "event=worker_startup_rollback component=trx worker={} action=join result=panic payload={}", + worker, + crate::component::panic_payload_description(payload.as_ref()) + ); + // The spawn report is the primary startup diagnostic. Forget the + // secondary payload after observing it rather than invoking an + // arbitrary payload destructor during rollback. + forget(payload); } } if join_panics != 0 { diff --git a/doradb-storage/src/trx/sys.rs b/doradb-storage/src/trx/sys.rs index 3ca8feb0..03e568ef 100644 --- a/doradb-storage/src/trx/sys.rs +++ b/doradb-storage/src/trx/sys.rs @@ -3,7 +3,8 @@ use crate::buffer::PoolGuards; use crate::catalog::{Catalog, CatalogCheckpointScanConfig, TableCache}; use crate::completion::Completion; use crate::component::{ - Component, ComponentRegistry, EnginePools, IndexPool, MemPool, MetaPool, ShelfScope, Supplier, + Component, ComponentRegistry, EnginePools, FirstPanic, IndexPool, MemPool, MetaPool, + ShelfScope, Supplier, panic_payload_description, }; use crate::conf::TrxSysConfig; use crate::error::{ @@ -47,7 +48,6 @@ use flume::{Receiver, Sender}; use parking_lot::{Mutex, MutexGuard}; use std::collections::BTreeMap; use std::mem::{forget, take}; -use std::panic::resume_unwind; use std::result::Result as StdResult; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; @@ -200,11 +200,15 @@ impl PendingTransactionPurgeStartup { self, trx_sys: QuiescentGuard, ) -> RuntimeOrFatalResult { - let purge_threads = - TransactionSystem::start_purge_threads(trx_sys, self.pool_guards, self.purge_rx) - .attach("phase=start_transaction_purge_workers") - .map_err(RuntimeOrFatalError::from)?; + let purge_threads = TransactionSystem::start_purge_threads( + trx_sys.clone(), + self.pool_guards, + self.purge_rx, + ) + .attach("phase=start_transaction_purge_workers") + .map_err(RuntimeOrFatalError::from)?; Ok(TransactionPurgeWorkersOwned { + _trx_sys: trx_sys, purge_tx: self.purge_tx, purge_threads: Mutex::new(purge_threads), shutdown_started: AtomicBool::new(false), @@ -247,6 +251,9 @@ impl PendingTransactionRedoStartup { /// Owned purge workers retained below the mandatory runtime. pub(crate) struct TransactionPurgeWorkersOwned { + // This explicit dependency pins the transaction domain if degraded + // registry release must leak a suspect purge owner. + _trx_sys: QuiescentGuard, purge_tx: Sender, purge_threads: Mutex>>, shutdown_started: AtomicBool, @@ -263,19 +270,23 @@ impl TransactionPurgeWorkersOwned { "event=worker_shutdown component=trx worker=purge action=signal_stop result=ignored reason=receiver_closed" ); } + let mut panics = FirstPanic::default(); let purge_threads = { take(&mut *self.purge_threads.lock()) }; for handle in purge_threads { - if let Err(payload) = handle.join().inspect_err(|_| { + let worker = handle.thread().name().unwrap_or("unknown").to_owned(); + if let Err(payload) = handle.join() { obs::error!( - "event=worker_shutdown component=trx worker=purge action=join result=error reason=panic" + "event=worker_shutdown component=trx worker={} action=join result=panic payload={}", + worker, + panic_payload_description(payload.as_ref()) ); - }) { - // Purge known failures should be represented before thread - // exit. A join panic is an invariant failure that must remain - // visible to the owner. - resume_unwind(payload); + panics.capture(payload); } } + // Panic safety: `Purge::Stop` is sent before all handles are taken, and + // every dispatcher/executor join is attempted before the first original + // payload is resumed. This does not make mid-purge mutation unwind-safe. + panics.resume(); } } @@ -381,12 +392,21 @@ impl TransactionRedoWorkersOwned { group_commit.queue.push_back(Commit::Shutdown); redo_log.group_commit.notify_one(); } + let mut panics = FirstPanic::default(); if let Some(handle) = self.log_thread.lock().take() && let Err(payload) = handle.join() { - resume_unwind(payload); + obs::error!( + "event=worker_shutdown component=trx worker=Log-Thread action=join result=panic payload={}", + panic_payload_description(payload.as_ref()) + ); + panics.capture(payload); } drop(redo_log.group_commit.lock().log_file.take()); + // Panic safety: group-commit admission and the shutdown marker precede + // the join, and active log-file release completes before propagation. + // Arbitrary redo-body unwind remains outside this containment boundary. + panics.resume(); } } @@ -1571,14 +1591,19 @@ impl TransactionSystem { group_commit_g.queue.push_back(Commit::Shutdown); redo_log.group_commit.notify_one(); } - handle - .join() - .inspect_err(|_| { + match handle.join() { + Ok(()) => true, + Err(payload) => { obs::error!( - "event=worker_startup_rollback component=trx worker=Log-Thread action=join result=error reason=panic" + "event=worker_startup_rollback component=trx worker=Log-Thread action=join result=panic payload={}", + panic_payload_description(payload.as_ref()) ); - }) - .is_ok() + // Startup already has a primary typed diagnostic. Retain that + // result and avoid a payload destructor on the rollback path. + forget(payload); + false + } + } } /// Submit abandoned transaction rollback cleanup. @@ -1718,7 +1743,11 @@ impl Component for TransactionSystem { } #[inline] - fn shutdown(_component: &Self::Owned) {} + fn shutdown(_component: &Self::Owned) { + // Panic safety: active redo, mandatory-runtime, and purge authority is + // owned by later worker components. This passive hook runs only after + // those hooks attempted terminal shutdown. + } } #[inline] @@ -1842,8 +1871,10 @@ pub(crate) mod tests { use crate::log::redo::{RowRedo, RowRedoKind}; use crate::recovery::stream::RedoSegmentCtsRange; use crate::session::tests::SessionTestExt; + use crate::thread::{SpawnTestEvent, observe_spawn_named}; use crate::trx::{PrecommitTrxPayload, RetiredRowPageBatch, SharedTrxStatus}; use crate::value::Val; + use std::panic::{self, AssertUnwindSafe}; use std::sync::{Arc, Barrier, OnceLock}; use std::thread::spawn; use tempfile::TempDir; @@ -1975,6 +2006,48 @@ pub(crate) mod tests { (temp_dir, engine) } + #[test] + fn redo_shutdown_releases_active_file_before_resuming_join_panic() { + let observer = observe_spawn_named(|event| { + if event == SpawnTestEvent::Finished("Log-Thread".to_owned()) { + panic::panic_any("injected redo finish panic"); + } + }); + let (_temp_dir, engine) = + smol::block_on(build_trx_sys_redo_test_engine_with_log_file_max_size( + "redo_shutdown_panic", + 1024 * 1024, + )); + assert!( + engine + .inner() + .trx_sys + .redo_log + .group_commit + .lock() + .log_file + .is_some() + ); + + let payload = panic::catch_unwind(AssertUnwindSafe(|| engine.shutdown())).unwrap_err(); + assert_eq!( + payload.downcast_ref::<&'static str>().copied(), + Some("injected redo finish panic") + ); + assert!( + engine + .inner() + .trx_sys + .redo_log + .group_commit + .lock() + .log_file + .is_none() + ); + engine.shutdown(); + drop(observer); + } + fn add_large_system_redo(sys_trx: &mut SysTrx, value_count: usize) { let values = (0..value_count as u64).map(Val::from).collect(); sys_trx.redo.insert_dml(