perf: clean-close fast path (skip full LPG serialize)

crates/grafeo-engine/src/database/flush.rs

@@ -21,10 +21,13 @@ use grafeo_storage::file::GrafeoFileManager;
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[non_exhaustive]
 pub enum FlushReason {
-    /// Periodic checkpoint (timer-driven) or database close.
-    #[allow(dead_code)] // Used by async_ops (async-storage feature)
+    /// Periodic checkpoint (timer-driven). Dirty sections only.
+    /// Clean database close also uses this when no container rewrite is required
+    /// (see `GrafeoDB::close_needs_full_checkpoint`).
+    #[allow(dead_code)] // Used by async_ops (async-storage feature) and close fast path
     Checkpoint,
-    /// User-initiated `CHECKPOINT` command or `wal_checkpoint()` API.
+    /// User-initiated `CHECKPOINT` / `wal_checkpoint()` API, or close when
+    /// mutations/recovery require a full container rewrite.
     Explicit,
 }
 

crates/grafeo-engine/src/database/index.rs

@@ -34,13 +34,20 @@ impl super::GrafeoDB {
     /// ```
     pub fn create_property_index(&self, property: &str) {
         self.lpg_store().create_property_index(property);
+        #[cfg(feature = "grafeo-file")]
+        self.mark_container_flush_required();
     }
 
     /// Drops an index on a node property.
     ///
     /// Returns `true` if the index existed and was removed.
     pub fn drop_property_index(&self, property: &str) -> bool {
-        self.lpg_store().drop_property_index(property)
+        let removed = self.lpg_store().drop_property_index(property);
+        #[cfg(feature = "grafeo-file")]
+        if removed {
+            self.mark_container_flush_required();
+        }
+        removed
     }
 
     /// Returns `true` if the property has an index.
@@ -176,6 +183,8 @@ impl super::GrafeoDB {
                     "Empty vector index created: :{label}({property}) - 0 vectors, {d} dimensions, metric={metric_name}",
                     metric_name = metric.name()
                 );
+                #[cfg(feature = "grafeo-file")]
+                self.mark_container_flush_required();
                 Ok(())
             } else {
                 Err(grafeo_common::utils::error::Error::Internal(format!(
@@ -246,6 +255,8 @@ impl super::GrafeoDB {
             metric_name = metric.name()
         );
 
+        #[cfg(feature = "grafeo-file")]
+        self.mark_container_flush_required();
         Ok(())
     }
 
@@ -308,6 +319,8 @@ impl super::GrafeoDB {
         let removed = self.lpg_store().remove_vector_index(label, property);
         if removed {
             grafeo_info!("Vector index dropped: :{label}({property})");
+            #[cfg(feature = "grafeo-file")]
+            self.mark_container_flush_required();
         }
         removed
     }
@@ -460,6 +473,8 @@ impl super::GrafeoDB {
 
         self.lpg_store()
             .add_text_index(label, property, Arc::new(RwLock::new(index)));
+        #[cfg(feature = "grafeo-file")]
+        self.mark_container_flush_required();
         Ok(())
     }
 
@@ -468,7 +483,12 @@ impl super::GrafeoDB {
     /// Returns `true` if the index existed and was removed.
     #[cfg(feature = "text-index")]
     pub fn drop_text_index(&self, label: &str, property: &str) -> bool {
-        self.lpg_store().remove_text_index(label, property)
+        let removed = self.lpg_store().remove_text_index(label, property);
+        #[cfg(feature = "grafeo-file")]
+        if removed {
+            self.mark_container_flush_required();
+        }
+        removed
     }
 
     /// Rebuilds a text index by re-scanning all matching nodes.

crates/grafeo-engine/src/database/mod.rs

@@ -185,6 +185,13 @@ pub struct GrafeoDB {
     /// Whether this database is open in read-only mode.
     /// When true, sessions automatically enforce read-only transactions.
     read_only: bool,
+    /// When true, close must rewrite the `.grafeo` container (full section
+    /// serialize). Set when WAL recovery applied unrecovered records, or when
+    /// a container-affecting mutation bypasses session WAL `record_count`
+    /// (e.g. vector/text index DDL). Session graph mutations already bump WAL
+    /// `record_count`; see [`Self::close_needs_full_checkpoint`].
+    #[cfg(feature = "grafeo-file")]
+    container_flush_required: std::sync::atomic::AtomicBool,
     /// Named graph projections (virtual subgraphs), shared with sessions.
     projections:
         Arc<RwLock<std::collections::HashMap<String, Arc<grafeo_core::graph::GraphProjection>>>>,
@@ -406,6 +413,12 @@ impl GrafeoDB {
             Vec<grafeo_common::types::EdgeId>,
         )> = None;
 
+        // Sidecar/legacy WAL recovery can leave unrecovered mutations only in
+        // RAM while WalManager::record_count resets to 0 on reopen. Track that
+        // so close still forces a full container checkpoint.
+        #[cfg(feature = "grafeo-file")]
+        let mut recovered_wal_needs_flush = false;
+
         // --- Single-file format (.grafeo) ---
         #[cfg(feature = "grafeo-file")]
         let file_manager: Option<Arc<GrafeoFileManager>> = if is_read_only {
@@ -504,6 +517,9 @@ impl GrafeoDB {
                 if config.wal_enabled && fm.has_sidecar_wal() {
                     let recovery = WalRecovery::new(fm.sidecar_wal_path());
                     let records = recovery.recover()?;
+                    if !records.is_empty() {
+                        recovered_wal_needs_flush = true;
+                    }
                     Self::apply_wal_records(
                         &store,
                         &catalog,
@@ -556,6 +572,10 @@ impl GrafeoDB {
                 if !is_single_file && wal_path.exists() {
                     let recovery = WalRecovery::new(&wal_path);
                     let records = recovery.recover()?;
+                    #[cfg(feature = "grafeo-file")]
+                    if !records.is_empty() {
+                        recovered_wal_needs_flush = true;
+                    }
                     Self::apply_wal_records(
                         &store,
                         &catalog,
@@ -656,6 +676,8 @@ impl GrafeoDB {
             current_graph: RwLock::new(None),
             current_schema: RwLock::new(None),
             read_only: is_read_only,
+            #[cfg(feature = "grafeo-file")]
+            container_flush_required: std::sync::atomic::AtomicBool::new(recovered_wal_needs_flush),
             projections: Arc::new(RwLock::new(std::collections::HashMap::new())),
             #[cfg(all(feature = "compact-store", feature = "lpg"))]
             layered_store: None,
@@ -720,6 +742,15 @@ impl GrafeoDB {
         // the consumers we're about to spill.
         db.apply_force_disk_overrides();
 
+        // Open-path index restore may call create_vector_index / similar
+        // helpers that mark container_flush_required. Reset to the recovery
+        // signal only so a clean no-write session can take the close fast path.
+        #[cfg(feature = "grafeo-file")]
+        db.container_flush_required.store(
+            recovered_wal_needs_flush,
+            std::sync::atomic::Ordering::Release,
+        );
+
         Ok(db)
     }
 
@@ -810,6 +841,8 @@ impl GrafeoDB {
             current_graph: RwLock::new(None),
             current_schema: RwLock::new(None),
             read_only: false,
+            #[cfg(feature = "grafeo-file")]
+            container_flush_required: std::sync::atomic::AtomicBool::new(false),
             projections: Arc::new(RwLock::new(std::collections::HashMap::new())),
             #[cfg(all(feature = "compact-store", feature = "lpg"))]
             layered_store: None,
@@ -901,6 +934,8 @@ impl GrafeoDB {
             current_graph: RwLock::new(None),
             current_schema: RwLock::new(None),
             read_only: true,
+            #[cfg(feature = "grafeo-file")]
+            container_flush_required: std::sync::atomic::AtomicBool::new(false),
             projections: Arc::new(RwLock::new(std::collections::HashMap::new())),
             #[cfg(all(feature = "compact-store", feature = "lpg"))]
             layered_store: None,
@@ -2433,13 +2468,25 @@ impl GrafeoDB {
             if let Some(ref wal) = self.wal {
                 wal.sync()?;
             }
-            let flush_result = self.checkpoint_to_file(fm, flush::FlushReason::Explicit)?;
+
+            // Clean-close fast path: ephemeral Section wrappers from
+            // build_sections() start with dirty=false, so FlushReason::Explicit
+            // always re-serializes the full LPG (dominates ~137s sidecar close).
+            // Prefer Checkpoint (dirty-only, usually a no-op) when this session
+            // has no container-affecting work; force Explicit when WAL-disabled,
+            // WAL has new records, or recovery/index DDL marked a flush.
+            let initial_reason = if self.close_needs_full_checkpoint() {
+                flush::FlushReason::Explicit
+            } else {
+                flush::FlushReason::Checkpoint
+            };
+            let flush_result = self.checkpoint_to_file(fm, initial_reason)?;
 
             // Safety check: if WAL has records but the checkpoint was a no-op
             // (zero sections written), the container file may not contain the
             // latest data. This can happen when sections are not marked dirty
-            // despite mutations going through the WAL. Force-dirty all sections
-            // and retry before removing the sidecar.
+            // despite mutations going through the WAL. Force a full Explicit
+            // flush and retry before removing the sidecar.
             #[cfg(feature = "wal")]
             let flush_result = if flush_result.sections_written == 0 {
                 if let Some(ref wal) = self.wal {
@@ -2523,6 +2570,12 @@ impl GrafeoDB {
     /// Logs a WAL record if WAL is enabled.
     #[cfg(feature = "wal")]
     pub(super) fn log_wal(&self, record: &WalRecord) -> Result<()> {
+        // Direct CRUD / persistence helpers go through this path. Session
+        // mutations write the WAL handle directly and are covered by
+        // record_count on close; still mark flush-required so wal_disabled
+        // databases (wal=None) force Explicit checkpoint on close.
+        #[cfg(feature = "grafeo-file")]
+        self.mark_container_flush_required();
         if let Some(ref wal) = self.wal {
             wal.log(record)?;
         }
@@ -2949,6 +3002,45 @@ impl GrafeoDB {
         backup::do_restore_to_epoch(backup_dir, target_epoch, output_path)
     }
 
+    /// Marks that close must rewrite the `.grafeo` container.
+    ///
+    /// Used for container-affecting mutations that do not bump WAL
+    /// `record_count` (index DDL) and for tests.
+    #[cfg(feature = "grafeo-file")]
+    pub(crate) fn mark_container_flush_required(&self) {
+        self.container_flush_required
+            .store(true, std::sync::atomic::Ordering::Release);
+    }
+
+    /// Whether close must force a full Explicit section serialize.
+    ///
+    /// Clean-close fast path is only taken when this returns false: WAL is
+    /// enabled, this session logged no WAL records, and no recovery/index DDL
+    /// required a container rewrite. Ephemeral `Section` wrappers from
+    /// [`Self::build_sections`] start clean, so `FlushReason::Checkpoint` then
+    /// writes zero sections and skips LPG serialization.
+    #[cfg(feature = "grafeo-file")]
+    fn close_needs_full_checkpoint(&self) -> bool {
+        if self
+            .container_flush_required
+            .load(std::sync::atomic::Ordering::Acquire)
+        {
+            return true;
+        }
+        #[cfg(feature = "wal")]
+        {
+            // WAL disabled (None) cannot use record_count as a dirty signal.
+            match self.wal.as_ref() {
+                None => true,
+                Some(wal) => wal.record_count() > 0,
+            }
+        }
+        #[cfg(not(feature = "wal"))]
+        {
+            true
+        }
+    }
+
     /// Writes the current database state to the `.grafeo` file using the unified flush.
     ///
     /// Does NOT remove the sidecar WAL: callers that want to clean up