reddb_server/storage/unified/store/

impl_entities.rs

use super::*;

impl UnifiedStore {
    pub(crate) fn canonical_collection(name: &str) -> &str {
        reddb_engine::storage::system_collection_names::canonical_or_self(name)
    }

    /// Vault versions are WAL-backed in memory until a committed checkpoint.
    /// Preparing a commit must not dirty pager pages: eviction or an auth
    /// flush could otherwise persist a version before its WAL batch exists.
    pub(crate) fn insert_vault_version(
        &self,
        collection: &str,
        entity: UnifiedEntity,
        metadata: Metadata,
    ) -> Result<(), StoreError> {
        assert!(matches!(entity.data, EntityData::Row(_)));
        let manager = self
            .get_collection(collection)
            .ok_or_else(|| StoreError::CollectionNotFound(collection.into()))?;
        self.pending_vault_versions.write().insert(entity.id);
        let id = manager.insert(entity)?;
        self.register_entity_id(id);
        if !metadata.fields.is_empty() {
            manager.set_metadata(id, metadata)?;
        }
        let entity = manager
            .get(id)
            .expect("inserted vault version remains in memory");
        self.persist_entities_to_pager_wal_only(collection, &[entity])
    }

    pub(crate) fn persist_entities_to_pager(
        &self,
        collection: &str,
        entities: &[UnifiedEntity],
    ) -> Result<(), StoreError> {
        self.persist_entities_impl(collection, entities, /* skip_btree= */ false)
    }

    /// Same as `persist_entities_to_pager` but takes `&[&UnifiedEntity]` so
    /// callers that already hold references (SQL UPDATE fast path) don't
    /// have to clone a `Vec<UnifiedEntity>` just to satisfy the signature.
    pub(crate) fn persist_entity_refs_to_pager(
        &self,
        collection: &str,
        entities: &[&UnifiedEntity],
    ) -> Result<(), StoreError> {
        self.persist_entity_refs_impl(collection, entities, /* skip_btree= */ false)
    }

    /// Reference-taking WAL-only variant — see
    /// `persist_entities_to_pager_wal_only` for semantics.
    pub(crate) fn persist_entity_refs_to_pager_wal_only(
        &self,
        collection: &str,
        entities: &[&UnifiedEntity],
    ) -> Result<(), StoreError> {
        self.persist_entity_refs_impl(collection, entities, /* skip_btree= */ true)
    }

    fn persist_entity_refs_impl(
        &self,
        collection: &str,
        entities: &[&UnifiedEntity],
        skip_btree: bool,
    ) -> Result<(), StoreError> {
        let collection = Self::canonical_collection(collection);
        let skip_btree = skip_btree || self.config.resident_memory_bytes_max.is_some();
        if entities.is_empty() {
            return Ok(());
        }
        if self.pager.is_none() && self.config.embedded_wal_path.is_none() {
            return Ok(());
        }
        let fv = self.format_version();
        let manager = self
            .get_collection(collection)
            .ok_or_else(|| StoreError::CollectionNotFound(collection.to_string()))?;
        let mut serialized: Vec<(Vec<u8>, Vec<u8>)> = entities
            .iter()
            .map(|entity| {
                let metadata = manager.get_metadata(entity.id);
                (
                    entity.id.raw().to_be_bytes().to_vec(),
                    Self::serialize_entity_record(entity, metadata.as_ref(), fv),
                )
            })
            .collect();
        serialized.sort_by(|a, b| a.0.cmp(&b.0));

        if !skip_btree
            && !Self::deferred_store_wal_capture_active()
            && self.config.resident_memory_bytes_max.is_none()
        {
            let Some(pager) = &self.pager else {
                let records: Vec<Vec<u8>> =
                    serialized.into_iter().map(|(_id, record)| record).collect();
                self.finish_paged_write([StoreWalAction::BulkUpsertEntityRecords {
                    collection: collection.to_string(),
                    records,
                }])?;
                return Ok(());
            };
            let mut btree_indices = self.btree_indices.write();
            let btree = btree_indices
                .entry(collection.to_string())
                .or_insert_with(|| Arc::new(BTree::new(Arc::clone(pager))));
            let root_before = btree.root_page_id();
            btree.upsert_batch_sorted(&serialized).map_err(|e| {
                StoreError::Io(std::io::Error::other(format!("B-tree upsert error: {}", e)))
            })?;
            let root_after = btree.root_page_id();
            drop(btree_indices);
            if root_before != root_after {
                self.mark_paged_registry_dirty();
            }
        }
        let records: Vec<Vec<u8>> = serialized.into_iter().map(|(_id, record)| record).collect();
        self.finish_paged_write_for_projection(
            [StoreWalAction::BulkUpsertEntityRecords {
                collection: collection.to_string(),
                records,
            }],
            skip_btree,
        )?;
        Ok(())
    }

    /// PG-HOT-like UPDATE persist: emit the WAL record so the update
    /// survives a crash, but skip the in-line B-tree upsert. Reads
    /// continue to see the new entity because `manager.get()` prefers
    /// the segment over the B-tree; WAL replay re-applies the upsert
    /// to the B-tree on recovery so the two views converge before
    /// any stale B-tree page can be surfaced.
    ///
    /// Safe to use only when the caller has already updated the
    /// segment in-place (so live reads see the new value). Cuts out
    /// the dominant cost of single-row UPDATEs: per-entity serialize
    /// + leaf descent + page decompress / compress / write_page.
    pub(crate) fn persist_entities_to_pager_wal_only(
        &self,
        collection: &str,
        entities: &[UnifiedEntity],
    ) -> Result<(), StoreError> {
        self.persist_entities_impl(collection, entities, /* skip_btree= */ true)
    }

    fn persist_entities_impl(
        &self,
        collection: &str,
        entities: &[UnifiedEntity],
        skip_btree: bool,
    ) -> Result<(), StoreError> {
        let collection = Self::canonical_collection(collection);
        let skip_btree = skip_btree || self.config.resident_memory_bytes_max.is_some();
        if entities.is_empty() {
            return Ok(());
        }

        if self.pager.is_none() && self.config.embedded_wal_path.is_none() {
            return Ok(());
        }

        let fv = self.format_version();
        let manager = self
            .get_collection(collection)
            .ok_or_else(|| StoreError::CollectionNotFound(collection.to_string()))?;
        let mut serialized: Vec<(Vec<u8>, Vec<u8>)> = entities
            .iter()
            .map(|entity| {
                let metadata = manager.get_metadata(entity.id);
                (
                    entity.id.raw().to_be_bytes().to_vec(),
                    Self::serialize_entity_record(entity, metadata.as_ref(), fv),
                )
            })
            .collect();
        // u64 IDs encoded as big-endian — lex order = numeric order.
        serialized.sort_by(|a, b| a.0.cmp(&b.0));

        if !skip_btree
            && !Self::deferred_store_wal_capture_active()
            && self.config.resident_memory_bytes_max.is_none()
        {
            let Some(pager) = &self.pager else {
                let records: Vec<Vec<u8>> =
                    serialized.into_iter().map(|(_id, record)| record).collect();
                self.finish_paged_write([StoreWalAction::BulkUpsertEntityRecords {
                    collection: collection.to_string(),
                    records,
                }])?;
                return Ok(());
            };
            let mut btree_indices = self.btree_indices.write();
            let btree = btree_indices
                .entry(collection.to_string())
                .or_insert_with(|| Arc::new(BTree::new(Arc::clone(pager))));
            let root_before = btree.root_page_id();

            // Walks each distinct leaf once, applies all in-place overwrites
            // that belong there under one read+write. Keys that miss or grow
            // fall back to the per-key `upsert` path internally.
            btree.upsert_batch_sorted(&serialized).map_err(|e| {
                StoreError::Io(std::io::Error::other(format!("B-tree upsert error: {}", e)))
            })?;
            let root_after = btree.root_page_id();
            drop(btree_indices);
            if root_before != root_after {
                self.mark_paged_registry_dirty();
            }
        }
        // One WAL action covering the whole bulk. Previously each entity
        // produced its own `UpsertEntityRecord` action, which then became
        // its own `WalRecord::PageWrite` triple inside
        // `StoreCommitCoordinator::append_actions` — N WAL records per
        // bulk. The batched variant is one WAL record per bulk.
        //
        // WAL is emitted even in the `skip_btree` variant — recovery
        // replay reapplies the upsert to the B-tree, so durability
        // is unchanged.
        let records: Vec<Vec<u8>> = serialized.into_iter().map(|(_id, record)| record).collect();
        self.finish_paged_write_for_projection(
            [StoreWalAction::BulkUpsertEntityRecords {
                collection: collection.to_string(),
                records,
            }],
            skip_btree,
        )?;

        Ok(())
    }

