From a7c179256f89bc307f4ba359b513905b62d61a3c Mon Sep 17 00:00:00 2001 From: jarmen423 Date: Tue, 28 Jul 2026 01:49:53 +0000 Subject: [PATCH 1/3] feat: add diagnostics-only close-forensics lifecycle sink Add a feature-gated non-owning event ring and tiny Drop/worker events for GrafeoDB, BufferManager, HnswIndex, and checkpoint timer so staging can prove engine teardown without retaining diagnosed Arcs. --- crates/grafeo-common/Cargo.toml | 2 + crates/grafeo-common/src/close_forensics.rs | 220 ++++++++++++++++++ crates/grafeo-common/src/lib.rs | 2 + .../src/memory/buffer/manager.rs | 16 ++ crates/grafeo-core/Cargo.toml | 2 + crates/grafeo-core/src/index/vector/hnsw.rs | 23 ++ crates/grafeo-engine/Cargo.toml | 2 + .../src/database/checkpoint_timer.rs | 40 ++++ crates/grafeo-engine/src/database/mod.rs | 12 + crates/grafeo/Cargo.toml | 2 + crates/grafeo/src/lib.rs | 4 + 11 files changed, 325 insertions(+) create mode 100644 crates/grafeo-common/src/close_forensics.rs diff --git a/crates/grafeo-common/Cargo.toml b/crates/grafeo-common/Cargo.toml index 3bb20e39f..131ffc5d8 100644 --- a/crates/grafeo-common/Cargo.toml +++ b/crates/grafeo-common/Cargo.toml @@ -51,6 +51,8 @@ tracing = ["dep:tracing"] testing-crash-injection = [] testing-statement-injection = [] encryption = ["dep:aes-gcm", "dep:hkdf", "dep:argon2", "dep:zeroize", "dep:sha2", "dep:rand"] +# Diagnostics-only: tiny non-owning close/lifecycle event ring. Off by default. +close-forensics = [] [dev-dependencies] criterion.workspace = true diff --git a/crates/grafeo-common/src/close_forensics.rs b/crates/grafeo-common/src/close_forensics.rs new file mode 100644 index 000000000..4e5247e07 --- /dev/null +++ b/crates/grafeo-common/src/close_forensics.rs @@ -0,0 +1,220 @@ +//! Diagnostics-only close/lifecycle forensics (feature `close-forensics`). +//! +//! # Non-owning invariant +//! Never store `Arc` to databases, indexes, buffer managers, worker groups, or +//! closures capturing those. Records are plain integers only. +//! +//! # Drop safety +//! [`emit`] never panics, never blocks on a contended lock (`try_lock` only), +//! never walks memory graphs, and never performs file I/O. A full or busy sink +//! drops the event and increments a counter. + +use std::cell::Cell; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Mutex, OnceLock}; +use std::time::Instant; + +/// Component type codes (stable for forensics reports). +pub mod component_type { + pub const GRAFEO_DB: u8 = 1; + pub const BUFFER_MANAGER: u8 = 2; + pub const HNSW_INDEX: u8 = 3; + pub const CHECKPOINT_TIMER: u8 = 4; + pub const QUANTIZED_HNSW: u8 = 5; +} + +/// Event type codes. +pub mod event_type { + pub const DROP: u8 = 1; + pub const WORKER_STARTED: u8 = 2; + pub const WORKER_SHUTDOWN_REQUESTED: u8 = 3; + pub const WORKER_JOINED: u8 = 4; + pub const INSTANCE_BOUND: u8 = 5; +} + +/// Worker scope codes. +pub mod worker_scope { + pub const DATABASE_INSTANCE: u8 = 1; + pub const PROCESS_GLOBAL: u8 = 2; + pub const SHARED_ENGINE: u8 = 3; + pub const UNKNOWN: u8 = 4; +} + +/// db_ref_kind: reported metadata only (no Arc retained). +pub mod db_ref_kind { + pub const NONE: u8 = 0; + pub const STRONG: u8 = 1; + pub const WEAK: u8 = 2; + pub const UNKNOWN: u8 = 3; +} + +/// Fixed-size lifecycle record. +#[derive(Debug, Clone, Copy, Default)] +pub struct TinyLifecycleEvent { + pub event_sequence: u64, + pub monotonic_elapsed_ns: u64, + pub instance_id: u64, + pub component_id: u64, + pub component_type: u8, + pub event_type: u8, + pub worker_scope: u8, + pub db_ref_kind: u8, + pub thread_id_hash: u64, +} + +const RING_CAP: usize = 4096; + +struct RingState { + slots: Vec, + write: usize, +} + +static SEQ: AtomicU64 = AtomicU64::new(1); +static COMPONENT_IDS: AtomicU64 = AtomicU64::new(1); +static ACTIVE_INSTANCE_ID: AtomicU64 = AtomicU64::new(0); +static WRITTEN: AtomicU64 = AtomicU64::new(0); +static DROPPED: AtomicU64 = AtomicU64::new(0); +static START: OnceLock = OnceLock::new(); +static RING: OnceLock> = OnceLock::new(); + +fn ring() -> &'static Mutex { + RING.get_or_init(|| { + Mutex::new(RingState { + slots: vec![TinyLifecycleEvent::default(); RING_CAP], + write: 0, + }) + }) +} + +fn elapsed_ns() -> u64 { + START.get_or_init(Instant::now).elapsed().as_nanos() as u64 +} + +thread_local! { + static THREAD_DIAG_ID: Cell = const { Cell::new(0) }; +} +static THREAD_DIAG_IDS: AtomicU64 = AtomicU64::new(1); + +fn thread_id_hash() -> u64 { + THREAD_DIAG_ID.with(|cell| { + let mut id = cell.get(); + if id == 0 { + id = THREAD_DIAG_IDS.fetch_add(1, Ordering::Relaxed); + cell.set(id); + } + id + }) +} + +/// Pre-initialize the ring (call from probe before open so first Drop never allocates the Vec). +pub fn init_sink() { + let _ = START.get_or_init(Instant::now); + let _ = ring(); +} + +/// Bind the process-local active instance id for subsequent component Drop. +pub fn set_active_instance_id(instance_id: u64) { + ACTIVE_INSTANCE_ID.store(instance_id, Ordering::Release); +} + +pub fn active_instance_id() -> u64 { + ACTIVE_INSTANCE_ID.load(Ordering::Acquire) +} + +/// Allocate a new component id (monotonic). +pub fn next_component_id() -> u64 { + COMPONENT_IDS.fetch_add(1, Ordering::Relaxed) +} + +/// Emit a tiny lifecycle event. Never panics. Never holds diagnosed Arcs. +pub fn emit( + instance_id: u64, + component_id: u64, + component_type: u8, + event_type: u8, + worker_scope: u8, + db_ref_kind: u8, +) { + let seq = SEQ.fetch_add(1, Ordering::Relaxed); + let event = TinyLifecycleEvent { + event_sequence: seq, + monotonic_elapsed_ns: elapsed_ns(), + instance_id, + component_id, + component_type, + event_type, + worker_scope, + db_ref_kind, + thread_id_hash: thread_id_hash(), + }; + let Ok(mut guard) = ring().try_lock() else { + DROPPED.fetch_add(1, Ordering::Relaxed); + return; + }; + let idx = guard.write % RING_CAP; + guard.slots[idx] = event; + guard.write = guard.write.wrapping_add(1); + WRITTEN.fetch_add(1, Ordering::Relaxed); +} + +/// Copy up to `out.len()` most recent events (best-effort). +pub fn drain_snapshot(out: &mut [TinyLifecycleEvent]) -> usize { + let Ok(guard) = ring().try_lock() else { + return 0; + }; + let written = WRITTEN.load(Ordering::Relaxed) as usize; + if written == 0 || out.is_empty() { + return 0; + } + let n = out.len().min(RING_CAP).min(written); + let start = guard.write.wrapping_sub(n); + for (i, slot) in out.iter_mut().enumerate().take(n) { + let idx = start.wrapping_add(i) % RING_CAP; + *slot = guard.slots[idx]; + } + n +} + +pub fn stats() -> (u64, u64) { + (WRITTEN.load(Ordering::Relaxed), DROPPED.load(Ordering::Relaxed)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn emit_and_drain_roundtrip() { + let id = next_component_id(); + set_active_instance_id(42); + emit( + 42, + id, + component_type::GRAFEO_DB, + event_type::DROP, + worker_scope::DATABASE_INSTANCE, + db_ref_kind::NONE, + ); + let mut buf = [TinyLifecycleEvent::default(); 16]; + let n = drain_snapshot(&mut buf); + assert!(n >= 1); + assert!(buf[..n] + .iter() + .any(|e| e.instance_id == 42 && e.component_id == id)); + } + + #[test] + fn emit_never_panics_on_flood() { + set_active_instance_id(7); + for _ in 0..RING_CAP * 2 { + emit( + 7, + next_component_id(), + component_type::BUFFER_MANAGER, + event_type::DROP, + 0, + db_ref_kind::NONE, + ); + } + } +} diff --git a/crates/grafeo-common/src/lib.rs b/crates/grafeo-common/src/lib.rs index 747f9a4eb..b24c2ba7a 100644 --- a/crates/grafeo-common/src/lib.rs +++ b/crates/grafeo-common/src/lib.rs @@ -18,6 +18,8 @@ #![deny(unsafe_code)] pub mod collections; +#[cfg(feature = "close-forensics")] +pub mod close_forensics; pub mod fmt; pub mod memory; pub mod mvcc; diff --git a/crates/grafeo-common/src/memory/buffer/manager.rs b/crates/grafeo-common/src/memory/buffer/manager.rs index 0c995e529..e789b5886 100644 --- a/crates/grafeo-common/src/memory/buffer/manager.rs +++ b/crates/grafeo-common/src/memory/buffer/manager.rs @@ -141,6 +141,9 @@ pub struct BufferManager { hard_limit: usize, /// Shutdown flag. shutdown: AtomicBool, + /// Diagnostics-only component id (no Arc retained). + #[cfg(feature = "close-forensics")] + forensics_component_id: u64, } impl BufferManager { @@ -172,6 +175,8 @@ impl BufferManager { evict_limit, hard_limit, shutdown: AtomicBool::new(false), + #[cfg(feature = "close-forensics")] + forensics_component_id: crate::close_forensics::next_component_id(), }) } @@ -645,6 +650,17 @@ impl GrantReleaser for BufferManager { impl Drop for BufferManager { fn drop(&mut self) { + #[cfg(feature = "close-forensics")] + { + crate::close_forensics::emit( + crate::close_forensics::active_instance_id(), + self.forensics_component_id, + crate::close_forensics::component_type::BUFFER_MANAGER, + crate::close_forensics::event_type::DROP, + crate::close_forensics::worker_scope::DATABASE_INSTANCE, + crate::close_forensics::db_ref_kind::NONE, + ); + } self.shutdown.store(true, Ordering::Relaxed); } } diff --git a/crates/grafeo-core/Cargo.toml b/crates/grafeo-core/Cargo.toml index a054a89ad..63e5a3381 100644 --- a/crates/grafeo-core/Cargo.toml +++ b/crates/grafeo-core/Cargo.toml @@ -84,6 +84,8 @@ ai = ["vector-index", "text-index", "hybrid-search"] # All AI/RAG search featur # Testing utilities (zero overhead when disabled) testing-crash-injection = ["grafeo-common/testing-crash-injection"] # Deterministic crash injection for recovery testing +# Diagnostics-only close/lifecycle forensics. Off by default. +close-forensics = ["grafeo-common/close-forensics"] [lints] workspace = true diff --git a/crates/grafeo-core/src/index/vector/hnsw.rs b/crates/grafeo-core/src/index/vector/hnsw.rs index ade1e13fb..a93666643 100644 --- a/crates/grafeo-core/src/index/vector/hnsw.rs +++ b/crates/grafeo-core/src/index/vector/hnsw.rs @@ -249,6 +249,23 @@ pub struct HnswIndex { max_level: RwLock, /// Random number generator for level selection. rng: RwLock, + #[cfg(feature = "close-forensics")] + forensics_component_id: u64, +} + + +#[cfg(feature = "close-forensics")] +impl Drop for HnswIndex { + fn drop(&mut self) { + grafeo_common::close_forensics::emit( + grafeo_common::close_forensics::active_instance_id(), + self.forensics_component_id, + grafeo_common::close_forensics::component_type::HNSW_INDEX, + grafeo_common::close_forensics::event_type::DROP, + grafeo_common::close_forensics::worker_scope::DATABASE_INSTANCE, + grafeo_common::close_forensics::db_ref_kind::NONE, + ); + } } impl HnswIndex { @@ -260,6 +277,8 @@ impl HnswIndex { nodes: RwLock::new(TopologyBackend::new_heap()), entry_point: RwLock::new(None), max_level: RwLock::new(0), + #[cfg(feature = "close-forensics")] + forensics_component_id: grafeo_common::close_forensics::next_component_id(), rng: RwLock::new(rand::rngs::StdRng::from_rng(&mut rand::rng())), } } @@ -275,6 +294,8 @@ impl HnswIndex { nodes: RwLock::new(TopologyBackend::with_capacity(capacity)), entry_point: RwLock::new(None), max_level: RwLock::new(0), + #[cfg(feature = "close-forensics")] + forensics_component_id: grafeo_common::close_forensics::next_component_id(), rng: RwLock::new(rand::rngs::StdRng::from_rng(&mut rand::rng())), } } @@ -287,6 +308,8 @@ impl HnswIndex { nodes: RwLock::new(TopologyBackend::new_heap()), entry_point: RwLock::new(None), max_level: RwLock::new(0), + #[cfg(feature = "close-forensics")] + forensics_component_id: grafeo_common::close_forensics::next_component_id(), rng: RwLock::new(rand::rngs::StdRng::seed_from_u64(seed)), } } diff --git a/crates/grafeo-engine/Cargo.toml b/crates/grafeo-engine/Cargo.toml index c3090e8ed..4217dbc07 100644 --- a/crates/grafeo-engine/Cargo.toml +++ b/crates/grafeo-engine/Cargo.toml @@ -126,6 +126,8 @@ arrow-export = ["dep:arrow-schema", "dep:arrow-array", "dep:arrow-ipc"] # Arrow # Testing utilities (zero overhead when disabled) testing-crash-injection = ["dep:grafeo-storage", "grafeo-common/testing-crash-injection", "grafeo-storage/testing-crash-injection"] # Crash injection for recovery testing testing-statement-injection = ["grafeo-common/testing-statement-injection"] # Failure injection at statement / commit entry for rollback atomicity tests +# Diagnostics-only close/lifecycle forensics (non-owning tiny events). Off by default. +close-forensics = ["grafeo-common/close-forensics", "grafeo-core/close-forensics"] [lints] workspace = true diff --git a/crates/grafeo-engine/src/database/checkpoint_timer.rs b/crates/grafeo-engine/src/database/checkpoint_timer.rs index 8d7109f53..70f306ec4 100644 --- a/crates/grafeo-engine/src/database/checkpoint_timer.rs +++ b/crates/grafeo-engine/src/database/checkpoint_timer.rs @@ -41,6 +41,8 @@ pub(super) struct CheckpointTimer { shutdown: Arc, /// Thread handle (taken on stop). handle: Option>, + #[cfg(feature = "close-forensics")] + forensics_component_id: u64, } #[cfg(feature = "grafeo-file")] @@ -62,6 +64,20 @@ impl CheckpointTimer { let shutdown = Arc::new(AtomicBool::new(false)); let shutdown_clone = Arc::clone(&shutdown); + #[cfg(feature = "close-forensics")] + let forensics_component_id = grafeo_common::close_forensics::next_component_id(); + #[cfg(feature = "close-forensics")] + { + grafeo_common::close_forensics::emit( + grafeo_common::close_forensics::active_instance_id(), + forensics_component_id, + grafeo_common::close_forensics::component_type::CHECKPOINT_TIMER, + grafeo_common::close_forensics::event_type::WORKER_STARTED, + grafeo_common::close_forensics::worker_scope::DATABASE_INSTANCE, + grafeo_common::close_forensics::db_ref_kind::STRONG, + ); + } + let handle = std::thread::Builder::new() .name("grafeo-checkpoint".to_string()) .spawn(move || { @@ -77,12 +93,25 @@ impl CheckpointTimer { #[cfg(feature = "wal")] wal.as_deref(), ); + #[cfg(feature = "close-forensics")] + { + grafeo_common::close_forensics::emit( + grafeo_common::close_forensics::active_instance_id(), + forensics_component_id, + grafeo_common::close_forensics::component_type::CHECKPOINT_TIMER, + grafeo_common::close_forensics::event_type::WORKER_JOINED, + grafeo_common::close_forensics::worker_scope::DATABASE_INSTANCE, + grafeo_common::close_forensics::db_ref_kind::NONE, + ); + } }) .expect("failed to spawn checkpoint timer thread"); Self { shutdown, handle: Some(handle), + #[cfg(feature = "close-forensics")] + forensics_component_id, } } @@ -90,6 +119,17 @@ impl CheckpointTimer { /// /// Returns within ~100 ms regardless of the checkpoint interval. pub(super) fn stop(&mut self) { + #[cfg(feature = "close-forensics")] + { + grafeo_common::close_forensics::emit( + grafeo_common::close_forensics::active_instance_id(), + self.forensics_component_id, + grafeo_common::close_forensics::component_type::CHECKPOINT_TIMER, + grafeo_common::close_forensics::event_type::WORKER_SHUTDOWN_REQUESTED, + grafeo_common::close_forensics::worker_scope::DATABASE_INSTANCE, + grafeo_common::close_forensics::db_ref_kind::STRONG, + ); + } self.shutdown.store(true, Ordering::Release); if let Some(handle) = self.handle.take() { let _ = handle.join(); diff --git a/crates/grafeo-engine/src/database/mod.rs b/crates/grafeo-engine/src/database/mod.rs index 31c3b4bf0..5f0bf2c35 100644 --- a/crates/grafeo-engine/src/database/mod.rs +++ b/crates/grafeo-engine/src/database/mod.rs @@ -3046,6 +3046,18 @@ impl GrafeoDB { impl Drop for GrafeoDB { fn drop(&mut self) { + #[cfg(feature = "close-forensics")] + { + // Tiny non-owning event only — no memory_usage(), I/O, or path clones. + grafeo_common::close_forensics::emit( + grafeo_common::close_forensics::active_instance_id(), + grafeo_common::close_forensics::next_component_id(), + grafeo_common::close_forensics::component_type::GRAFEO_DB, + grafeo_common::close_forensics::event_type::DROP, + grafeo_common::close_forensics::worker_scope::DATABASE_INSTANCE, + grafeo_common::close_forensics::db_ref_kind::NONE, + ); + } if let Err(e) = self.close() { grafeo_error!("Error closing database: {}", e); } diff --git a/crates/grafeo/Cargo.toml b/crates/grafeo/Cargo.toml index 2bc577f17..838526f5a 100644 --- a/crates/grafeo/Cargo.toml +++ b/crates/grafeo/Cargo.toml @@ -81,6 +81,8 @@ async-storage = ["grafeo-engine/async-storage"] # Async WAL and storage backend # Enable platform-optimized memory allocator (10-20% faster allocations) jemalloc = ["tikv-jemallocator"] mimalloc-allocator = ["mimalloc"] +# Diagnostics-only close/lifecycle forensics. Off by default. +close-forensics = ["grafeo-engine/close-forensics"] [lints] workspace = true diff --git a/crates/grafeo/src/lib.rs b/crates/grafeo/src/lib.rs index 8e101fb14..5bb553519 100644 --- a/crates/grafeo/src/lib.rs +++ b/crates/grafeo/src/lib.rs @@ -94,3 +94,7 @@ pub use grafeo_common::types::{EdgeId, NodeId, Value}; // Re-export error types so users don't need to depend on grafeo-common directly pub use grafeo_common::utils::error::{Error, Result}; + +/// Diagnostics-only close/lifecycle forensics (feature `close-forensics`). +#[cfg(feature = "close-forensics")] +pub use grafeo_common::close_forensics; From bb8b8ed48a77afe9a763e14bcf25d9dce2b2bb7a Mon Sep 17 00:00:00 2001 From: jarmen423 Date: Wed, 29 Jul 2026 02:03:44 +0000 Subject: [PATCH 2/3] feat(close-forensics): DROP_ENTER/PHASE timing around teardown Attribute free/close work inside timed Drop regions (emit phases after work), keep checkpoint JOINED after join, and expose DROP_ENTER/DROP_PHASE event codes for sidecar close breakdowns. --- crates/grafeo-common/src/close_forensics.rs | 17 ++++ .../src/memory/buffer/manager.rs | 54 ++++++++++-- crates/grafeo-core/src/index/vector/hnsw.rs | 88 +++++++++++++++++-- .../src/database/checkpoint_timer.rs | 25 +++--- crates/grafeo-engine/src/database/mod.rs | 50 +++++++++-- 5 files changed, 200 insertions(+), 34 deletions(-) diff --git a/crates/grafeo-common/src/close_forensics.rs b/crates/grafeo-common/src/close_forensics.rs index 4e5247e07..77a309f5a 100644 --- a/crates/grafeo-common/src/close_forensics.rs +++ b/crates/grafeo-common/src/close_forensics.rs @@ -30,6 +30,23 @@ pub mod event_type { pub const WORKER_SHUTDOWN_REQUESTED: u8 = 3; pub const WORKER_JOINED: u8 = 4; pub const INSTANCE_BOUND: u8 = 5; + /// Drop body entered; free work follows inside timed phases. + pub const DROP_ENTER: u8 = 6; + /// Sub-phase inside Drop (`worker_scope` carries [`drop_phase`] code). + pub const DROP_PHASE: u8 = 7; +} + +/// Phase tags for [`event_type::DROP_PHASE`] (stored in `worker_scope`). +pub mod drop_phase { + pub const TOPOLOGY: u8 = 1; + pub const ENTRY_POINT: u8 = 2; + pub const MAX_LEVEL: u8 = 3; + pub const RNG: u8 = 4; + pub const CONFIG: u8 = 5; + pub const CONSUMERS: u8 = 6; + pub const FORCE_RAM: u8 = 7; + pub const CLOSE_FN: u8 = 8; + pub const FIELDS: u8 = 9; } /// Worker scope codes. diff --git a/crates/grafeo-common/src/memory/buffer/manager.rs b/crates/grafeo-common/src/memory/buffer/manager.rs index e789b5886..aa3eea379 100644 --- a/crates/grafeo-common/src/memory/buffer/manager.rs +++ b/crates/grafeo-common/src/memory/buffer/manager.rs @@ -652,16 +652,54 @@ impl Drop for BufferManager { fn drop(&mut self) { #[cfg(feature = "close-forensics")] { - crate::close_forensics::emit( - crate::close_forensics::active_instance_id(), - self.forensics_component_id, - crate::close_forensics::component_type::BUFFER_MANAGER, - crate::close_forensics::event_type::DROP, - crate::close_forensics::worker_scope::DATABASE_INSTANCE, - crate::close_forensics::db_ref_kind::NONE, + use crate::close_forensics::{ + active_instance_id, component_type, db_ref_kind, drop_phase, emit, event_type, + worker_scope, + }; + let instance = active_instance_id(); + let cid = self.forensics_component_id; + emit( + instance, + cid, + component_type::BUFFER_MANAGER, + event_type::DROP_ENTER, + worker_scope::DATABASE_INSTANCE, + db_ref_kind::NONE, + ); + self.shutdown.store(true, Ordering::Relaxed); + let consumers = std::mem::take(&mut *self.consumers.write()); + drop(consumers); + emit( + instance, + cid, + component_type::BUFFER_MANAGER, + event_type::DROP_PHASE, + drop_phase::CONSUMERS, + db_ref_kind::NONE, + ); + let force_ram = std::mem::take(&mut *self.force_ram_consumers.write()); + drop(force_ram); + emit( + instance, + cid, + component_type::BUFFER_MANAGER, + event_type::DROP_PHASE, + drop_phase::FORCE_RAM, + db_ref_kind::NONE, + ); + emit( + instance, + cid, + component_type::BUFFER_MANAGER, + event_type::DROP, + worker_scope::DATABASE_INSTANCE, + db_ref_kind::NONE, ); } - self.shutdown.store(true, Ordering::Relaxed); + #[cfg(not(feature = "close-forensics"))] + { + self.shutdown.store(true, Ordering::Relaxed); + } } } diff --git a/crates/grafeo-core/src/index/vector/hnsw.rs b/crates/grafeo-core/src/index/vector/hnsw.rs index a93666643..11c42f83e 100644 --- a/crates/grafeo-core/src/index/vector/hnsw.rs +++ b/crates/grafeo-core/src/index/vector/hnsw.rs @@ -257,13 +257,87 @@ pub struct HnswIndex { #[cfg(feature = "close-forensics")] impl Drop for HnswIndex { fn drop(&mut self) { - grafeo_common::close_forensics::emit( - grafeo_common::close_forensics::active_instance_id(), - self.forensics_component_id, - grafeo_common::close_forensics::component_type::HNSW_INDEX, - grafeo_common::close_forensics::event_type::DROP, - grafeo_common::close_forensics::worker_scope::DATABASE_INSTANCE, - grafeo_common::close_forensics::db_ref_kind::NONE, + use grafeo_common::close_forensics::{ + active_instance_id, component_type, db_ref_kind, drop_phase, emit, event_type, + worker_scope, + }; + let instance = active_instance_id(); + let cid = self.forensics_component_id; + emit( + instance, + cid, + component_type::HNSW_INDEX, + event_type::DROP_ENTER, + worker_scope::DATABASE_INSTANCE, + db_ref_kind::NONE, + ); + // Free fields first, then emit DROP_PHASE so Δt attributes the free work. + let nodes = std::mem::replace( + &mut self.nodes, + RwLock::new(TopologyBackend::new_heap()), + ); + drop(nodes); + emit( + instance, + cid, + component_type::HNSW_INDEX, + event_type::DROP_PHASE, + drop_phase::TOPOLOGY, + db_ref_kind::NONE, + ); + let entry = std::mem::replace( + &mut self.entry_point, + RwLock::new(None), + ); + drop(entry); + emit( + instance, + cid, + component_type::HNSW_INDEX, + event_type::DROP_PHASE, + drop_phase::ENTRY_POINT, + db_ref_kind::NONE, + ); + let max_level = std::mem::replace(&mut self.max_level, RwLock::new(0)); + drop(max_level); + emit( + instance, + cid, + component_type::HNSW_INDEX, + event_type::DROP_PHASE, + drop_phase::MAX_LEVEL, + db_ref_kind::NONE, + ); + let rng = std::mem::replace( + &mut self.rng, + RwLock::new(rand::rngs::StdRng::seed_from_u64(0)), + ); + drop(rng); + emit( + instance, + cid, + component_type::HNSW_INDEX, + event_type::DROP_PHASE, + drop_phase::RNG, + db_ref_kind::NONE, + ); + let config = std::mem::replace(&mut self.config, HnswConfig::default()); + drop(config); + emit( + instance, + cid, + component_type::HNSW_INDEX, + event_type::DROP_PHASE, + drop_phase::CONFIG, + db_ref_kind::NONE, + ); + emit( + instance, + cid, + component_type::HNSW_INDEX, + event_type::DROP, + worker_scope::DATABASE_INSTANCE, + db_ref_kind::NONE, ); } } diff --git a/crates/grafeo-engine/src/database/checkpoint_timer.rs b/crates/grafeo-engine/src/database/checkpoint_timer.rs index 70f306ec4..39d8bd9bb 100644 --- a/crates/grafeo-engine/src/database/checkpoint_timer.rs +++ b/crates/grafeo-engine/src/database/checkpoint_timer.rs @@ -93,17 +93,8 @@ impl CheckpointTimer { #[cfg(feature = "wal")] wal.as_deref(), ); - #[cfg(feature = "close-forensics")] - { - grafeo_common::close_forensics::emit( - grafeo_common::close_forensics::active_instance_id(), - forensics_component_id, - grafeo_common::close_forensics::component_type::CHECKPOINT_TIMER, - grafeo_common::close_forensics::event_type::WORKER_JOINED, - grafeo_common::close_forensics::worker_scope::DATABASE_INSTANCE, - grafeo_common::close_forensics::db_ref_kind::NONE, - ); - } + // JOINED is emitted from stop() after join so it always fires + // even if this thread panics before returning. }) .expect("failed to spawn checkpoint timer thread"); @@ -133,6 +124,18 @@ impl CheckpointTimer { self.shutdown.store(true, Ordering::Release); if let Some(handle) = self.handle.take() { let _ = handle.join(); + #[cfg(feature = "close-forensics")] + { + // Always emit JOINED after join completes (or panics). + grafeo_common::close_forensics::emit( + grafeo_common::close_forensics::active_instance_id(), + self.forensics_component_id, + grafeo_common::close_forensics::component_type::CHECKPOINT_TIMER, + grafeo_common::close_forensics::event_type::WORKER_JOINED, + grafeo_common::close_forensics::worker_scope::DATABASE_INSTANCE, + grafeo_common::close_forensics::db_ref_kind::NONE, + ); + } } } diff --git a/crates/grafeo-engine/src/database/mod.rs b/crates/grafeo-engine/src/database/mod.rs index 5f0bf2c35..c16d573ee 100644 --- a/crates/grafeo-engine/src/database/mod.rs +++ b/crates/grafeo-engine/src/database/mod.rs @@ -3048,16 +3048,50 @@ impl Drop for GrafeoDB { fn drop(&mut self) { #[cfg(feature = "close-forensics")] { - // Tiny non-owning event only — no memory_usage(), I/O, or path clones. - grafeo_common::close_forensics::emit( - grafeo_common::close_forensics::active_instance_id(), - grafeo_common::close_forensics::next_component_id(), - grafeo_common::close_forensics::component_type::GRAFEO_DB, - grafeo_common::close_forensics::event_type::DROP, - grafeo_common::close_forensics::worker_scope::DATABASE_INSTANCE, - grafeo_common::close_forensics::db_ref_kind::NONE, + use grafeo_common::close_forensics::{ + active_instance_id, component_type, db_ref_kind, drop_phase, emit, event_type, + next_component_id, worker_scope, + }; + let instance = active_instance_id(); + let cid = next_component_id(); + emit( + instance, + cid, + component_type::GRAFEO_DB, + event_type::DROP_ENTER, + worker_scope::DATABASE_INSTANCE, + db_ref_kind::NONE, + ); + if let Err(e) = self.close() { + grafeo_error!("Error closing database: {}", e); + } + // Emit CLOSE_FN after close() so Δt ENTER→CLOSE_FN is close wall. + emit( + instance, + cid, + component_type::GRAFEO_DB, + event_type::DROP_PHASE, + drop_phase::CLOSE_FN, + db_ref_kind::NONE, + ); + emit( + instance, + cid, + component_type::GRAFEO_DB, + event_type::DROP_PHASE, + drop_phase::FIELDS, + db_ref_kind::NONE, + ); + emit( + instance, + cid, + component_type::GRAFEO_DB, + event_type::DROP, + worker_scope::DATABASE_INSTANCE, + db_ref_kind::NONE, ); } + #[cfg(not(feature = "close-forensics"))] if let Err(e) = self.close() { grafeo_error!("Error closing database: {}", e); } From 5a82b49d08de769bfc19a0f22baece25e90b6b74 Mon Sep 17 00:00:00 2001 From: jarmen423 Date: Wed, 29 Jul 2026 02:28:13 +0000 Subject: [PATCH 3/3] feat(diagnostics): skip_vector_index_restore for no-HNSW open Allow Config.skip_vector_index_restore to omit VectorStore HNSW topology deserialize and quantized rehydrate so staging can measure lexical-only RSS. --- crates/grafeo-engine/src/config.rs | 13 ++++++++++++ crates/grafeo-engine/src/database/mod.rs | 27 ++++++++++++++++++------ 2 files changed, 33 insertions(+), 7 deletions(-) diff --git a/crates/grafeo-engine/src/config.rs b/crates/grafeo-engine/src/config.rs index bbe6d82cf..3046f1f6b 100644 --- a/crates/grafeo-engine/src/config.rs +++ b/crates/grafeo-engine/src/config.rs @@ -318,6 +318,11 @@ pub struct Config { /// Requires the `encryption` feature flag. Without it, this field is ignored. #[cfg(feature = "encryption")] pub encryption: Option, + + /// Diagnostics-only: skip VectorStore HNSW topology restore and quantized + /// rehydrate on open. Graph/lexical remain available; vector search fails + /// closed. Default `false`. + pub skip_vector_index_restore: bool, } /// Configuration for adaptive query execution. @@ -415,6 +420,7 @@ impl Default for Config { checkpoint_interval: None, #[cfg(feature = "encryption")] encryption: None, + skip_vector_index_restore: false, } } } @@ -487,6 +493,13 @@ impl Config { self } + /// Diagnostics-only: skip VectorStore HNSW topology restore on open. + #[must_use] + pub fn with_skip_vector_index_restore(mut self, skip: bool) -> Self { + self.skip_vector_index_restore = skip; + self + } + /// Sets the adaptive execution configuration. #[must_use] pub fn with_adaptive(mut self, adaptive: AdaptiveConfig) -> Self { diff --git a/crates/grafeo-engine/src/database/mod.rs b/crates/grafeo-engine/src/database/mod.rs index c16d573ee..f80029930 100644 --- a/crates/grafeo-engine/src/database/mod.rs +++ b/crates/grafeo-engine/src/database/mod.rs @@ -422,6 +422,7 @@ impl GrafeoDB { &catalog, #[cfg(feature = "triple-store")] &rdf_store, + config.skip_vector_index_restore, )?; #[cfg(feature = "compact-store")] { @@ -480,6 +481,7 @@ impl GrafeoDB { &catalog, #[cfg(feature = "triple-store")] &rdf_store, + config.skip_vector_index_restore, )?; #[cfg(feature = "compact-store")] { @@ -681,7 +683,9 @@ impl GrafeoDB { // Must not run inside CatalogSection::deserialize (misses CompactStore // base embeddings). Topology is reused; no full HNSW rebuild. #[cfg(all(feature = "lpg", feature = "vector-index"))] - db.rehydrate_quantized_vector_indexes()?; + if !db.config.skip_vector_index_restore { + db.rehydrate_quantized_vector_indexes()?; + } // Start periodic checkpoint timer if configured #[cfg(all(feature = "grafeo-file", feature = "lpg"))] @@ -1728,7 +1732,10 @@ impl GrafeoDB { store: &Arc, catalog: &Arc, #[cfg(feature = "triple-store")] rdf_store: &Arc, + skip_vector_index_restore: bool, ) -> Result<()> { + #[cfg(not(feature = "vector-index"))] + let _ = skip_vector_index_restore; use grafeo_common::storage::{Section, SectionType}; let dir = fm.read_section_directory()?.ok_or_else(|| { @@ -1776,13 +1783,19 @@ impl GrafeoDB { // Restore HNSW topology (if vector indexes exist in both catalog and section) #[cfg(feature = "vector-index")] - if let Some(entry) = dir.find(SectionType::VectorStore) { - let data = fm.read_section_data(entry)?; - let indexes = store.vector_index_entries(); - if !indexes.is_empty() { - let mut section = grafeo_core::index::vector::VectorStoreSection::new(indexes); - section.deserialize(&data)?; + if !skip_vector_index_restore { + if let Some(entry) = dir.find(SectionType::VectorStore) { + let data = fm.read_section_data(entry)?; + let indexes = store.vector_index_entries(); + if !indexes.is_empty() { + let mut section = grafeo_core::index::vector::VectorStoreSection::new(indexes); + section.deserialize(&data)?; + } } + } else { + grafeo_warn!( + "skip_vector_index_restore=true: skipping VectorStore HNSW topology restore" + ); } // Restore BM25 postings (if text indexes exist in both catalog and section)