    /// Insert a label→entity_id mapping into the graph label index.
    pub(crate) fn update_graph_label_index(
        &self,
        collection: &str,
        label: &str,
        entity_id: EntityId,
    ) {
        self.resident
            .update_graph_label_index(collection, label, entity_id)
    }

    /// Remove a specific entity_id from the graph label index (called on delete).
    pub(crate) fn remove_from_graph_label_index(&self, collection: &str, entity_id: EntityId) {
        self.resident
            .remove_from_graph_label_index(collection, entity_id)
    }

    pub(crate) fn remove_from_graph_label_index_batch(
        &self,
        collection: &str,
        entity_ids: &[EntityId],
    ) {
        self.resident
            .remove_from_graph_label_index_batch(collection, entity_ids)
    }

    /// Look up entity IDs for a graph node label across all collections.
    pub fn lookup_graph_nodes_by_label(&self, label: &str) -> Vec<EntityId> {
        self.resident.lookup_graph_nodes_by_label(label)
    }

    /// Look up entity IDs for a graph node label scoped to a single collection.
    pub fn lookup_graph_nodes_by_label_in(&self, collection: &str, label: &str) -> Vec<EntityId> {
        self.resident
            .lookup_graph_nodes_by_label_in(collection, label)
    }

    pub fn create_collection(&self, name: impl Into<String>) -> Result<(), StoreError> {
        let name = reddb_engine::storage::system_collection_names::canonical_owned(name.into());
        // Collection creation publishes the same manifest that flush and
        // retention update. Fence the entire create, including registry/WAL.
        let _publication = self.vault_publication_lock.lock();
        let mut collections = self.resident.collections.write();

        if collections.contains_key(&name) {
            return Err(StoreError::CollectionExists(name));
        }

        let manager = SegmentManager::with_config(&name, self.config.manager_config.clone());
        collections.insert(name.clone(), Arc::new(manager));
        drop(collections);

        let create = match self.prepare_operational_collection_create(&name) {
            Ok(create) => create,
            Err(err) => {
                let mut collections = self.resident.collections.write();
                collections.remove(&name);
                return Err(err);
            }
        };
        if let Some(create) = create {
            if let Err(err) = create.publish().map_err(StoreError::Io) {
                let mut collections = self.resident.collections.write();
                collections.remove(&name);
                return Err(err);
            }
        }

        self.mark_paged_registry_dirty();
        self.finish_paged_write([StoreWalAction::CreateCollection { name }])?;

        Ok(())
    }

    /// Get or create a collection
    pub fn get_or_create_collection(&self, name: impl Into<String>) -> Arc<SegmentManager> {
        self.resident.get_or_create_collection(
            reddb_engine::storage::system_collection_names::canonical_owned(name.into()),
            &self.config.manager_config,
        )
    }

    /// Get a collection
    pub fn get_collection(&self, name: &str) -> Option<Arc<SegmentManager>> {
        self.resident
            .get_collection(Self::canonical_collection(name))
    }

    /// Get the context index for cross-structure search.
    pub fn context_index(&self) -> &ContextIndex {
        self.resident.context_index()
    }

    /// Drain WAL-replayed `VectorInsert` records for `collection`
    /// (issue #694). Returns `None` when nothing was captured at open
    /// time (in-memory mode, fresh database, or non-turbo collection).
    /// One-shot: the caller owns the records after this call, the
    /// store's internal buffer is cleared for that collection.
    pub fn take_replayed_turbo_inserts(&self, collection: &str) -> Option<Vec<(u64, Vec<f32>)>> {
        let mut map = self.replayed_turbo_inserts.lock();
        map.remove(Self::canonical_collection(collection))
    }

    pub(crate) fn take_replayed_probabilistic_deltas(&self) -> Vec<(u8, u8, String, Vec<Vec<u8>>)> {
        let mut deltas = self.replayed_probabilistic_deltas.lock();
        std::mem::take(&mut *deltas)
    }

    /// Set multiple config KV pairs at once from a JSON tree.
    /// Keys are flattened with dot-notation: `{"a":{"b":1}}` → `a.b = 1`.
    ///
    /// `red_config` is last-writer-wins (ADR 0014): each write replaces the
    /// key's earlier rows. Appending instead grew the store on every boot,
    /// since boot re-seeds keys such as `red.system.*`.
    pub fn set_config_tree(&self, prefix: &str, json: &crate::serde_json::Value) -> usize {
        // Historical best-effort adapter. Authoritative callers propagate the
        // result of try_set_config_tree instead of ignoring failed persistence.
        self.try_set_config_tree(prefix, json).unwrap_or_default()
    }

    pub fn try_set_config_tree(
        &self,
        prefix: &str,
        json: &crate::serde_json::Value,
    ) -> Result<usize, StoreError> {
        let _publication = self.vault_publication_lock.lock();
        let manager = self.get_or_create_collection("red_config");
        let mut pairs = Vec::new();
        flatten_config_json(prefix, json, &mut pairs);
        // Existing rows per key, with the value each one holds.
        let mut previous: HashMap<String, Vec<(EntityId, Option<reddb_types::Value>)>> = pairs
            .iter()
            .map(|(key, _)| (key.clone(), Vec::new()))
            .collect();
        let mut capture = |entity: &UnifiedEntity| {
            if let EntityData::Row(row) = &entity.data {
                if let Some(reddb_types::Value::Text(key)) = row.get_field("key") {
                    if let Some(rows) = previous.get_mut(key.as_ref()) {
                        rows.push((entity.id, row.get_field("value").cloned()));
                    }
                }
            }
            true
        };
        if manager.has_backing() {
            for (key, _) in &pairs {
                manager.try_for_each_key_entity(key, &mut capture)?;
            }
        } else {
            manager.try_for_each_entity(usize::MAX, &mut capture)?;
        }
        let mut saved = 0;
        for (key, value) in pairs {
            let stale = previous.remove(&key).unwrap_or_default();
            // Rewriting a key with the value it already holds is a no-op.
            // Boot re-seeds the same config every time, and each write is a
            // durable WAL append with its own fsync.
            if let [(_, Some(current))] = stale.as_slice() {
                if *current == value {
                    saved += 1;
                    continue;
                }
            }
            let stale: Vec<EntityId> = stale.into_iter().map(|(id, _)| id).collect();
            // Insert before removing the old rows: a failure in between
            // leaves a duplicate that readers resolve to the newest row,
            // never a missing key.
            let entity = UnifiedEntity::new(
                EntityId::new(0),
                EntityKind::TableRow {
                    table: Arc::from("red_config"),
                    row_id: 0,
                },
                EntityData::Row(RowData {
                    columns: Vec::new(),
                    named: Some(
                        [
                            ("key".to_string(), reddb_types::Value::text(key)),
                            ("value".to_string(), value),
                        ]
                        .into_iter()
                        .collect(),
                    ),
                    schema: None,
                }),
            );
            self.insert_auto("red_config", entity)?;
            saved += 1;
            if !stale.is_empty() {
                self.delete_batch("red_config", &stale)?;
            }
        }
        Ok(saved)
    }

    /// Read a single config value from `red_config` by dot-notation key.
    /// When a key has several rows the newest one wins, like every other
    /// `red_config` reader.
    pub fn get_config(&self, key: &str) -> Option<reddb_types::Value> {
        // Historical best-effort adapter; bootstrap and authoritative runtime
        // callers use try_get_config so a fault cannot become a default value.
        self.try_get_config(key).ok().flatten()
    }

    pub fn try_get_config(&self, key: &str) -> Result<Option<reddb_types::Value>, StoreError> {
        let Some(manager) = self.get_collection("red_config") else {
            return Ok(None);
        };
        let _publication = manager
            .has_backing()
            .then(|| self.vault_publication_lock.lock());
        let mut latest: Option<(u64, reddb_types::Value)> = None;
        manager.try_for_each_key_entity(key, |entity| {
            if let EntityData::Row(row) = &entity.data {
                let key_matches = matches!(row.get_field("key"), Some(reddb_types::Value::Text(stored)) if stored.as_ref() == key);
                if key_matches && latest.as_ref().is_none_or(|(id, _)| entity.id.raw() >= *id) {
                    if let Some(value) = row.get_field("value") {
                        latest = Some((entity.id.raw(), value.clone()));
                    }
                }
            }
            true
        })?;
        Ok(latest.map(|(_, value)| value))
    }

    /// Replace the opaque store-level auxiliary metadata blob. Persisted
    /// inside the binary dump (store format V10+); see the `aux_metadata`
    /// field. RedDB uses it to carry collection contracts through the
    /// single-file artifact.
    pub fn set_aux_metadata(&self, bytes: Vec<u8>) {
        self.resident.set_aux_metadata(bytes)
    }

    /// Persist single-file contracts through the same commit fence as rows.
    /// Memory-only stores retain the same auxiliary snapshot behavior.
    pub(crate) fn persist_aux_metadata(&self, bytes: Vec<u8>) -> Result<(), StoreError> {
        let bytes = reddb_engine::catalog::codec::canonicalize_collection_contract_blob(&bytes)
            .map_err(|err| StoreError::Serialization(err.to_string()))?;
        let _publication = self.vault_publication_lock.lock();
        self.finish_paged_write([StoreWalAction::SetAuxMetadata {
            bytes: bytes.clone(),
        }])?;
        self.resident.set_aux_metadata(bytes);
        Ok(())
    }

    /// Read the opaque store-level auxiliary metadata blob. Empty when unset.
    pub fn aux_metadata(&self) -> Vec<u8> {
        self.resident.aux_metadata()
    }

    /// List all collections
    pub fn list_collections(&self) -> Vec<String> {
        self.resident.list_collections()
    }

    /// Drop a collection
    pub fn drop_collection(&self, name: &str) -> Result<(), StoreError> {
        let name = Self::canonical_collection(name);
        let _vault_guard = self.vault_publication_lock.lock();
        let drop = self.prepare_operational_collection_drop(name)?;
        let manager = {
            let mut collections = self.resident.collections.write();

            collections
                .remove(name)
                .ok_or_else(|| StoreError::CollectionNotFound(name.to_string()))?
        };

        let entities = manager.query_all(|_| true);
        let entity_ids: Vec<EntityId> = entities.iter().map(|entity| entity.id).collect();

        for entity_id in &entity_ids {
            self.resident.context_index.remove_entity(*entity_id);
            let _ = self.unindex_cross_refs(*entity_id);
        }

        if let Some(tree) = self.btree_indices.write().remove(name) {
            self.retire_btree(tree);
        }
        if let Some(index) = self.key_indices.write().remove(name) {
            for tree in index.trees() {
                self.retire_btree(Arc::clone(tree));
            }
        }
        self.secondary_index_specs.write().remove(name);

        self.resident
            .entity_cache
            .retain(|entity_id, (collection, _)| {
                collection != name && !entity_ids.iter().any(|id| id.raw() == entity_id)
            });

        self.resident.cross_refs.write().retain(|source_id, refs| {
            refs.retain(|(target_id, _, target_collection)| {
                target_collection != name && !entity_ids.iter().any(|id| id == target_id)
            });
            !entity_ids.iter().any(|id| id == source_id)
        });

        self.resident
            .reverse_refs
            .write()
            .retain(|target_id, refs| {
                refs.retain(|(source_id, _, source_collection)| {
                    source_collection != name && !entity_ids.iter().any(|id| id == source_id)
                });
                !entity_ids.iter().any(|id| id == target_id)
            });

        self.mark_paged_registry_dirty();
        self.finish_paged_write([StoreWalAction::DropCollection {
            name: name.to_string(),
        }])?;
        if let Some(drop) = drop {
            drop.finish().map_err(StoreError::Io)?;
        }

        Ok(())
    }

    /// Insert an entity into a collection
    pub fn insert(&self, collection: &str, entity: UnifiedEntity) -> Result<EntityId, StoreError> {
        let collection = Self::canonical_collection(collection);
        let manager = self
            .get_collection(collection)
            .ok_or_else(|| StoreError::CollectionNotFound(collection.to_string()))?;

        let mut entity = entity;
        entity.canonicalize_system_collection_names();
        if entity.id.raw() == 0 {
            entity.id = self.next_entity_id();
        } else {
            self.register_entity_id(entity.id);
        }
        // Assign per-table sequential row_id if not set
        if let EntityKind::TableRow { ref mut row_id, .. } = entity.kind {
            if *row_id == 0 {
                *row_id = manager.next_row_id();
            } else {
                manager.register_row_id(*row_id);
            }
        }
        entity.ensure_table_logical_id();
        // Capture graph node label before entity is moved into the manager
        let graph_node_label: Option<String> = if let EntityKind::GraphNode(ref node) = entity.kind
        {
            Some(node.label.clone())
        } else {
            None
        };

        let id = manager.insert(entity)?;
        self.register_entity_id(id);

        // Update graph label index for GraphNode entities
        if let Some(ref label) = graph_node_label {
            self.update_graph_label_index(collection, label, id);
        }

        // Captured transactions stay WAL-only until commit; pager flushes must
        // never make an unfinished row durable.
        let mut registry_dirty = false;
        if let Some(pager) = self.pager.as_ref().filter(|_| {
            !Self::deferred_store_wal_capture_active()
                && self.config.resident_memory_bytes_max.is_none()
        }) {
            if let Some(entity) = manager.get(id) {
                let mut btree_indices = self.btree_indices.write();
                let btree = btree_indices
                    .entry(collection.to_string())
                    .or_insert_with(|| Arc::new(BTree::new(Arc::clone(pager))));
                let root_before = btree.root_page_id();

                let key = id.raw().to_be_bytes();
                let metadata = manager.get_metadata(id);
                let value = Self::serialize_entity_record(
                    &entity,
                    metadata.as_ref(),
                    self.format_version(),
                );
                // Ignore duplicate key errors (update scenario)
                let _ = btree.insert(&key, &value);
                registry_dirty = root_before != btree.root_page_id();
            }
        }

        // Index cross-references if enabled
        if self.config.auto_index_refs {
            if let Some(entity) = manager.get(id) {
                self.index_cross_refs(&entity, collection)?;
            }
        }

        // Perf: skip WAL-action construction when the store is
        // pagerless. For in-memory benchmarks this saved another
        // `manager.get(id)` + `serialize_entity_record` per call.
        if self.pager.is_some() || self.config.embedded_wal_path.is_some() {
            let actions = manager
                .get(id)
                .map(|entity| {
                    let metadata = manager.get_metadata(id);
                    vec![StoreWalAction::upsert_entity(
                        collection,
                        &entity,
                        metadata.as_ref(),
                        self.format_version(),
                    )]
                })
                .unwrap_or_default();
            if registry_dirty {
                self.mark_paged_registry_dirty();
            }
            self.finish_paged_write(actions)?;
        }

        Ok(id)
    }

    /// Turbo bulk insert — optimized fast path.
    ///
    /// Single lock for the entire batch. Skips bloom filter,
    /// context index, and cross-ref indexing. B-tree writes are batched.
    pub fn bulk_insert(
        &self,
        collection: &str,
        mut entities: Vec<UnifiedEntity>,
    ) -> Result<Vec<EntityId>, StoreError> {
        let collection = Self::canonical_collection(collection);
        for entity in &mut entities {
            entity.canonicalize_system_collection_names();
        }
        // A transaction can begin after checkpoint visibility is frozen.
        // Fence the entire row batch, including its captured or committed WAL.
        let _publication = self.vault_publication_lock.lock();
        // REDDB_BULK_TIMING=1 prints a per-call breakdown of the bulk
        // insert path to stderr. Off by default — used by the reddb
        // benchmark harness to locate ingest bottlenecks.
        let trace = matches!(
            std::env::var("REDDB_BULK_TIMING").ok().as_deref(),
            Some("1") | Some("true") | Some("on")
        );
        let t_start = std::time::Instant::now();
        let n = entities.len();
        let manager = self.get_or_create_collection(collection);
        let t_get_coll = t_start.elapsed();

        // Assign IDs and per-table row_ids before serialization. Bulk
        // insert must follow the same global ID semantics as
        // insert()/insert_auto(). Reserve ID ranges in single fetch_add
        // calls instead of one-per-entity. The overwhelming common case
        // is "every entity comes in with id==0 and kind==TableRow with
        // row_id==0" (the wire bulk insert path). Any entity that
        // already carries an id or row_id falls back to the individual
        // register_* path to preserve those semantics exactly.
        let t0 = std::time::Instant::now();
        let n_missing_entity_ids = entities.iter().filter(|e| e.id.raw() == 0).count() as u64;
        let n_missing_row_ids = entities
            .iter()
            .filter(|e| matches!(e.kind, EntityKind::TableRow { row_id: 0, .. }))
            .count() as u64;
        let mut entity_id_range = if n_missing_entity_ids > 0 {
            self.reserve_entity_ids(n_missing_entity_ids)
        } else {
            0..0
        };
        let mut row_id_range = if n_missing_row_ids > 0 {
            manager.reserve_row_ids(n_missing_row_ids)
        } else {
            0..0
        };
        for entity in &mut entities {
            if entity.id.raw() == 0 {
                let next = entity_id_range
                    .next()
                    .expect("reserved entity-id range exhausted");
                entity.id = EntityId::new(next);
            } else {
                self.register_entity_id(entity.id);
            }
            if let EntityKind::TableRow { ref mut row_id, .. } = entity.kind {
                if *row_id == 0 {
                    *row_id = row_id_range
                        .next()
                        .expect("reserved row-id range exhausted");
                } else {
                    manager.register_row_id(*row_id);
                }
            }
            entity.ensure_table_logical_id();
        }
        let t_assign_ids = t0.elapsed();

        // Capture graph node labels before entities are moved into the segment manager
        let graph_labels: Vec<Option<(String, EntityId)>> = entities
            .iter()
            .map(|e| {
                if let EntityKind::GraphNode(ref node) = e.kind {
                    Some((node.label.clone(), e.id))
                } else {
                    None
                }
            })
            .collect();

        // Move entities into segment before serializing their canonical image:
        // bulk insertion assigns sequence identities used by checkpoint proofs.
        let t0 = std::time::Instant::now();
        let ids = manager.bulk_insert(entities)?;
        let t_manager = t0.elapsed();
        for id in &ids {
            self.register_entity_id(*id);
        }

        // Serialize for B-tree and WAL from the installed segment rows.
        // `serialize_entity_record` is pure; with `n >= 1024` the
        // rayon dispatch cost is amortised and the 16-core serialize
        // fan-out becomes the biggest single win on insert_bulk. For
        // smaller batches, stay serial — micro-batches pay more in
        // work-stealing overhead than they save.
        let t0 = std::time::Instant::now();
        let serialized: Option<Vec<(Vec<u8>, Vec<u8>)>> =
            if self.pager.is_some() || self.config.embedded_wal_path.is_some() {
                let fv = self.format_version();
                let serial_map = |id: &EntityId| {
                    manager
                        .get_with(*id, |entity| {
                            (
                                id.raw().to_be_bytes().to_vec(),
                                Self::serialize_entity_record(entity, None, fv),
                            )
                        })
                        .expect("bulk-inserted entity must remain in the fenced segment")
                };
                // Gate chosen to match the bench's `BULK_BATCH_SIZE=1000`.
                // Rayon's dispatch overhead is ~30µs/batch — on a 15-col
                // row serializing for ~40µs the break-even is around
                // 200-300 rows on a 16-core host. Keep a safety margin
                // at 512.
                if ids.len() >= 512 {
                    use rayon::prelude::*;
                    Some(ids.par_iter().map(serial_map).collect())
                } else {
                    Some(ids.iter().map(serial_map).collect())
                }
            } else {
                None
            };
        let t_serialize = t0.elapsed();

        // Update graph label index for bulk-inserted GraphNode entities
        for (label, entity_id) in graph_labels.iter().flatten() {
            self.update_graph_label_index(collection, label, *entity_id);
        }

        // REDDB_BULK_SKIP_PERSIST_UNSAFE=1 skips the persistent B-tree index
        // during bulk ingest.
        //
        // UNSAFE: for ephemeral benchmark containers ONLY.
        // This flag is silently ignored when a pager (durable storage) is active.
        // In persistent mode, bulk inserts ALWAYS write to the B-tree so the data
        // survives a cold restart without any manual rebuild step.
        //
        // The flag is only honoured when self.pager is None (in-memory / ephemeral).
        let skip_btree_requested = matches!(
            std::env::var("REDDB_BULK_SKIP_PERSIST_UNSAFE")
                .ok()
                .as_deref(),
            Some("1") | Some("true") | Some("on")
        );
        // Honour the flag only when there is no durable pager.
        // If a pager exists we are in persistent mode → always persist.
        let skip_btree = skip_btree_requested && self.pager.is_none();
        if skip_btree_requested && !skip_btree {
            // Flag was set but we ignored it because we have a real pager.
            static IGNORED: std::sync::OnceLock<()> = std::sync::OnceLock::new();
            IGNORED.get_or_init(|| {
                tracing::warn!(
                    "REDDB_BULK_SKIP_PERSIST_UNSAFE set but durable pager is \
                     active — flag ignored; bulk inserts will be persisted normally"
                );
            });
        } else if skip_btree {
            // Ephemeral mode and flag is active — warn once.
            static WARNED: std::sync::OnceLock<()> = std::sync::OnceLock::new();
            WARNED.get_or_init(|| {
                tracing::warn!(
                    "REDDB_BULK_SKIP_PERSIST_UNSAFE set (ephemeral/no-pager mode) — \
                     bulk inserts NOT durable; data will be lost on restart"
                );
            });
        }

        // Batch B-tree write from pre-serialized data.
        // Uses sorted bulk insert: walks to a leaf once, appends many entries,
        // writes each leaf exactly once per batch — O(N) instead of O(N²).
        let mut t_btree_lock = std::time::Duration::ZERO;
        let mut t_btree_insert = std::time::Duration::ZERO;
        let mut t_flush = std::time::Duration::ZERO;
        if !skip_btree
            && !Self::deferred_store_wal_capture_active()
            && self.config.resident_memory_bytes_max.is_none()
        {
            if let (Some(pager), Some(batch)) = (&self.pager, serialized.as_ref()) {
                let t0 = std::time::Instant::now();
                let mut btree_indices = self.btree_indices.write();
                let btree = btree_indices
                    .entry(collection.to_string())
                    .or_insert_with(|| Arc::new(BTree::new(Arc::clone(pager))));
                let root_before = btree.root_page_id();
                t_btree_lock = t0.elapsed();

                let t0 = std::time::Instant::now();
                let _ = btree.bulk_insert_sorted(batch);
                t_btree_insert = t0.elapsed();
                let registry_dirty = root_before != btree.root_page_id();

                let t0 = std::time::Instant::now();
                if registry_dirty {
                    self.mark_paged_registry_dirty();
                }
                t_flush = t0.elapsed();
            }
        }

        // Fold 25k per-entity `UpsertEntityRecord` actions into one
        // `BulkUpsertEntityRecords` — same on-disk semantics (replay
        // applies each record by id), one `WalRecord::PageWrite`
        // instead of N, and `record.clone()` drops out because we
        // consume `serialized`.
        let actions = serialized
            .map(|batch| {
                let records: Vec<Vec<u8>> =
                    batch.into_iter().map(|(_key, record)| record).collect();
                vec![StoreWalAction::BulkUpsertEntityRecords {
                    collection: collection.to_string(),
                    records,
                }]
            })
            .unwrap_or_default();
        self.finish_paged_write(actions)?;

        if trace {
            tracing::debug!(
                n,
                total = ?t_start.elapsed(),
                get_coll = ?t_get_coll,
                assign = ?t_assign_ids,
                serialize = ?t_serialize,
                manager = ?t_manager,
                btree_lock = ?t_btree_lock,
                btree = ?t_btree_insert,
                flush = ?t_flush,
                "bulk_insert timing"
            );
        }

        Ok(ids)
    }

    /// Insert an entity, creating collection if needed
    pub fn insert_auto(
        &self,
        collection: &str,
        entity: UnifiedEntity,
    ) -> Result<EntityId, StoreError> {
        let collection = Self::canonical_collection(collection);
        // Include the WAL append in the checkpoint publication fence.
        let _publication = self.vault_publication_lock.lock();
        let manager = self.get_or_create_collection(collection);
        let mut entity = entity;
        entity.canonicalize_system_collection_names();
        if entity.id.raw() == 0 {
            entity.id = self.next_entity_id();
        } else {
            self.register_entity_id(entity.id);
        }
        // Assign per-table sequential row_id if not set
        if let EntityKind::TableRow { ref mut row_id, .. } = entity.kind {
            if *row_id == 0 {
                *row_id = manager.next_row_id();
            } else {
                manager.register_row_id(*row_id);
            }
        }
        entity.ensure_table_logical_id();

        // Capture graph node label before entity is moved into the manager
        let graph_node_label: Option<String> = if let EntityKind::GraphNode(ref node) = entity.kind
        {
            Some(node.label.clone())
        } else {
            None
        };
        // Index into context index before consuming the entity
        self.resident
            .context_index
            .index_entity(collection, &entity);

        // Serialize the canonical segment image after insertion: the segment
        // assigns its sequence identity. Borrow it instead of cloning payloads.
        let persist_record = self.pager.is_some() || self.config.embedded_wal_path.is_some();
        if self.config.auto_index_refs {
            self.index_cross_refs(&entity, collection)?;
        }

        // #2376: `persist()` rebuilds every collection B-tree from the
        // manager under `btree_indices.write()`. Hold the read side from the
        // manager insert through the B-tree insert so a rebuild observes
        // both or neither. Otherwise the rebuild re-inserts this row from the
        // manager and one of the two B-tree inserts fails on a duplicate key,
        // or lands mid-rebuild and tears the tree.
        let paged_btree = if persist_record
            && !Self::deferred_store_wal_capture_active()
            && self.config.resident_memory_bytes_max.is_none()
        {
            self.get_or_create_btree(collection)
        } else {
            None
        };
        let persist_fence = paged_btree
            .as_ref()
            .map(|_| self.btree_indices.read_recursive());
        let before = paged_btree.as_ref().map(|_| manager.checkpoint_revisions());
        let unversioned = entity.xmin == 0 && entity.xmax == 0;

        let id = manager.insert(entity)?;
        let serialized_record = if persist_record {
            Some(
                manager
                    .get_with(id, |entity| {
                        Self::serialize_entity_record(entity, None, self.format_version())
                    })
                    .ok_or(StoreError::EntityNotFound(id))?,
            )
        } else {
            None
        };
        // `register_entity_id` already advances the atomic counter on
        // the allocation path above (`self.next_entity_id()`), so the
        // second call here is a no-op CAS loop on the hot path. Only
        // needed for the caller-supplied-id branch which happens via
        // the `register_entity_id` call on line 573.

        // Update graph label index for GraphNode entities
        if let Some(ref label) = graph_node_label {
            self.update_graph_label_index(collection, label, id);
        }

        let mut registry_dirty = false;
        if let (Some(btree), Some(record)) = (paged_btree.as_ref(), serialized_record.as_ref()) {
            let root_before = btree.root_page_id();

            let key = id.raw().to_be_bytes();
            btree.insert(&key, record).map_err(|e| {
                StoreError::Io(std::io::Error::other(format!(
                    "B-tree insert error while inserting '{collection}'/{id}: {e}"
                )))
            })?;
            registry_dirty = root_before != btree.root_page_id();
        }
        drop(persist_fence);

        // Perf: pagerless → skip WAL-action construction (saves a
        // third manager.get + entity serialize per insert). For
        // in-memory runtimes finish_paged_write is a no-op.
        if self.pager.is_some() || self.config.embedded_wal_path.is_some() {
            let actions = serialized_record
                .map(|record| {
                    vec![StoreWalAction::UpsertEntityRecord {
                        collection: collection.to_string(),
                        record,
                    }]
                })
                .unwrap_or_default();
            if registry_dirty {
                self.mark_paged_registry_dirty();
            }
            if let Some(pending) = self.enqueue_grouped_actions(&actions)? {
                drop(_publication);
                self.complete_grouped_actions(&pending, &actions)?;
            } else {
                self.finish_paged_write(actions)?;
            }
        }
        if unversioned && !Self::deferred_store_wal_capture_active() {
            if let Some(before) = before {
                self.note_paged_insert(collection, &manager, &before, id);
            }
        }
        Ok(id)
    }

    /// Get an entity from a collection
    ///
    /// Prefers the live SegmentManager view so reads after update/delete observe
    /// the current in-memory state even when the paged B-tree image has not been
    /// refreshed yet. Falls back to the B-tree image for recovery-oriented reads.
    pub fn get(&self, collection: &str, id: EntityId) -> Option<UnifiedEntity> {
        // Historical best-effort adapter. Authoritative host reads use try_get
        // so a failed cold read cannot become a successful "missing" result.
        self.try_get(collection, id, usize::MAX).ok().flatten()
    }

    pub fn try_get(
        &self,
        collection: &str,
        id: EntityId,
        limit_bytes: usize,
    ) -> Result<Option<UnifiedEntity>, StoreError> {
        let Some(manager) = self.get_collection(collection) else {
            return Ok(None);
        };
        manager.try_get(id, limit_bytes)
    }

    /// Resolve a table row by stable logical identity.
    ///
    /// Legacy rows use `logical_id == id`, so the physical-id lookup remains
    /// the fast path. Versioned rows may have a different physical id and fall
    /// back to a scan until the history-store/index slices add a dedicated map.
    pub fn get_table_row_by_logical_id(
        &self,
        collection: &str,
        logical_id: EntityId,
    ) -> Option<UnifiedEntity> {
        if let Some(entity) = self.get(collection, logical_id) {
            if matches!(entity.kind, EntityKind::TableRow { .. })
                && entity.logical_id() == logical_id
                && entity.xmax == 0
            {
                return Some(entity);
            }
        }

        let manager = self.get_collection(collection)?;
        let mut matches = manager.query_all(|entity| {
            matches!(entity.kind, EntityKind::TableRow { .. }) && entity.logical_id() == logical_id
        });
        matches
            .iter()
            .find(|entity| entity.xmax == 0)
            .cloned()
            .or_else(|| matches.pop())
    }

    pub fn table_row_versions_by_logical_id(
        &self,
        collection: &str,
        logical_id: EntityId,
    ) -> Vec<UnifiedEntity> {
        self.resident
            .table_row_versions_by_logical_id(collection, logical_id)
    }

    pub(crate) fn try_for_each_table_row_version(
        &self,
        collection: &str,
        logical_id: EntityId,
        mut visit: impl FnMut(&UnifiedEntity) -> bool,
    ) -> Result<(), StoreError> {
        let Some(manager) = self.get_collection(collection) else {
            return Ok(());
        };
        manager.try_for_each_entity(usize::MAX, |entity| {
            !matches!(entity.kind, EntityKind::TableRow { .. })
                || entity.logical_id() != logical_id
                || visit(entity)
        })
    }

    pub(crate) fn try_get_table_row_by_logical_id(
        &self,
        collection: &str,
        logical_id: EntityId,
    ) -> Result<Option<UnifiedEntity>, StoreError> {
        let Some(manager) = self.get_collection(collection) else {
            return Ok(None);
        };
        if let Some(entity) = manager.try_get(logical_id, usize::MAX)? {
            if matches!(entity.kind, EntityKind::TableRow { .. })
                && entity.logical_id() == logical_id
                && entity.xmax == 0
            {
                return Ok(Some(entity));
            }
        }
        // Preserve the current-row policy: first live version, otherwise the
        // last matching physical version. Retain only one owned candidate.
        let mut selected = None;
        self.try_for_each_table_row_version(collection, logical_id, |entity| {
            selected = Some(entity.clone());
            entity.xmax != 0
        })?;
        manager.restore_owned_row_schemas(selected.iter_mut());
        Ok(selected)
    }

    pub fn vacuum_mvcc_history(
        &self,
        collection: &str,
        cutoff_xid: u64,
    ) -> Result<MvccVacuumStats, StoreError> {
        let manager = self
            .get_collection(collection)
            .ok_or_else(|| StoreError::CollectionNotFound(collection.to_string()))?;
        let mut logical = HashMap::<EntityId, (bool, u64)>::new();
        manager.for_each_entity(|entity| {
            if !matches!(entity.kind, EntityKind::TableRow { .. }) {
                return true;
            }
            let entry = logical.entry(entity.logical_id()).or_insert((false, 0));
            if entity.xmax == 0 {
                entry.0 = true;
            }
            entry.1 = entry.1.max(entity.xmin);
            true
        });

        let mut stats = MvccVacuumStats::default();
        let mut reclaim_ids = Vec::new();
        manager.for_each_entity(|entity| {
            if !matches!(entity.kind, EntityKind::TableRow { .. }) || entity.xmax == 0 {
                return true;
            }
            stats.scanned_versions += 1;
            let (has_live_version, max_xmin) = logical
                .get(&entity.logical_id())
                .copied()
                .unwrap_or((false, entity.xmin));
            let is_delete_tombstone = !has_live_version && entity.xmin == max_xmin;
            if entity.xmax < cutoff_xid {
                stats.reclaimed_versions += 1;
                if is_delete_tombstone {
                    stats.reclaimed_tombstones += 1;
                } else {
                    stats.reclaimed_history_versions += 1;
                }
                reclaim_ids.push(entity.id);
            } else {
                stats.retained_versions += 1;
                if is_delete_tombstone {
                    stats.retained_tombstones += 1;
                } else {
                    stats.retained_history_versions += 1;
                }
            }
            true
        });

        if !reclaim_ids.is_empty() {
            self.delete_batch(collection, &reclaim_ids)?;
        }
        Ok(stats)
    }

    pub(crate) fn install_versioned_table_row_update(
        &self,
        collection: &str,
        old_version: UnifiedEntity,
        mut new_version: UnifiedEntity,
        metadata: Option<&Metadata>,
    ) -> Result<(), StoreError> {
        let manager = self
            .get_collection(collection)
            .ok_or_else(|| StoreError::CollectionNotFound(collection.to_string()))?;

        let old_id = old_version.id;
        let new_id = new_version.id;
        let inherited_metadata = metadata.cloned().or_else(|| manager.get_metadata(old_id));

        self.resident.entity_cache.remove(old_id.raw());
        self.resident.entity_cache.remove(new_id.raw());
        manager.update(old_version.clone())?;
        self.resident.context_index.remove_entity(old_id);

        self.register_entity_id(new_version.id);
        if let EntityKind::TableRow { ref mut row_id, .. } = new_version.kind {
            if *row_id == 0 {
                *row_id = manager.next_row_id();
            } else {
                manager.register_row_id(*row_id);
            }
        }
        new_version.ensure_table_logical_id();
        manager.insert(new_version.clone())?;
        if let Some(metadata) = inherited_metadata {
            manager.set_metadata(new_id, metadata)?;
        }
        self.resident
            .context_index
            .index_entity(collection, &new_version);
        if self.config.auto_index_refs {
            self.index_cross_refs(&new_version, collection)?;
        }

        let old_metadata = manager.get_metadata(old_id);
        let new_metadata = manager.get_metadata(new_id);
        let fv = self.format_version();
        // The inserted segment assigns the sequence identity. Persist that
        // canonical image, rather than the caller's pre-insertion clone.
        let records = vec![
            manager
                .get_with(old_id, |entity| {
                    Self::serialize_entity_record(entity, old_metadata.as_ref(), fv)
                })
                .ok_or(StoreError::EntityNotFound(old_id))?,
            manager
                .get_with(new_id, |entity| {
                    Self::serialize_entity_record(entity, new_metadata.as_ref(), fv)
                })
                .ok_or(StoreError::EntityNotFound(new_id))?,
        ];
        self.finish_paged_write_for_projection(
            [StoreWalAction::BulkUpsertEntityRecords {
                collection: collection.to_string(),
                records,
            }],
            true,
        )?;

        Ok(())
    }

    /// Batch-fetch multiple entities from the same collection in minimal lock acquisitions.
    ///
    /// Preferred over N individual `get()` calls in indexed-scan loops (sorted index,
    /// bitmap, hash). Reduces lock acquisitions from N×3 to 2-3 total.
    /// Preserves input order: `result[i]` corresponds to `ids[i]`.
    pub fn get_batch(&self, collection: &str, ids: &[EntityId]) -> Vec<Option<UnifiedEntity>> {
        match self.get_collection(collection) {
            Some(manager) => manager.get_many(ids),
            None => vec![None; ids.len()],
        }
    }

    /// Preserve I/O failures when fetching indexed query candidates.
    pub fn try_get_batch(
        &self,
        collection: &str,
        ids: &[EntityId],
        read_bytes_max: usize,
    ) -> Result<Vec<Option<UnifiedEntity>>, StoreError> {
        match self.get_collection(collection) {
            Some(manager) => manager.try_get_many(ids, read_bytes_max),
            None => Ok(vec![None; ids.len()]),
        }
    }

    /// Get an entity from any collection
    pub fn get_any(&self, id: EntityId) -> Option<(String, UnifiedEntity)> {
        self.resident.get_any(id)
    }

    /// Hit rate of the store's entity cache (`get_any` lookups).
    ///
    /// Returns `None` until the cache has served at least one lookup.
    /// Exposed for observability — e.g. dashboards distinguishing graph
    /// workloads (high hit rate) from OLTP DML (≈ 0 % hit rate).
    pub fn entity_cache_hit_rate(&self) -> Option<f64> {
        self.resident.entity_cache_hit_rate()
    }

    /// Snapshot of cache hit / miss / eviction counters and current size.
    pub fn entity_cache_stats(&self) -> super::super::entity_cache::EntityCacheStats {
        self.resident.entity_cache_stats()
    }

    /// Delete an entity
    pub fn delete(&self, collection: &str, id: EntityId) -> Result<bool, StoreError> {
        let collection = Self::canonical_collection(collection);
        let publication = self.vault_publication_lock.lock();
        // Invalidate entity cache (read-lock probe; no write lock when cold).
        self.resident.entity_cache.remove(id.raw());
        let manager = self
            .get_collection(collection)
            .ok_or_else(|| StoreError::CollectionNotFound(collection.to_string()))?;

        let deleted = manager.delete(id)?;
        if !deleted {
            return Ok(false);
        }

        // Remove from B-tree index if active
        let mut registry_dirty = false;
        if self.pager.is_some() && self.config.resident_memory_bytes_max.is_none() {
            let btree_indices = self.btree_indices.read();
            if let Some(btree) = btree_indices.get(collection) {
                let root_before = btree.root_page_id();
                let key = id.raw().to_be_bytes();
                let _ = btree.delete(&key);
                registry_dirty = root_before != btree.root_page_id();
            }
        }

        // Remove cross-references
        self.unindex_cross_refs(id)?;

        // Remove from graph label index
        self.remove_from_graph_label_index(collection, id);

        if registry_dirty {
            self.mark_paged_registry_dirty();
        }
        let actions = [StoreWalAction::DeleteEntityRecord {
            collection: collection.to_string(),
            entity_id: id.raw(),
        }];
        if let Some(pending) = self.enqueue_grouped_actions(&actions)? {
            drop(publication);
            self.complete_grouped_actions(&pending, &actions)?;
        } else {
            self.finish_paged_write(actions)?;
        }

        Ok(true)
    }

    pub fn delete_batch(
        &self,
        collection: &str,
        ids: &[EntityId],
    ) -> Result<Vec<EntityId>, StoreError> {
        let collection = Self::canonical_collection(collection);
        if ids.is_empty() {
            return Ok(Vec::new());
        }
        // Physical reclamation must publish its mutation and WAL record under
        // the same fence as checkpoint snapshots and WAL truncation.
        let _publication = self.vault_publication_lock.lock();

        // Sharded invalidation. The bounded LRU's shard write lock is
        // only acquired when the shard actually carries one of these ids,
        // so the bench-typical zero-hit `delete_sequential` workload never
        // takes a write lock here at all.
        self.resident
            .entity_cache
            .remove_many(ids.iter().map(|id| id.raw()));

        let manager = self
            .get_collection(collection)
            .ok_or_else(|| StoreError::CollectionNotFound(collection.to_string()))?;

        let before = self.pager.as_ref().map(|_| manager.checkpoint_revisions());
        let deleted_ids = manager.delete_batch(ids)?;
        if deleted_ids.is_empty() {
            return Ok(deleted_ids);
        }

        let mut registry_dirty = false;
        let mut updated_tree = false;
        if self.pager.is_some() && self.config.resident_memory_bytes_max.is_none() {
            let btree_indices = self.btree_indices.read();
            if let Some(btree) = btree_indices.get(collection) {
                updated_tree = true;
                let root_before = btree.root_page_id();
                for id in &deleted_ids {
                    let key = id.raw().to_be_bytes();
                    // An aborted, never-published row is already absent. Its
                    // durable physical removal also leaves the tree in sync.
                    updated_tree &= btree.delete(&key).is_ok();
                }
                registry_dirty = root_before != btree.root_page_id();
            }
        }

        self.unindex_cross_refs_batch(&deleted_ids)?;
        self.remove_from_graph_label_index_batch(collection, &deleted_ids);
        if registry_dirty {
            self.mark_paged_registry_dirty();
        }
        let actions = deleted_ids
            .iter()
            .map(|id| StoreWalAction::DeleteEntityRecord {
                collection: collection.to_string(),
                entity_id: id.raw(),
            })
            .collect::<Vec<_>>();
        self.finish_paged_write(actions)?;

        if updated_tree {
            if let Some(before) = before {
                self.note_paged_delete(collection, &manager, &before, &deleted_ids);
            }
            // Paced GC must also pace publication: otherwise grouped WAL
            // workloads can retain detached pages until a distant checkpoint.
            // Publish roots and the deferred-page ledger without re-encoding
            // unchanged row payloads.
            self.flush_paged_registry()?;
        }
        Ok(deleted_ids)
    }

    /// Set metadata for an entity
    pub fn set_metadata(
        &self,
        collection: &str,
        id: EntityId,
        metadata: Metadata,
    ) -> Result<(), StoreError> {
        let manager = self
            .get_collection(collection)
            .ok_or_else(|| StoreError::CollectionNotFound(collection.to_string()))?;

        manager.set_metadata(id, metadata)?;
        if let Some(entity) = manager.get(id) {
            self.persist_entities_to_pager(collection, std::slice::from_ref(&entity))?;
        }
        Ok(())
    }

    /// Get metadata for an entity
    pub fn get_metadata(&self, collection: &str, id: EntityId) -> Option<Metadata> {
        self.get_collection(collection)?.get_metadata(id)
    }

    /// Fallible metadata access for authoritative reads and mutation preimages.
    pub fn try_get_metadata(
        &self,
        collection: &str,
        id: EntityId,
        limit_bytes: usize,
    ) -> Result<Option<Metadata>, StoreError> {
        let Some(manager) = self.get_collection(collection) else {
            return Ok(None);
        };
        manager.try_get_metadata(id, limit_bytes)
    }

    /// Add a cross-reference between entities
    pub fn add_cross_ref(
        &self,
        source_collection: &str,
        source_id: EntityId,
        target_collection: &str,
        target_id: EntityId,
        ref_type: RefType,
        weight: f32,
    ) -> Result<(), StoreError> {
        let source_collection = Self::canonical_collection(source_collection);
        let target_collection = Self::canonical_collection(target_collection);
        let registry_dirty = self.resident.add_cross_ref_in_memory(
            source_collection,
            source_id,
            target_collection,
            target_id,
            ref_type,
            weight,
            self.config.max_cross_refs,
            |entity| {
                self.persist_entities_to_pager(source_collection, std::slice::from_ref(entity))
            },
        )?;
        if registry_dirty
            && matches!(
                self.config.durability_mode,
                crate::api::DurabilityMode::Strict
            )
        {
            self.flush_paged_state()?;
        }
        Ok(())
    }

    /// Get cross-references from an entity
    pub fn get_refs_from(&self, id: EntityId) -> Vec<(EntityId, RefType, String)> {
        self.resident.get_refs_from(id)
    }

    /// Get cross-references to an entity
    pub fn get_refs_to(&self, id: EntityId) -> Vec<(EntityId, RefType, String)> {
        self.resident.get_refs_to(id)
    }

    /// Expand cross-references to get related entities
    pub fn expand_refs(
        &self,
        id: EntityId,
        depth: u32,
        ref_types: Option<&[RefType]>,
    ) -> Vec<(UnifiedEntity, u32, RefType)> {
        let mut results = Vec::new();
        let mut visited = std::collections::HashSet::new();
        visited.insert(id);

        self.expand_refs_recursive(id, depth, ref_types, &mut visited, &mut results, 1);

        results
    }

    fn expand_refs_recursive(
        &self,
        id: EntityId,
        max_depth: u32,
        ref_types: Option<&[RefType]>,
        visited: &mut std::collections::HashSet<EntityId>,
        results: &mut Vec<(UnifiedEntity, u32, RefType)>,
        current_depth: u32,
    ) {
        if current_depth > max_depth {
            return;
        }

        for (target_id, ref_type, target_collection) in self.get_refs_from(id) {
            if visited.contains(&target_id) {
                continue;
            }

            if let Some(types) = ref_types {
                if !types.contains(&ref_type) {
                    continue;
                }
            }

            visited.insert(target_id);

            if let Some(entity) = self.get(&target_collection, target_id) {
                results.push((entity, current_depth, ref_type));

                // Recurse
                self.expand_refs_recursive(
                    target_id,
                    max_depth,
                    ref_types,
                    visited,
                    results,
                    current_depth + 1,
                );
            }
        }
    }

    /// Index cross-references from an entity
    pub(crate) fn index_cross_refs(
        &self,
        entity: &UnifiedEntity,
        collection: &str,
    ) -> Result<(), StoreError> {
        self.resident.index_cross_refs(entity, collection)
    }

    /// Remove cross-references for an entity
    pub(crate) fn unindex_cross_refs(&self, id: EntityId) -> Result<(), StoreError> {
        self.resident.unindex_cross_refs(id)
    }

    pub(crate) fn unindex_cross_refs_batch(&self, ids: &[EntityId]) -> Result<(), StoreError> {
        self.resident.unindex_cross_refs_batch(ids)
    }

    /// Test/observability hook: number of times `unindex_cross_refs_batch`
    /// took the read-only fast path. Used to pin the early-exit in tests
    /// and as a cheap signal in delete-heavy benchmarks.
    pub fn unindex_cross_refs_fast_path_hits(&self) -> u64 {
        self.resident.unindex_cross_refs_fast_path_hits()
    }

    /// Query across all collections with a filter
    pub fn query_all<F>(&self, filter: F) -> Vec<(String, UnifiedEntity)>
    where
        F: Fn(&UnifiedEntity) -> bool + Clone + Send + Sync,
    {
        self.resident.query_all(filter)
    }

    /// Scan every collection under an explicit MVCC snapshot.
    pub fn scan<F>(
        &self,
        snapshot: Option<&crate::runtime::impl_core::SnapshotContext>,
        filter: F,
    ) -> Vec<(String, UnifiedEntity)>
    where
        F: Fn(&UnifiedEntity) -> bool + Clone + Send + Sync,
    {
        self.resident.scan(snapshot, filter)
    }

    /// Filter by metadata across all collections
    pub fn filter_metadata_all(
        &self,
        filters: &[(String, MetadataFilter)],
    ) -> Vec<(String, EntityId)> {
        self.resident.filter_metadata_all(filters)
    }

    /// Get statistics
    pub fn stats(&self) -> StoreStats {
        // Issue #815 — snapshot the manager handles, drop the `collections`
        // read lock, then gather per-collection stats off the map lock. The
        // readiness probe (via `health()`) walks `stats()`, so holding the
        // store lock across a full-store stats scan was part of the same
        // wedge that froze `/catalog` during a large replica re-apply.
        let pairs: Vec<(String, Arc<SegmentManager>)> = {
            let collections = self.resident.collections.read();
            collections
                .iter()
                .map(|(name, mgr)| (name.clone(), Arc::clone(mgr)))
                .collect()
        };

        let mut stats = StoreStats {
            collection_count: pairs.len(),
            ..Default::default()
        };

        for (name, manager) in &pairs {
            let manager_stats = manager.stats();
            stats.total_entities += manager_stats.total_entities;
            stats.total_memory_bytes += manager_stats.total_memory_bytes;
            stats.total_memory_bytes += manager.checkpoint_tracking_bytes();
            stats.collections.insert(name.clone(), manager_stats);
        }

        stats.total_memory_bytes += self.checkpoint_durable_rows.read().memory_bytes();
        stats.total_memory_bytes += self
            .checkpoint_versions
            .read()
            .values()
            .filter_map(|version| version.deferred_rows.as_ref())
            .map(|rows| rows.capacity() * std::mem::size_of::<EntityId>())
            .sum::<usize>();
        stats
    }

    /// Run maintenance on all collections
    pub fn run_maintenance(&self) -> Result<(), StoreError> {
        let collections = self.resident.collections.read();
        for manager in collections.values() {
            manager.run_maintenance()?;
        }
        Ok(())
    }

    /// Bytes pressure consolidation could plausibly return across collections.
    pub fn reclaimable_segment_bytes(&self) -> u64 {
        let collections = self.resident.collections.read();
        collections
            .values()
            .map(|manager| manager.reclaimable_bytes())
            .fold(0, u64::saturating_add)
    }

    /// Advance pressure-triggered consolidation by one paced tick per
    /// collection. Returns true if any collection started or advanced a run.
    pub fn pressure_consolidation_tick(&self) -> bool {
        let managers: Vec<_> = self.resident.collections.read().values().cloned().collect();
        // A plain loop, not `any()`: every collection must receive its tick —
        // short-circuiting on the first advanced run would starve the rest.
        let mut advanced = false;
        for manager in &managers {
            advanced |= manager.pressure_consolidation_tick();
        }
        advanced
    }
}

/// Flatten a JSON value into dot-notation key-value pairs for red_config.
fn flatten_config_json(
    prefix: &str,
    value: &crate::serde_json::Value,
    out: &mut Vec<(String, reddb_types::Value)>,
) {
    use reddb_types::Value;
    match value {
        crate::serde_json::Value::Object(map) => {
            for (k, v) in map {
                let key = if prefix.is_empty() {
                    k.clone()
                } else {
                    format!("{prefix}.{k}")
                };
                flatten_config_json(&key, v, out);
            }
        }
        crate::serde_json::Value::String(s) => {
            out.push((prefix.to_string(), Value::text(s.clone())));
        }
        crate::serde_json::Value::Integer(n) => {
            out.push((prefix.to_string(), Value::Integer(*n)));
        }
        crate::serde_json::Value::Number(n) => {
            if n.fract().abs() < f64::EPSILON {
                out.push((prefix.to_string(), Value::UnsignedInteger(*n as u64)));
            } else {
                out.push((prefix.to_string(), Value::Float(*n)));
            }
        }
        crate::serde_json::Value::Decimal(n) => {
            out.push((prefix.to_string(), Value::DecimalText(n.clone())));
        }
        crate::serde_json::Value::Bool(b) => {
            out.push((prefix.to_string(), Value::Boolean(*b)));
        }
        crate::serde_json::Value::Null => {
            out.push((prefix.to_string(), Value::Null));
        }
        crate::serde_json::Value::Array(_) => {
            let json_str = crate::serde_json::to_string(value).unwrap_or_default();
            out.push((prefix.to_string(), Value::text(json_str)));
        }
    }
}