From 9100687b19c6b2184c444a86fe6ad6ef4282e1ed Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 01:50:12 +0200
Subject: [PATCH 01/16] fix: trim mutation receipts once per admission interval

---
 src/internal/store.rs | 48 +++++++++++++++++++++++++++----------------
 1 file changed, 30 insertions(+), 18 deletions(-)

diff --git a/src/internal/store.rs b/src/internal/store.rs
index 4e0021c..3f37a54 100644
--- a/src/internal/store.rs
+++ b/src/internal/store.rs
@@ -219,6 +219,8 @@ const SEARCH_STAGE_MAGIC: [u8; 2] = *b"SS";
 const SEARCH_ORDER_KEY: &[u8] = b"Q";
 const SEARCH_ORDER_FORMAT: u16 = 1;
 const RECEIPT_RETENTION: usize = 4_096;
+/// Admissions between receipt trims; each pass removes every receipt past retention.
+const RECEIPT_TRIM_INTERVAL: u64 = 256;
 const RECEIPT_EXPIRED_KEY: &[u8] = b"\0receipt-expired";
 const BATCH_RECEIPT_PREFIX: [u8; 2] = [0, b'B'];
 const BATCH_REVERSE_PREFIX: [u8; 2] = [0, b'R'];
@@ -11426,7 +11428,13 @@ impl GraphStore {
         }
         #[cfg(test)]
         self.receipt_writes.fetch_add(1, Ordering::Relaxed);
-        self.trim_receipts(batch)?;
+        // A trim scans the retained receipts, so it runs once per interval, not per write.
+        if stored
+            .admission_sequence
+            .is_multiple_of(RECEIPT_TRIM_INTERVAL)
+        {
+            self.trim_receipts(batch)?;
+        }
         batch
             .inner
             .insert(&self.receipts, stored.id.0, postcard::to_allocvec(&stored)?);
@@ -11495,22 +11503,18 @@ impl GraphStore {
         })
     }
 
+    /// Makes room for one more batch receipt by removing the oldest past retention.
     fn trim_batch_receipts(&self, batch: &mut fjall::OwnedWriteBatch) -> Result<()> {
-        let mut oldest = None;
         let mut count = 0usize;
         for guard in self.receipt_order.prefix(BATCH_ORDER_PREFIX) {
-            let (key, value) = guard.into_inner()?;
-            if oldest.is_none() {
-                oldest = Some((key.to_vec(), value.to_vec()));
-            }
+            guard.into_inner()?;
             count += 1;
-            if count == RECEIPT_RETENTION {
-                break;
-            }
         }
-        if count == RECEIPT_RETENTION
-            && let Some((order, mapping_key)) = oldest
-        {
+        let excess = (count + 1)
+            .saturating_sub(RECEIPT_RETENTION)
+            .min(QV_BUILD_ROWS);
+        for guard in self.receipt_order.prefix(BATCH_ORDER_PREFIX).take(excess) {
+            let (order, mapping_key) = guard.into_inner()?;
             if let Some(value) = self.receipts.get(&mapping_key)? {
                 let mapping: StoredBatchReceipt = postcard::from_bytes(value.as_ref())?;
                 batch.remove(&self.receipts, batch_reverse_key(&mapping.id));
@@ -11550,7 +11554,9 @@ impl GraphStore {
         {
             return Err(StoreError::ReceiptConflict);
         }
-        self.trim_batch_receipts(&mut batch.inner)?;
+        if mapping.sequence.is_multiple_of(RECEIPT_TRIM_INTERVAL) {
+            self.trim_batch_receipts(&mut batch.inner)?;
+        }
         batch
             .inner
             .insert(&self.receipts, key, postcard::to_allocvec(&mapping)?);
@@ -12162,10 +12168,16 @@ mod tests {
             store.mutation_receipt(&receipt.id).unwrap().unwrap()
         };
         let first = stage(0);
-        let second = stage(1);
-        for nanos in 2..=RECEIPT_RETENTION {
-            stage(nanos);
+        // The trim pass at the last admission keeps exactly the newest retained receipts.
+        let total = RECEIPT_RETENTION + RECEIPT_TRIM_INTERVAL as usize;
+        let mut kept = None;
+        for nanos in 1..total {
+            let receipt = stage(nanos);
+            if nanos == total - RECEIPT_RETENTION {
+                kept = Some(receipt);
+            }
         }
+        let kept = kept.unwrap();
         let lookup = |receipt: &MutationReceipt| MutationLookup {
             graph: graph.clone(),
             id: receipt.id,
@@ -12177,11 +12189,11 @@ mod tests {
             MutationStatus::Expired
         ));
         assert!(matches!(
-            store.receipt_status(&lookup(&second)).unwrap(),
+            store.receipt_status(&lookup(&kept)).unwrap(),
             MutationStatus::Known(_)
         ));
 
-        let order = receipt_order_key(&second);
+        let order = receipt_order_key(&kept);
         store.receipt_order.remove(order).unwrap();
         drop(store);
         let reopened = GraphStore::open(dir.path()).unwrap();

From 297f9e089fcf9e90299d241f46c60fc19bb406d9 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 08:14:01 +0200
Subject: [PATCH 02/16] fix: publish search generations without rebuilding
 every scope

---
 src/internal/search/bench.rs |  2 +-
 src/internal/search/mod.rs   | 52 +++++++++++++++++-------------------
 2 files changed, 26 insertions(+), 28 deletions(-)

diff --git a/src/internal/search/bench.rs b/src/internal/search/bench.rs
index dff6cba..92e0751 100644
--- a/src/internal/search/bench.rs
+++ b/src/internal/search/bench.rs
@@ -487,7 +487,7 @@ fn generation_publication() {
                 covered: 0,
             })
             .collect();
-        index.publish_generations(GenerationView::from_rows(index.index_id, rows));
+        index.publish_generations(GenerationView::from_rows(rows));
         let switches = 200;
         let mut lock_ns = Vec::with_capacity(switches);
         let started = Instant::now();
diff --git a/src/internal/search/mod.rs b/src/internal/search/mod.rs
index bd52836..9b8df46 100644
--- a/src/internal/search/mod.rs
+++ b/src/internal/search/mod.rs
@@ -215,31 +215,38 @@ struct RepairIntents {
     retry_at_ms: u64,
 }
 
+/// Shared keys keep the per-publish copy of this map free of string allocations.
 #[derive(Clone, Default)]
 struct GenerationView {
-    by_graph: HashMap<String, GenerationId>,
-    active: HashSet<Vec<u8>>,
+    by_graph: HashMap<Arc<str>, GenerationId>,
 }
 
 impl GenerationView {
+    /// Whether `scope` names the active generation of its graph in this index.
+    fn is_active(&self, index_id: [u8; 16], scope: &[u8]) -> bool {
+        let Some(graph) = scope_graph(scope) else {
+            return false;
+        };
+        scope.starts_with(&index_id)
+            && self
+                .by_graph
+                .get(graph)
+                .is_some_and(|generation| scope.ends_with(&generation.0.to_be_bytes()))
+    }
+
     #[cfg(test)]
-    fn from_rows(index_id: [u8; 16], rows: Vec<GraphGeneration>) -> Self {
+    fn from_rows(rows: Vec<GraphGeneration>) -> Self {
         let mut view = Self::default();
-        view.extend(index_id, rows);
+        view.extend(rows);
         view
     }
 
     #[cfg(test)]
-    fn extend(&mut self, index_id: [u8; 16], rows: Vec<GraphGeneration>) {
+    fn extend(&mut self, rows: Vec<GraphGeneration>) {
         self.by_graph.extend(rows.into_iter().filter_map(|row| {
             row.active
-                .map(|generation| (row.graph.as_str().to_string(), generation))
+                .map(|generation| (Arc::from(row.graph.as_str()), generation))
         }));
-        self.active = self
-            .by_graph
-            .iter()
-            .map(|(graph, generation)| generation_scope(index_id, graph, *generation))
-            .collect();
     }
 }
 
@@ -1065,7 +1072,7 @@ impl SearchIndex {
 
     #[cfg(test)]
     fn publish_rows(&self, rows: Vec<GraphGeneration>) {
-        self.publish_generations(GenerationView::from_rows(self.index_id, rows));
+        self.publish_generations(GenerationView::from_rows(rows));
     }
 
     fn publish_generations(&self, generations: GenerationView) {
@@ -1083,17 +1090,12 @@ impl SearchIndex {
         let mut generations = (*published.generations).clone();
         match generation {
             Some(generation) => {
-                generations.by_graph.insert(graph.to_string(), generation);
+                generations.by_graph.insert(Arc::from(graph), generation);
             }
             None => {
                 generations.by_graph.remove(graph);
             }
         }
-        generations.active = generations
-            .by_graph
-            .iter()
-            .map(|(graph, generation)| generation_scope(self.index_id, graph, *generation))
-            .collect();
         *published = Arc::new(SearchView {
             searcher: self.reader.searcher(),
             generations: Arc::new(generations),
@@ -1222,7 +1224,7 @@ impl SearchIndex {
                     }
                     count = count.saturating_add(1);
                     if row.live {
-                        generations.by_graph.insert(graph.to_string(), generation);
+                        generations.by_graph.insert(Arc::from(graph), generation);
                     }
                 }
             }
@@ -1238,11 +1240,6 @@ impl SearchIndex {
                 ));
             }
         }
-        generations.active = generations
-            .by_graph
-            .iter()
-            .map(|(graph, generation)| generation_scope(self.index_id, graph, *generation))
-            .collect();
         let digest = snapshot.digest(self.index_id)?;
         Ok(ManifestState {
             generations,
@@ -2235,9 +2232,10 @@ impl SearchIndex {
     #[cfg(test)]
     fn collect_top_docs(&self, req: TopRequest<'_>) -> Result<Vec<SearchHit>> {
         let generations = req.view.generations.clone();
+        let index_id = self.index_id;
         let collector = BytesFilterCollector::new(
             GENERATION_SCOPE_FIELD.to_string(),
-            move |scope: &[u8]| generations.active.contains(scope),
+            move |scope: &[u8]| generations.is_active(index_id, scope),
             TopDocs::with_limit(req.limit).order_by_score(),
         );
         let top_docs = req.view.searcher.search(req.query, &collector)?;
@@ -3367,7 +3365,7 @@ impl SearchIndex {
                 detail: "generation scope",
             }));
         };
-        if !req.view.generations.active.contains(scope.as_slice()) || !(req.allows)(graph)? {
+        if !req.view.generations.is_active(self.index_id, &scope) || !(req.allows)(graph)? {
             return Ok(None);
         }
         let stable = column_bytes(req.stable, req.doc).and_then(|stable| {
@@ -4994,7 +4992,7 @@ mod tests {
         assert_eq!(1, repaired.len());
         assert_eq!(graph.as_str(), repaired[0].graph_id);
         let scope = generation_scope(node.search.index_id, graph.as_str(), generation);
-        assert!(pinned.generations.active.contains(scope.as_slice()));
+        assert!(pinned.generations.is_active(node.search.index_id, &scope));
     }
 
     #[test]

From 97e29d9b4d5dd729b7b4fd82b27a220e84129515 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 09:32:18 +0200
Subject: [PATCH 03/16] fix: expand RO-Crate documents with only the context
 terms they use

---
 src/internal/rocrate.rs | 185 +++++++++++++++++++++++++++++++++++++++-
 1 file changed, 182 insertions(+), 3 deletions(-)

diff --git a/src/internal/rocrate.rs b/src/internal/rocrate.rs
index 56a322c..a97ba77 100644
--- a/src/internal/rocrate.rs
+++ b/src/internal/rocrate.rs
@@ -4,7 +4,7 @@
 
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
 use std::io;
-use std::sync::Arc;
+use std::sync::{Arc, LazyLock};
 use std::time::{Duration, Instant};
 
 use crate::RoCrateVersion;
@@ -2878,11 +2878,18 @@ fn jsonld_quads_bounded(
         collect_context_terms(context, &mut terms, false);
     }
     label_blank_nodes(&mut prepared, "", true, &terms);
-    let jsonld = serde_json::to_vec(&prepared)?;
+    inline_used_context(&mut prepared);
+    parse_jsonld(&serde_json::to_vec(&prepared)?, limits)
+}
+
+fn parse_jsonld(
+    jsonld: &[u8],
+    limits: Option<&RoCrateImportLimits>,
+) -> Result<Vec<Quad>, RoCrateError> {
     let parser = JsonLdParser::new()
         .with_base_iri(JSONLD_BASE_IRI)
         .map_err(|error| RoCrateError::JsonLd(error.to_string()))?
-        .for_slice(&jsonld)
+        .for_slice(jsonld)
         .with_load_document_callback(|url, _| load_context(url));
 
     let mut quads = Vec::new();
@@ -2905,6 +2912,120 @@ fn jsonld_quads_bounded(
     Ok(quads)
 }
 
+/// Term maps of the embedded contexts, parsed once per process.
+static CONTEXT_TERMS: LazyLock<HashMap<&'static str, serde_json::Map<String, serde_json::Value>>> =
+    LazyLock::new(|| {
+        [
+            (ROCRATE_CONTEXT_11, RoCrateVersion::V1_1.context_bytes()),
+            (ROCRATE_CONTEXT_12, RoCrateVersion::V1_2.context_bytes()),
+            (ROCRATE_CONTEXT_13, RoCrateVersion::V1_3.context_bytes()),
+            (WORKFLOW_CONTEXT_URL, WORKFLOW_RUN_CONTEXT),
+        ]
+        .into_iter()
+        .filter_map(|(url, bytes)| {
+            let mut document = serde_json::from_slice::<serde_json::Value>(bytes).ok()?;
+            match document.get_mut("@context")?.take() {
+                serde_json::Value::Object(terms) => Some((url, terms)),
+                _ => None,
+            }
+        })
+        .collect()
+    });
+
+/// Replaces top-level embedded contexts with the definitions this document can use.
+/// Unused definitions cannot change expansion, and each full context has thousands.
+fn inline_used_context(document: &mut serde_json::Value) {
+    let Some(object) = document.as_object_mut() else {
+        return;
+    };
+    let mut used = HashSet::new();
+    for (key, value) in object.iter() {
+        if key == "@context" {
+            collect_context_used(value, &mut used);
+        } else {
+            used_term(key, &mut used);
+            collect_used(value, &mut used);
+        }
+    }
+    let Some(context) = object.get_mut("@context") else {
+        return;
+    };
+    match context {
+        serde_json::Value::String(url) => {
+            if let Some(terms) = used_context(url, &used) {
+                *context = terms;
+            }
+        }
+        serde_json::Value::Array(items) => {
+            for item in items {
+                if let serde_json::Value::String(url) = item
+                    && let Some(terms) = used_context(url, &used)
+                {
+                    *item = terms;
+                }
+            }
+        }
+        _ => {}
+    }
+}
+
+fn used_context(url: &str, used: &HashSet<String>) -> Option<serde_json::Value> {
+    let terms = CONTEXT_TERMS.get(url)?;
+    let mut kept = serde_json::Map::new();
+    let mut pending = terms
+        .keys()
+        .filter(|term| term.starts_with('@') || used.contains(term.as_str()))
+        .cloned()
+        .collect::<Vec<_>>();
+    // A kept definition may name its IRI through another term used as a prefix.
+    while let Some(term) = pending.pop() {
+        if kept.contains_key(&term) {
+            continue;
+        }
+        let Some(definition) = terms.get(&term) else {
+            continue;
+        };
+        if let Some((prefix, _)) = definition.as_str().and_then(|iri| iri.split_once(':')) {
+            pending.push(prefix.to_string());
+        }
+        kept.insert(term, definition.clone());
+    }
+    Some(serde_json::Value::Object(kept))
+}
+
+fn used_term(value: &str, used: &mut HashSet<String>) {
+    if let Some((prefix, _)) = value.split_once(':') {
+        used.insert(prefix.to_string());
+    }
+    used.insert(value.to_string());
+}
+
+fn collect_used(value: &serde_json::Value, used: &mut HashSet<String>) {
+    match value {
+        serde_json::Value::String(text) => used_term(text, used),
+        serde_json::Value::Array(items) => items.iter().for_each(|item| collect_used(item, used)),
+        serde_json::Value::Object(object) => {
+            for (key, value) in object {
+                used_term(key, used);
+                collect_used(value, used);
+            }
+        }
+        _ => {}
+    }
+}
+
+/// Inline context objects may define terms through the embedded context's terms.
+fn collect_context_used(context: &serde_json::Value, used: &mut HashSet<String>) {
+    match context {
+        serde_json::Value::Array(items) => items
+            .iter()
+            .filter(|item| item.is_object())
+            .for_each(|item| collect_used(item, used)),
+        serde_json::Value::Object(_) => collect_used(context, used),
+        _ => {}
+    }
+}
+
 fn collect_import_usage(quad: &Quad, blank_nodes: &mut HashSet<String>, literal_bytes: &mut usize) {
     if let SubjectNode::BlankNode(node) = &quad.subject {
         blank_nodes.insert(node.as_str().to_owned());
@@ -4087,6 +4208,64 @@ mod tests {
     use super::*;
     use std::sync::atomic::{AtomicBool, Ordering};
 
+    fn labeled(mut value: serde_json::Value) -> serde_json::Value {
+        let mut terms = HashMap::new();
+        if let Some(context) = value.get("@context") {
+            collect_context_terms(context, &mut terms, false);
+        }
+        label_blank_nodes(&mut value, "", true, &terms);
+        value
+    }
+
+    /// Inlining only the used context terms must not change the expanded triples.
+    #[test]
+    fn inlined_context_equivalent() {
+        let fixtures = [
+            "valid-1.1.json",
+            "valid-1.2.json",
+            "valid-1.3.json",
+            "array-1.3.json",
+            "bioschemas-1.3.json",
+            "custom-1.3.json",
+        ]
+        .map(|name| {
+            let path = format!(
+                "{}/tests/fixtures/rocrate/{name}",
+                env!("CARGO_MANIFEST_DIR")
+            );
+            serde_json::from_str::<serde_json::Value>(&std::fs::read_to_string(path).unwrap())
+                .unwrap()
+        });
+        let synthetic = serde_json::json!({
+            "@context": [ROCRATE_CONTEXT_11, {"extra": "rdf:value"}],
+            "@graph": [
+                {"@id": "ro-crate-metadata.json", "@type": "CreativeWork", "about": {"@id": "./"}},
+                {"@id": "./", "@type": ["Dataset", "schema:Thing"], "name": "Root",
+                 "hasPart": [{"@id": "data.csv"}], "extra": "kept", "encodingFormat": "HTML",
+                 "author": {"@type": "Person", "name": "Nested"},
+                 "@reverse": {"isPartOf": {"@id": "#part"}}},
+                {"@id": "data.csv", "@type": "File", "contentSize": "12"}
+            ]
+        });
+        let mut changed = 0;
+        for document in fixtures.into_iter().chain([synthetic]) {
+            let document = labeled(document);
+            let mut full = parse_jsonld(&serde_json::to_vec(&document).unwrap(), None).unwrap();
+            let mut inlined = document.clone();
+            inline_used_context(&mut inlined);
+            changed += usize::from(inlined["@context"] != document["@context"]);
+            let mut pruned = parse_jsonld(&serde_json::to_vec(&inlined).unwrap(), None).unwrap();
+            full.sort_by_key(ToString::to_string);
+            pruned.sort_by_key(ToString::to_string);
+            assert!(!full.is_empty());
+            assert_eq!(full, pruned, "{document}");
+        }
+        assert!(
+            changed >= 5,
+            "only {changed} documents used an embedded context"
+        );
+    }
+
     use crate::sync::{
         CraqleGraphEvent, CraqleGraphSync, CraqleSyncError, SyncResult, TopicCatchup,
     };

From d554994558d361ee2362a3c39361348dbfc1b4cf Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 09:32:19 +0200
Subject: [PATCH 04/16] fix: keep receipt trims and search coverage checks off
 old tombstones

---
 src/internal/store.rs | 342 ++++++++++++++++++++++++++++++------------
 1 file changed, 250 insertions(+), 92 deletions(-)

diff --git a/src/internal/store.rs b/src/internal/store.rs
index 3f37a54..1ba8c39 100644
--- a/src/internal/store.rs
+++ b/src/internal/store.rs
@@ -6,7 +6,7 @@ use std::cell::{Cell, OnceCell, RefCell};
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
 use std::ops::Bound::{self, Excluded, Included, Unbounded};
 use std::path::Path;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::{
     Arc, Mutex, MutexGuard, PoisonError, RwLock, RwLockReadGuard as ReadGuard,
     RwLockWriteGuard as WriteGuard, TryLockError,
@@ -219,8 +219,6 @@ const SEARCH_STAGE_MAGIC: [u8; 2] = *b"SS";
 const SEARCH_ORDER_KEY: &[u8] = b"Q";
 const SEARCH_ORDER_FORMAT: u16 = 1;
 const RECEIPT_RETENTION: usize = 4_096;
-/// Admissions between receipt trims; each pass removes every receipt past retention.
-const RECEIPT_TRIM_INTERVAL: u64 = 256;
 const RECEIPT_EXPIRED_KEY: &[u8] = b"\0receipt-expired";
 const BATCH_RECEIPT_PREFIX: [u8; 2] = [0, b'B'];
 const BATCH_REVERSE_PREFIX: [u8; 2] = [0, b'R'];
@@ -1010,6 +1008,62 @@ struct DurableCommit {
     batch: fjall::OwnedWriteBatch,
     pending_fts: PendingFts,
     pending_receipts: Vec<MutationReceipt>,
+    receipt_trim: ReceiptTrim,
+}
+
+/// Receipt count changes and trim positions a batch applies once it commits.
+#[derive(Default)]
+struct ReceiptTrim {
+    receipts: TrimChange,
+    batches: TrimChange,
+}
+
+#[derive(Default)]
+struct TrimChange {
+    added: usize,
+    removed: usize,
+    floor: Option<Vec<u8>>,
+}
+
+/// Committed entries of one retained kind; races only lower the count, so a
+/// trim never evicts a retained entry. The floor skips earlier trims' tombstones.
+#[derive(Default)]
+struct TrimCount {
+    live: AtomicUsize,
+    floor: Mutex<Option<Vec<u8>>>,
+}
+
+impl TrimCount {
+    fn excess(&self, change: &TrimChange) -> usize {
+        (self.live.load(Ordering::SeqCst) + change.added + 1)
+            .saturating_sub(change.removed)
+            .saturating_sub(RECEIPT_RETENTION)
+            .min(QV_BUILD_ROWS)
+    }
+
+    fn start(&self, change: &TrimChange) -> Option<Vec<u8>> {
+        change.floor.clone().or_else(|| {
+            self.floor
+                .lock()
+                .unwrap_or_else(PoisonError::into_inner)
+                .clone()
+        })
+    }
+
+    fn settle(&self, change: TrimChange) {
+        self.live.fetch_add(change.added, Ordering::SeqCst);
+        let _ = self
+            .live
+            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |live| {
+                Some(live.saturating_sub(change.removed))
+            });
+        if let Some(floor) = change.floor {
+            let mut current = self.floor.lock().unwrap_or_else(PoisonError::into_inner);
+            if current.as_ref().is_none_or(|current| *current < floor) {
+                *current = Some(floor);
+            }
+        }
+    }
 }
 
 pub struct WriteBatch {
@@ -1021,6 +1075,7 @@ pub struct WriteBatch {
     /// Queue keys staged and tokenized under the queue lock at commit.
     pending_fts: PendingFts,
     pending_receipts: Vec<MutationReceipt>,
+    receipt_trim: ReceiptTrim,
 }
 
 impl WriteBatch {
@@ -1032,6 +1087,7 @@ impl WriteBatch {
             publish: PendingPublish::default(),
             pending_fts: PendingFts::default(),
             pending_receipts: Vec::new(),
+            receipt_trim: ReceiptTrim::default(),
         }
     }
 
@@ -1278,6 +1334,8 @@ pub struct GraphStore {
     write_locks: Vec<Mutex<()>>,
     receipt_locks: Vec<Mutex<()>>,
     receipt_next: AtomicU64,
+    receipt_count: TrimCount,
+    batch_receipt_count: TrimCount,
     #[cfg(test)]
     receipt_lookups: AtomicU64,
     #[cfg(test)]
@@ -1372,6 +1430,8 @@ pub struct GraphStore {
     /// Serializes FTS queue mutations and is innermost in the store lock order.
     fts_queue_lock: Mutex<()>,
     dirty_counter: AtomicU64,
+    /// No queued search work has a lower token; skips acknowledged queue tombstones.
+    queue_floor: AtomicU64,
     dirty_committed: AtomicU64,
     /// Number of graph diagnostics recomputations by this store instance.
     diagnostics_computed: AtomicU64,
@@ -7510,6 +7570,8 @@ impl GraphStore {
             write_locks: (0..GRAPH_LOCK_SHARDS).map(|_| Mutex::new(())).collect(),
             receipt_locks: (0..COMMIT_LOCK_SHARDS).map(|_| Mutex::new(())).collect(),
             receipt_next: AtomicU64::new(1),
+            receipt_count: TrimCount::default(),
+            batch_receipt_count: TrimCount::default(),
             #[cfg(test)]
             receipt_lookups: AtomicU64::new(0),
             #[cfg(test)]
@@ -7588,6 +7650,7 @@ impl GraphStore {
             delete_stalled: std::sync::atomic::AtomicBool::new(false),
             fts_queue_lock: Mutex::new(()),
             dirty_counter: AtomicU64::new(1),
+            queue_floor: AtomicU64::new(0),
             dirty_committed: AtomicU64::new(0),
             diagnostics_computed: AtomicU64::new(0),
             #[cfg(test)]
@@ -7759,11 +7822,13 @@ impl GraphStore {
             .unwrap_or(0);
         let mut batch = self.buffered_batch();
         let mut staged = 0usize;
+        let mut live = 0usize;
         for guard in self.receipts.iter() {
             let (key, value) = guard.into_inner()?;
             if key.len() != 32 {
                 continue;
             }
+            live += 1;
             let receipt: MutationReceipt = postcard::from_bytes(value.as_ref())?;
             highest = highest.max(receipt.admission_sequence);
             let order = receipt_order_key(&receipt);
@@ -7781,6 +7846,15 @@ impl GraphStore {
         }
         self.receipt_next
             .store(highest.saturating_add(1).max(1), Ordering::SeqCst);
+        self.receipt_count.live.store(live, Ordering::SeqCst);
+        let mut batches = 0usize;
+        for guard in self.receipt_order.prefix(BATCH_ORDER_PREFIX) {
+            guard.into_inner()?;
+            batches += 1;
+        }
+        self.batch_receipt_count
+            .live
+            .store(batches, Ordering::SeqCst);
         Ok(())
     }
 
@@ -11062,13 +11136,12 @@ impl GraphStore {
         if manifest.epoch != coverage.manifest_epoch {
             return Err(StoreError::InvalidSearchState("search-manifest-changed"));
         }
-        if let Some(guard) = self.search_queue.prefix([SEARCH_ORDER_PREFIX]).next() {
-            let (key, _) = guard.into_inner()?;
-            if decode_search_order(key.as_ref())?.token <= coverage.covered {
-                return Err(StoreError::InvalidSearchState(
-                    "search-coverage-debt-remains",
-                ));
-            }
+        if let Some(head) = self.queue_head()?
+            && head.token <= coverage.covered
+        {
+            return Err(StoreError::InvalidSearchState(
+                "search-coverage-debt-remains",
+            ));
         }
         let settled = SearchCoverage {
             rebuild: None,
@@ -11110,13 +11183,12 @@ impl GraphStore {
         {
             return Err(StoreError::InvalidSearchState("search-manifest-changed"));
         }
-        if let Some(guard) = self.search_queue.prefix([SEARCH_ORDER_PREFIX]).next() {
-            let (key, _) = guard.into_inner()?;
-            if decode_search_order(key.as_ref())?.token <= coverage.covered {
-                return Err(StoreError::InvalidSearchState(
-                    "search-coverage-debt-remains",
-                ));
-            }
+        if let Some(head) = self.queue_head()?
+            && head.token <= coverage.covered
+        {
+            return Err(StoreError::InvalidSearchState(
+                "search-coverage-debt-remains",
+            ));
         }
         if let Some(current) = self.search_coverage()?
             && (current.rebuild.is_some()
@@ -11358,12 +11430,23 @@ impl GraphStore {
         Ok(())
     }
 
-    /// Makes room for one more receipt, sparing receipts staged in this batch.
+    /// Makes room for one more receipt by removing the oldest past retention, sparing
+    /// receipts staged in this batch. It scans forward from the last committed trim.
     fn trim_receipts(&self, batch: &mut WriteBatch) -> Result<()> {
+        let excess = self.receipt_count.excess(&batch.receipt_trim.receipts);
+        if excess == 0 {
+            return Ok(());
+        }
+        let start = self
+            .receipt_count
+            .start(&batch.receipt_trim.receipts)
+            .map_or(Bound::Unbounded, Bound::Excluded);
         let mut expired = None;
-        let mut kept = 0usize;
         let mut removed = 0usize;
-        for guard in self.receipt_order.iter().rev() {
+        for guard in self
+            .receipt_order
+            .range::<Vec<u8>, _>((start, Bound::Unbounded))
+        {
             let (key, id) = guard.into_inner()?;
             if key.len() != 40 {
                 continue;
@@ -11372,21 +11455,22 @@ impl GraphStore {
                 .pending_receipts
                 .iter()
                 .any(|receipt| receipt.id.0[..] == id[..]);
-            if kept + 1 < RECEIPT_RETENTION || staged {
-                kept += 1;
+            if staged {
                 continue;
             }
-            if removed == QV_BUILD_ROWS {
-                break;
-            }
             if let Some(value) = self.receipts.get(&id)? {
                 let receipt: MutationReceipt = postcard::from_bytes(value.as_ref())?;
                 expired = expired.max(Some(receipt.admission_sequence));
             }
-            batch.inner.remove(&self.receipt_order, key);
+            batch.inner.remove(&self.receipt_order, key.clone());
             batch.inner.remove(&self.receipts, id);
+            batch.receipt_trim.receipts.floor = Some(key.to_vec());
             removed += 1;
+            if removed == excess {
+                break;
+            }
         }
+        batch.receipt_trim.receipts.removed += removed;
         if let Some(expired) = expired {
             let stored = self
                 .receipt_order
@@ -11428,13 +11512,8 @@ impl GraphStore {
         }
         #[cfg(test)]
         self.receipt_writes.fetch_add(1, Ordering::Relaxed);
-        // A trim scans the retained receipts, so it runs once per interval, not per write.
-        if stored
-            .admission_sequence
-            .is_multiple_of(RECEIPT_TRIM_INTERVAL)
-        {
-            self.trim_receipts(batch)?;
-        }
+        self.trim_receipts(batch)?;
+        batch.receipt_trim.receipts.added += 1;
         batch
             .inner
             .insert(&self.receipts, stored.id.0, postcard::to_allocvec(&stored)?);
@@ -11504,24 +11583,40 @@ impl GraphStore {
     }
 
     /// Makes room for one more batch receipt by removing the oldest past retention.
-    fn trim_batch_receipts(&self, batch: &mut fjall::OwnedWriteBatch) -> Result<()> {
-        let mut count = 0usize;
-        for guard in self.receipt_order.prefix(BATCH_ORDER_PREFIX) {
-            guard.into_inner()?;
-            count += 1;
+    fn trim_batch_receipts(&self, batch: &mut WriteBatch) -> Result<()> {
+        let change = &batch.receipt_trim.batches;
+        let excess = self.batch_receipt_count.excess(change);
+        if excess == 0 {
+            return Ok(());
         }
-        let excess = (count + 1)
-            .saturating_sub(RECEIPT_RETENTION)
-            .min(QV_BUILD_ROWS);
-        for guard in self.receipt_order.prefix(BATCH_ORDER_PREFIX).take(excess) {
+        let start = self.batch_receipt_count.start(change).map_or(
+            Bound::Included(BATCH_ORDER_PREFIX.to_vec()),
+            Bound::Excluded,
+        );
+        let mut removed = 0usize;
+        for guard in self
+            .receipt_order
+            .range::<Vec<u8>, _>((start, Bound::Unbounded))
+        {
             let (order, mapping_key) = guard.into_inner()?;
+            if !order.starts_with(&BATCH_ORDER_PREFIX) {
+                break;
+            }
             if let Some(value) = self.receipts.get(&mapping_key)? {
                 let mapping: StoredBatchReceipt = postcard::from_bytes(value.as_ref())?;
-                batch.remove(&self.receipts, batch_reverse_key(&mapping.id));
+                batch
+                    .inner
+                    .remove(&self.receipts, batch_reverse_key(&mapping.id));
+            }
+            batch.inner.remove(&self.receipt_order, order.clone());
+            batch.inner.remove(&self.receipts, mapping_key);
+            batch.receipt_trim.batches.floor = Some(order.to_vec());
+            removed += 1;
+            if removed == excess {
+                break;
             }
-            batch.remove(&self.receipt_order, order);
-            batch.remove(&self.receipts, mapping_key);
         }
+        batch.receipt_trim.batches.removed += removed;
         Ok(())
     }
 
@@ -11554,9 +11649,8 @@ impl GraphStore {
         {
             return Err(StoreError::ReceiptConflict);
         }
-        if mapping.sequence.is_multiple_of(RECEIPT_TRIM_INTERVAL) {
-            self.trim_batch_receipts(&mut batch.inner)?;
-        }
+        self.trim_batch_receipts(batch)?;
+        batch.receipt_trim.batches.added += 1;
         batch
             .inner
             .insert(&self.receipts, key, postcard::to_allocvec(&mapping)?);
@@ -11809,6 +11903,29 @@ impl GraphStore {
 
     /// Take the FTS queue lock, recovering from poison: the state it guards
     /// lives in fjall, not behind the mutex.
+    /// Oldest queued search work. Callers hold the queue lock, under which tokens are
+    /// minted in increasing order, so no work can appear below a head once seen.
+    fn queue_head(&self) -> Result<Option<QueueCursor>> {
+        let floor = self.queue_floor.load(Ordering::SeqCst);
+        let mut start = [0u8; 9];
+        start[0] = SEARCH_ORDER_PREFIX;
+        start[1..].copy_from_slice(&floor.to_be_bytes());
+        let head = match self
+            .search_queue
+            .range(start.to_vec()..vec![SEARCH_ORDER_PREFIX + 1])
+            .next()
+        {
+            Some(guard) => Some(decode_search_order(guard.into_inner()?.0.as_ref())?),
+            None => None,
+        };
+        let next = head.map_or_else(
+            || self.dirty_counter.load(Ordering::SeqCst),
+            |head| head.token,
+        );
+        self.queue_floor.fetch_max(next, Ordering::SeqCst);
+        Ok(head)
+    }
+
     fn fts_queue_guard(&self) -> MutexGuard<'_, ()> {
         self.fts_queue_lock
             .lock()
@@ -11875,6 +11992,7 @@ impl GraphStore {
             mut batch,
             pending_fts,
             mut pending_receipts,
+            receipt_trim,
         } = commit;
         let _queue = self.fts_queue_guard();
         let mut search_token = None;
@@ -11901,10 +12019,12 @@ impl GraphStore {
             );
         }
         let result = self.commit_fjall_batch(batch);
-        if result.is_ok()
-            && let Some(token) = search_token
-        {
-            self.dirty_committed.fetch_max(token, Ordering::SeqCst);
+        if result.is_ok() {
+            self.receipt_count.settle(receipt_trim.receipts);
+            self.batch_receipt_count.settle(receipt_trim.batches);
+            if let Some(token) = search_token {
+                self.dirty_committed.fetch_max(token, Ordering::SeqCst);
+            }
         }
         result
     }
@@ -11917,11 +12037,13 @@ impl GraphStore {
             publish,
             pending_fts,
             pending_receipts,
+            receipt_trim,
         } = batch;
         let commit = DurableCommit {
             batch: inner,
             pending_fts,
             pending_receipts,
+            receipt_trim,
         };
         self.apply_commit(commit, publish)
     }
@@ -12132,52 +12254,83 @@ mod tests {
         assert_eq!(resume_range(&[], None), (Included(Vec::new()), Unbounded));
     }
 
+    fn pending_receipt(graph: &GraphId, nanos: usize) -> MutationReceipt {
+        MutationReceipt {
+            id: MutationId::new(),
+            admission_sequence: 0,
+            graph: graph.clone(),
+            request_digest: [7; 32],
+            event_id: None,
+            topic: None,
+            publish_after: None,
+            topic_epoch: None,
+            topic_genesis: None,
+            search_token: None,
+            repair_graphs: Vec::new(),
+            source: SourceOutcome::Prepared,
+            persistence: crate::sync::PersistenceOutcome::Pending,
+            repairs: crate::sync::RepairState {
+                diagnostics: RepairOutcome::Pending,
+                shacl: RepairOutcome::Pending,
+                search: RepairOutcome::Pending,
+                query_view: RepairOutcome::Pending,
+            },
+            source_version: [0; 32],
+            updated_unix_nanos: i64::try_from(nanos).unwrap(),
+        }
+    }
+
+    fn commit_receipt(store: &GraphStore, graph: &GraphId, nanos: usize) -> MutationReceipt {
+        let receipt = pending_receipt(graph, nanos);
+        let mut batch = store.new_batch();
+        assert!(store.stage_receipt(&mut batch, &receipt).unwrap().is_none());
+        store.commit(batch).unwrap();
+        store.mutation_receipt(&receipt.id).unwrap().unwrap()
+    }
+
+    /// A trim in a batch that never commits changes neither the count nor the trim
+    /// position, and a reopened store still retains exactly the newest receipts.
+    #[test]
+    fn dropped_trim_ignored() {
+        let dir = tempfile::tempdir().unwrap();
+        let graph = GraphId::new("urn:test:receipt-dropped");
+        let store = GraphStore::open(dir.path()).unwrap();
+        let first = commit_receipt(&store, &graph, 0);
+        let second = commit_receipt(&store, &graph, 1);
+        let third = commit_receipt(&store, &graph, 2);
+        for nanos in 3..RECEIPT_RETENTION {
+            commit_receipt(&store, &graph, nanos);
+        }
+        let mut dropped = store.new_batch();
+        store
+            .stage_receipt(&mut dropped, &pending_receipt(&graph, RECEIPT_RETENTION))
+            .unwrap();
+        drop(dropped);
+        assert!(store.mutation_receipt(&first.id).unwrap().is_some());
+
+        commit_receipt(&store, &graph, RECEIPT_RETENTION + 1);
+        assert!(store.mutation_receipt(&first.id).unwrap().is_none());
+        assert!(store.mutation_receipt(&second.id).unwrap().is_some());
+        drop(store);
+
+        let store = GraphStore::open(dir.path()).unwrap();
+        commit_receipt(&store, &graph, RECEIPT_RETENTION + 2);
+        assert!(store.mutation_receipt(&second.id).unwrap().is_none());
+        assert!(store.mutation_receipt(&third.id).unwrap().is_some());
+    }
+
     /// Receipts that never settle still leave retention; reopen orders untracked ones again.
     #[test]
     fn pending_receipts_expire() {
         let dir = tempfile::tempdir().unwrap();
         let store = GraphStore::open(dir.path()).unwrap();
         let graph = GraphId::new("urn:test:receipt-retention");
-        let stage = |nanos: usize| {
-            let receipt = MutationReceipt {
-                id: MutationId::new(),
-                admission_sequence: 0,
-                graph: graph.clone(),
-                request_digest: [7; 32],
-                event_id: None,
-                topic: None,
-                publish_after: None,
-                topic_epoch: None,
-                topic_genesis: None,
-                search_token: None,
-                repair_graphs: Vec::new(),
-                source: SourceOutcome::Prepared,
-                persistence: crate::sync::PersistenceOutcome::Pending,
-                repairs: crate::sync::RepairState {
-                    diagnostics: RepairOutcome::Pending,
-                    shacl: RepairOutcome::Pending,
-                    search: RepairOutcome::Pending,
-                    query_view: RepairOutcome::Pending,
-                },
-                source_version: [0; 32],
-                updated_unix_nanos: i64::try_from(nanos).unwrap(),
-            };
-            let mut batch = store.new_batch();
-            assert!(store.stage_receipt(&mut batch, &receipt).unwrap().is_none());
-            store.commit(batch).unwrap();
-            store.mutation_receipt(&receipt.id).unwrap().unwrap()
-        };
+        let stage = |nanos: usize| commit_receipt(&store, &graph, nanos);
         let first = stage(0);
-        // The trim pass at the last admission keeps exactly the newest retained receipts.
-        let total = RECEIPT_RETENTION + RECEIPT_TRIM_INTERVAL as usize;
-        let mut kept = None;
-        for nanos in 1..total {
-            let receipt = stage(nanos);
-            if nanos == total - RECEIPT_RETENTION {
-                kept = Some(receipt);
-            }
+        let second = stage(1);
+        for nanos in 2..=RECEIPT_RETENTION {
+            stage(nanos);
         }
-        let kept = kept.unwrap();
         let lookup = |receipt: &MutationReceipt| MutationLookup {
             graph: graph.clone(),
             id: receipt.id,
@@ -12189,11 +12342,11 @@ mod tests {
             MutationStatus::Expired
         ));
         assert!(matches!(
-            store.receipt_status(&lookup(&kept)).unwrap(),
+            store.receipt_status(&lookup(&second)).unwrap(),
             MutationStatus::Known(_)
         ));
 
-        let order = receipt_order_key(&kept);
+        let order = receipt_order_key(&second);
         store.receipt_order.remove(order).unwrap();
         drop(store);
         let reopened = GraphStore::open(dir.path()).unwrap();
@@ -12650,6 +12803,7 @@ mod tests {
             publish,
             pending_fts,
             pending_receipts: _,
+            receipt_trim: _,
         } = batch;
         (inner, publish, pending_fts)
     }
@@ -12981,6 +13135,7 @@ mod tests {
                 batch: inner,
                 pending_fts,
                 pending_receipts: Vec::new(),
+                receipt_trim: ReceiptTrim::default(),
             })
             .unwrap();
         store.indexes_write().publish(&publish);
@@ -13025,6 +13180,7 @@ mod tests {
                     batch: inner,
                     pending_fts,
                     pending_receipts: Vec::new(),
+                    receipt_trim: ReceiptTrim::default(),
                 })
                 .unwrap();
             store.indexes_write().publish(&publish);
@@ -15722,11 +15878,13 @@ mod tests {
                 publish,
                 pending_fts,
                 pending_receipts: _,
+                receipt_trim: _,
             } = batch;
             let mut durable = DurableCommit {
                 batch: inner,
                 pending_fts,
                 pending_receipts: Vec::new(),
+                receipt_trim: ReceiptTrim::default(),
             };
             let owner = store.qv_gate.try_acquire().unwrap();
             store

From d68789c0212ae3c2912ded2f3718aa97560adb1f Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 09:32:19 +0200
Subject: [PATCH 05/16] test: profile single write latency on a large store

---
 Cargo.toml                  |   4 +
 tests/perf/write_latency.rs | 163 ++++++++++++++++++++++++++++++++++++
 2 files changed, 167 insertions(+)
 create mode 100644 tests/perf/write_latency.rs

diff --git a/Cargo.toml b/Cargo.toml
index 256832e..471272c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -231,6 +231,10 @@ path = "tests/perf/multigraph_query.rs"
 name = "perf_query_matrix"
 path = "tests/perf/query_matrix.rs"
 
+[[test]]
+name = "perf_write_latency"
+path = "tests/perf/write_latency.rs"
+
 [[test]]
 name = "perf_rocrate_workflows"
 path = "tests/perf/rocrate_workflows.rs"
diff --git a/tests/perf/write_latency.rs b/tests/perf/write_latency.rs
new file mode 100644
index 0000000..5baf3e4
--- /dev/null
+++ b/tests/perf/write_latency.rs
@@ -0,0 +1,163 @@
+//! Measures single write latency on a store that already holds many RO-Crate graphs.
+// Copyright (c) 2026 ArunaStorage Team @ JLU Giessen
+// SPDX-License-Identifier: MIT
+
+#[cfg(test)]
+mod tests {
+    use std::time::{Duration, Instant};
+
+    use craqle::*;
+
+    const DEFAULT_GRAPHS: usize = 5_000;
+    const DEFAULT_SAMPLES: usize = 200;
+
+    fn env_usize(name: &str, default: usize) -> usize {
+        std::env::var(name)
+            .ok()
+            .and_then(|value| value.parse().ok())
+            .unwrap_or(default)
+    }
+
+    fn graph_id(index: usize) -> GraphId {
+        GraphId::new(&format!("https://w3id.org/aruna/write-{index:06}"))
+    }
+
+    fn create_graph(node: &CraqleNode, index: usize, durability: CraqleRequestDurability) {
+        let request = CreateCrateRequest::new(
+            graph_id(index),
+            format!("Write latency dataset {index}"),
+            format!("Write latency corpus graph {index}"),
+            "2026-01-01",
+            Some("https://creativecommons.org/licenses/by/4.0/".to_string()),
+            GraphPolicy {
+                public: true,
+                permission_paths: vec![format!("/realm/g/group/meta/doc-{index}")],
+            },
+        );
+        node.create_crate_with_durability_as(&AllowAllAuthorizer, request, durability, None)
+            .unwrap();
+    }
+
+    fn report(label: &str, mut samples: Vec<Duration>) {
+        samples.sort();
+        let at = |pct: usize| samples[(samples.len() - 1) * pct / 100];
+        println!(
+            "{label}: n={} p50={:?} p95={:?} p99={:?} max={:?}",
+            samples.len(),
+            at(50),
+            at(95),
+            at(99),
+            samples.last().copied().unwrap_or_default()
+        );
+    }
+
+    fn timed(samples: usize, mut run: impl FnMut(usize)) -> Vec<Duration> {
+        (0..samples)
+            .map(|sample| {
+                let started = Instant::now();
+                run(sample);
+                started.elapsed()
+            })
+            .collect()
+    }
+
+    #[test]
+    #[ignore = "release-only write latency profile"]
+    fn write_latency_profile() {
+        let graphs = env_usize("CRAQLE_PERF_WRITE_GRAPHS", DEFAULT_GRAPHS);
+        let samples = env_usize("CRAQLE_PERF_WRITE_SAMPLES", DEFAULT_SAMPLES);
+        let dir = tempfile::tempdir().unwrap();
+        let node = CraqleNode::open_with_options(
+            dir.path(),
+            CraqleOptions::new().with_graph_store_persist_mode(CraqleFjallPersistMode::Buffer),
+        )
+        .unwrap();
+
+        let seeded = Instant::now();
+        let seed = timed(graphs, |index| {
+            create_graph(&node, index, CraqleRequestDurability::WalAlreadyDurable)
+        });
+        println!("seeded {graphs} graphs in {:?}", seeded.elapsed());
+        report(
+            "seed create (last samples)",
+            seed[graphs.saturating_sub(samples)..].to_vec(),
+        );
+
+        let mut next = graphs;
+        report(
+            "create wal",
+            timed(samples, |_| {
+                create_graph(&node, next, CraqleRequestDurability::WalAlreadyDurable);
+                next += 1;
+            }),
+        );
+        report(
+            "create durable",
+            timed(samples, |_| {
+                create_graph(&node, next, CraqleRequestDurability::Durable);
+                next += 1;
+            }),
+        );
+
+        let documents = (0..samples)
+            .map(|sample| {
+                let graph = graph_id(sample * graphs / samples);
+                let exported = node.export_rocrate(&AllowAllAuthorizer, &graph).unwrap();
+                let mut document: serde_json::Value = serde_json::from_str(&exported).unwrap();
+                let entities = document["@graph"].as_array_mut().unwrap();
+                entities.push(serde_json::json!({
+                    "@id": format!("./data/file-{sample}.csv"),
+                    "@type": "File",
+                    "name": format!("Measured file {sample}"),
+                    "encodingFormat": "text/csv"
+                }));
+                let root = entities
+                    .iter_mut()
+                    .find(|entity| entity["@id"] == graph.as_str())
+                    .unwrap();
+                root["hasPart"] =
+                    serde_json::json!([{ "@id": format!("./data/file-{sample}.csv") }]);
+                (graph, document.to_string())
+            })
+            .collect::<Vec<_>>();
+        report(
+            "apply update",
+            timed(samples, |sample| {
+                let (graph, jsonld) = &documents[sample];
+                node.apply_rocrate_document_checked_with_policy_and_durability_as(
+                    &AllowAllAuthorizer,
+                    graph.clone(),
+                    jsonld,
+                    GraphPolicy {
+                        public: true,
+                        permission_paths: vec![format!("/realm/g/group/meta/doc-{sample}")],
+                    },
+                    CraqleRequestDurability::WalAlreadyDurable,
+                    None,
+                )
+                .unwrap();
+            }),
+        );
+        report(
+            "set policy",
+            timed(samples, |sample| {
+                node.set_graph_policy(
+                    &AllowAllAuthorizer,
+                    &graph_id(sample),
+                    GraphPolicy {
+                        public: false,
+                        permission_paths: vec![format!("/realm/g/group/meta/doc-{sample}")],
+                    },
+                )
+                .unwrap();
+            }),
+        );
+        report(
+            "delete graph",
+            timed(samples, |sample| {
+                node.delete_graph(&AllowAllAuthorizer, &graph_id(graphs - 1 - sample))
+                    .unwrap();
+            }),
+        );
+    }
+}

From 0f99ecb28192a5322d0ba030e45eef3f2fa35ba6 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 09:48:23 +0200
Subject: [PATCH 06/16] chore: use the renamed atomic update for receipt counts

---
 src/internal/store.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/internal/store.rs b/src/internal/store.rs
index 1ba8c39..54af55c 100644
--- a/src/internal/store.rs
+++ b/src/internal/store.rs
@@ -1054,7 +1054,7 @@ impl TrimCount {
         self.live.fetch_add(change.added, Ordering::SeqCst);
         let _ = self
             .live
-            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |live| {
+            .try_update(Ordering::SeqCst, Ordering::SeqCst, |live| {
                 Some(live.saturating_sub(change.removed))
             });
         if let Some(floor) = change.floor {

From ed25a3f3844fa3a747627046d9a5bc9ef1918640 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 09:48:23 +0200
Subject: [PATCH 07/16] test: size updated crates in the write latency profile

---
 tests/perf/write_latency.rs | 71 ++++++++++++++++++++++---------------
 1 file changed, 43 insertions(+), 28 deletions(-)

diff --git a/tests/perf/write_latency.rs b/tests/perf/write_latency.rs
index 5baf3e4..0df5440 100644
--- a/tests/perf/write_latency.rs
+++ b/tests/perf/write_latency.rs
@@ -66,6 +66,7 @@ mod tests {
     fn write_latency_profile() {
         let graphs = env_usize("CRAQLE_PERF_WRITE_GRAPHS", DEFAULT_GRAPHS);
         let samples = env_usize("CRAQLE_PERF_WRITE_SAMPLES", DEFAULT_SAMPLES);
+        let entities = env_usize("CRAQLE_PERF_WRITE_ENTITIES", 0);
         let dir = tempfile::tempdir().unwrap();
         let node = CraqleNode::open_with_options(
             dir.path(),
@@ -99,43 +100,57 @@ mod tests {
             }),
         );
 
+        let policy = |sample: usize| GraphPolicy {
+            public: true,
+            permission_paths: vec![format!("/realm/g/group/meta/doc-{sample}")],
+        };
+        // Each updated crate first holds `entities` files; the timed update adds one more.
+        let with_files = |graph: &GraphId, sample: usize, count: usize| {
+            let exported = node.export_rocrate(&AllowAllAuthorizer, graph).unwrap();
+            let mut document: serde_json::Value = serde_json::from_str(&exported).unwrap();
+            let files = (0..count)
+                .map(|file| format!("./data/file-{sample}-{file}.csv"))
+                .collect::<Vec<_>>();
+            let items = document["@graph"].as_array_mut().unwrap();
+            items.extend(files.iter().map(|id| {
+                serde_json::json!({"@id": id, "@type": "File", "name": id, "encodingFormat": "text/csv"})
+            }));
+            let root = items
+                .iter_mut()
+                .find(|entity| entity["@id"] == graph.as_str())
+                .unwrap();
+            root["hasPart"] = files
+                .iter()
+                .map(|id| serde_json::json!({"@id": id}))
+                .collect();
+            document.to_string()
+        };
+        let apply = |graph: &GraphId, sample: usize, jsonld: &str| {
+            node.apply_rocrate_document_checked_with_policy_and_durability_as(
+                &AllowAllAuthorizer,
+                graph.clone(),
+                jsonld,
+                policy(sample),
+                CraqleRequestDurability::WalAlreadyDurable,
+                None,
+            )
+            .unwrap();
+        };
         let documents = (0..samples)
             .map(|sample| {
                 let graph = graph_id(sample * graphs / samples);
-                let exported = node.export_rocrate(&AllowAllAuthorizer, &graph).unwrap();
-                let mut document: serde_json::Value = serde_json::from_str(&exported).unwrap();
-                let entities = document["@graph"].as_array_mut().unwrap();
-                entities.push(serde_json::json!({
-                    "@id": format!("./data/file-{sample}.csv"),
-                    "@type": "File",
-                    "name": format!("Measured file {sample}"),
-                    "encodingFormat": "text/csv"
-                }));
-                let root = entities
-                    .iter_mut()
-                    .find(|entity| entity["@id"] == graph.as_str())
-                    .unwrap();
-                root["hasPart"] =
-                    serde_json::json!([{ "@id": format!("./data/file-{sample}.csv") }]);
-                (graph, document.to_string())
+                if entities > 0 {
+                    apply(&graph, sample, &with_files(&graph, sample, entities));
+                }
+                let jsonld = with_files(&graph, sample, entities + 1);
+                (graph, jsonld)
             })
             .collect::<Vec<_>>();
         report(
             "apply update",
             timed(samples, |sample| {
                 let (graph, jsonld) = &documents[sample];
-                node.apply_rocrate_document_checked_with_policy_and_durability_as(
-                    &AllowAllAuthorizer,
-                    graph.clone(),
-                    jsonld,
-                    GraphPolicy {
-                        public: true,
-                        permission_paths: vec![format!("/realm/g/group/meta/doc-{sample}")],
-                    },
-                    CraqleRequestDurability::WalAlreadyDurable,
-                    None,
-                )
-                .unwrap();
+                apply(graph, sample, jsonld);
             }),
         );
         report(

From e05945960e22bc7ab0b77ea60bc529d98eed8a17 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 10:38:25 +0200
Subject: [PATCH 08/16] chore: keep the queue lock comment on its function

---
 src/internal/store.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/internal/store.rs b/src/internal/store.rs
index 54af55c..ff201a5 100644
--- a/src/internal/store.rs
+++ b/src/internal/store.rs
@@ -11901,8 +11901,6 @@ impl GraphStore {
         Ok(true)
     }
 
-    /// Take the FTS queue lock, recovering from poison: the state it guards
-    /// lives in fjall, not behind the mutex.
     /// Oldest queued search work. Callers hold the queue lock, under which tokens are
     /// minted in increasing order, so no work can appear below a head once seen.
     fn queue_head(&self) -> Result<Option<QueueCursor>> {
@@ -11926,6 +11924,8 @@ impl GraphStore {
         Ok(head)
     }
 
+    /// Take the FTS queue lock, recovering from poison: the state it guards
+    /// lives in fjall, not behind the mutex.
     fn fts_queue_guard(&self) -> MutexGuard<'_, ()> {
         self.fts_queue_lock
             .lock()

From efe58fe062d926700f88a31e2d02b04765b6f68e Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 11:19:39 +0200
Subject: [PATCH 09/16] fix: keep receipt retention and search coverage exact
 under concurrent commits

---
 src/internal/search/mod.rs |   8 +-
 src/internal/store.rs      | 457 +++++++++++++++++++++++++++----------
 src/lib.rs                 |  30 ++-
 3 files changed, 369 insertions(+), 126 deletions(-)

diff --git a/src/internal/search/mod.rs b/src/internal/search/mod.rs
index 9b8df46..2a54689 100644
--- a/src/internal/search/mod.rs
+++ b/src/internal/search/mod.rs
@@ -715,6 +715,7 @@ struct PrepareSubject<'a> {
 }
 
 struct OrphanInput<'a> {
+    store: &'a GraphStore,
     graph: &'a GraphId,
     graph_tid: TermId,
     byte_limit: usize,
@@ -740,7 +741,11 @@ impl StoreSyncCaches {
     fn orphaned(&mut self, input: OrphanInput<'_>) -> Result<&HashSet<TermId>> {
         if !self.orphaned_subjects.contains_key(input.graph) {
             let artifact_limit = input.byte_limit / 32;
-            let orphaned = match self.snapshot.orphaned_ids(input.graph_tid, artifact_limit) {
+            let scope = crate::store::OrphanScope {
+                graph: input.graph_tid,
+                byte_limit: artifact_limit,
+            };
+            let orphaned = match input.store.search_orphan_ids(&self.snapshot, scope) {
                 Ok(orphaned) => orphaned,
                 Err(crate::store::StoreError::LimitExceeded {
                     resource: "search diagnostics rows",
@@ -3595,6 +3600,7 @@ fn prepare_subject_op(
     let cache_before = caches.orphan_bytes;
     let hidden = caches
         .orphaned(OrphanInput {
+            store: req.store,
             graph: req.graph,
             graph_tid,
             byte_limit: text_limit,
diff --git a/src/internal/store.rs b/src/internal/store.rs
index ff201a5..2b7b5e4 100644
--- a/src/internal/store.rs
+++ b/src/internal/store.rs
@@ -6,7 +6,7 @@ use std::cell::{Cell, OnceCell, RefCell};
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
 use std::ops::Bound::{self, Excluded, Included, Unbounded};
 use std::path::Path;
-use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{
     Arc, Mutex, MutexGuard, PoisonError, RwLock, RwLockReadGuard as ReadGuard,
     RwLockWriteGuard as WriteGuard, TryLockError,
@@ -1011,58 +1011,34 @@ struct DurableCommit {
     receipt_trim: ReceiptTrim,
 }
 
-/// Receipt count changes and trim positions a batch applies once it commits.
+/// Batch receipt mappings a batch stages; retention settles when the batch commits.
 #[derive(Default)]
 struct ReceiptTrim {
-    receipts: TrimChange,
-    batches: TrimChange,
+    mappings: Vec<(Vec<u8>, u64)>,
 }
 
+/// Committed receipt counts and trim floors. Commits that change receipt order hold its
+/// lock from trim selection until the counts are updated, so both stay exact.
 #[derive(Default)]
-struct TrimChange {
-    added: usize,
-    removed: usize,
-    floor: Option<Vec<u8>>,
+struct RetentionState {
+    receipts: usize,
+    batches: usize,
+    /// No committed order key sorts below a floor; `None` scans from the start.
+    receipt_floor: Option<Vec<u8>>,
+    batch_floor: Option<Vec<u8>>,
 }
 
-/// Committed entries of one retained kind; races only lower the count, so a
-/// trim never evicts a retained entry. The floor skips earlier trims' tombstones.
-#[derive(Default)]
-struct TrimCount {
-    live: AtomicUsize,
-    floor: Mutex<Option<Vec<u8>>>,
+/// One commit's view of retention: the committed state and what the batch writes.
+struct RetentionTrim<'a> {
+    state: &'a RetentionState,
+    pending: &'a [MutationReceipt],
+    mappings: &'a [(Vec<u8>, u64)],
 }
 
-impl TrimCount {
-    fn excess(&self, change: &TrimChange) -> usize {
-        (self.live.load(Ordering::SeqCst) + change.added + 1)
-            .saturating_sub(change.removed)
-            .saturating_sub(RECEIPT_RETENTION)
-            .min(QV_BUILD_ROWS)
-    }
-
-    fn start(&self, change: &TrimChange) -> Option<Vec<u8>> {
-        change.floor.clone().or_else(|| {
-            self.floor
-                .lock()
-                .unwrap_or_else(PoisonError::into_inner)
-                .clone()
-        })
-    }
-
-    fn settle(&self, change: TrimChange) {
-        self.live.fetch_add(change.added, Ordering::SeqCst);
-        let _ = self
-            .live
-            .try_update(Ordering::SeqCst, Ordering::SeqCst, |live| {
-                Some(live.saturating_sub(change.removed))
-            });
-        if let Some(floor) = change.floor {
-            let mut current = self.floor.lock().unwrap_or_else(PoisonError::into_inner);
-            if current.as_ref().is_none_or(|current| *current < floor) {
-                *current = Some(floor);
-            }
-        }
+/// Lowers a floor to an order key a commit inserts below it.
+fn lower_floor(floor: &mut Option<Vec<u8>>, key: &[u8]) {
+    if floor.as_deref().is_some_and(|current| key < current) {
+        *floor = Some(key.to_vec());
     }
 }
 
@@ -1334,8 +1310,7 @@ pub struct GraphStore {
     write_locks: Vec<Mutex<()>>,
     receipt_locks: Vec<Mutex<()>>,
     receipt_next: AtomicU64,
-    receipt_count: TrimCount,
-    batch_receipt_count: TrimCount,
+    retention: Mutex<RetentionState>,
     #[cfg(test)]
     receipt_lookups: AtomicU64,
     #[cfg(test)]
@@ -1430,7 +1405,7 @@ pub struct GraphStore {
     /// Serializes FTS queue mutations and is innermost in the store lock order.
     fts_queue_lock: Mutex<()>,
     dirty_counter: AtomicU64,
-    /// No queued search work has a lower token; skips acknowledged queue tombstones.
+    /// No queued search work has a lower token; the scan skips acknowledged tombstones.
     queue_floor: AtomicU64,
     dirty_committed: AtomicU64,
     /// Number of graph diagnostics recomputations by this store instance.
@@ -2523,6 +2498,12 @@ pub(crate) struct QvRead<'a> {
     pub(crate) costs: &'a crate::query::context::QueryCost,
 }
 
+/// The graph and byte budget of one search orphan lookup.
+pub(crate) struct OrphanScope {
+    pub(crate) graph: TermId,
+    pub(crate) byte_limit: usize,
+}
+
 #[derive(Clone)]
 pub(crate) struct SearchSnapshot {
     snapshot: Snapshot,
@@ -6607,6 +6588,26 @@ impl GraphStore {
         self.commit_fjall_batch(batch).unwrap();
     }
 
+    /// Orphan ids at a search snapshot. Diagnostics commit after their source batch, so
+    /// a snapshot between the two recomputes them instead of failing the subject.
+    pub(crate) fn search_orphan_ids(
+        &self,
+        snapshot: &SearchSnapshot,
+        scope: OrphanScope,
+    ) -> Result<HashSet<TermId>> {
+        match snapshot.orphaned_ids(scope.graph, scope.byte_limit) {
+            Err(StoreError::InvalidSearchState(
+                "search-diagnostics-stale" | "search-diagnostics-missing",
+            )) => self.snapshot_orphan_ids(
+                &snapshot.snapshot,
+                &crate::query::context::ReadContext::default(),
+                scope.graph,
+                &self.orphan_vocab()?,
+            ),
+            result => result,
+        }
+    }
+
     /// Resolves the vocabulary ids used by orphan detection.
     fn orphan_vocab(&self) -> Result<OrphanVocab> {
         let id = |named_node: oxrdf::NamedNode| {
@@ -7570,8 +7571,7 @@ impl GraphStore {
             write_locks: (0..GRAPH_LOCK_SHARDS).map(|_| Mutex::new(())).collect(),
             receipt_locks: (0..COMMIT_LOCK_SHARDS).map(|_| Mutex::new(())).collect(),
             receipt_next: AtomicU64::new(1),
-            receipt_count: TrimCount::default(),
-            batch_receipt_count: TrimCount::default(),
+            retention: Mutex::new(RetentionState::default()),
             #[cfg(test)]
             receipt_lookups: AtomicU64::new(0),
             #[cfg(test)]
@@ -7846,15 +7846,17 @@ impl GraphStore {
         }
         self.receipt_next
             .store(highest.saturating_add(1).max(1), Ordering::SeqCst);
-        self.receipt_count.live.store(live, Ordering::SeqCst);
         let mut batches = 0usize;
         for guard in self.receipt_order.prefix(BATCH_ORDER_PREFIX) {
             guard.into_inner()?;
             batches += 1;
         }
-        self.batch_receipt_count
-            .live
-            .store(batches, Ordering::SeqCst);
+        *self.retention_guard() = RetentionState {
+            receipts: live,
+            batches,
+            receipt_floor: None,
+            batch_floor: None,
+        };
         Ok(())
     }
 
@@ -11430,17 +11432,63 @@ impl GraphStore {
         Ok(())
     }
 
-    /// Makes room for one more receipt by removing the oldest past retention, sparing
-    /// receipts staged in this batch. It scans forward from the last committed trim.
-    fn trim_receipts(&self, batch: &mut WriteBatch) -> Result<()> {
-        let excess = self.receipt_count.excess(&batch.receipt_trim.receipts);
+    fn retention_guard(&self) -> MutexGuard<'_, RetentionState> {
+        self.retention
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+    }
+
+    /// Stages retention trims against committed receipts; callers hold the retention lock.
+    fn stage_retention(
+        &self,
+        batch: &mut fjall::OwnedWriteBatch,
+        trim: RetentionTrim<'_>,
+    ) -> Result<RetentionState> {
+        let (receipts, receipt_floor) = self.trim_receipt_order(batch, &trim)?;
+        let (batches, batch_floor) = self.trim_batch_mappings(batch, &trim)?;
+        Ok(RetentionState {
+            receipts,
+            batches,
+            receipt_floor,
+            batch_floor,
+        })
+    }
+
+    /// Removes the oldest committed receipts past retention, returning the new count and
+    /// floor. Receipts this batch writes count once and are never removed by it.
+    fn trim_receipt_order(
+        &self,
+        batch: &mut fjall::OwnedWriteBatch,
+        trim: &RetentionTrim<'_>,
+    ) -> Result<(usize, Option<Vec<u8>>)> {
+        let mut staged = HashSet::new();
+        let mut added = 0usize;
+        let mut floor = trim.state.receipt_floor.clone();
+        let mut lowest: Option<Vec<u8>> = None;
+        for receipt in trim.pending {
+            if !staged.insert(receipt.id.0) {
+                continue;
+            }
+            if !self.receipts.contains_key(receipt.id.0)? {
+                added += 1;
+            }
+            let order = receipt_order_key(receipt);
+            lower_floor(&mut floor, &order);
+            if lowest
+                .as_deref()
+                .is_none_or(|lowest| order.as_slice() < lowest)
+            {
+                lowest = Some(order.to_vec());
+            }
+        }
+        let live = trim.state.receipts + added;
+        let excess = live.saturating_sub(RECEIPT_RETENTION).min(QV_BUILD_ROWS);
         if excess == 0 {
-            return Ok(());
+            return Ok((live, floor));
         }
-        let start = self
-            .receipt_count
-            .start(&batch.receipt_trim.receipts)
-            .map_or(Bound::Unbounded, Bound::Excluded);
+        let start = floor.clone().map_or(Bound::Unbounded, Bound::Included);
+        let mut next = floor;
+        let mut advance = true;
         let mut expired = None;
         let mut removed = 0usize;
         for guard in self
@@ -11451,39 +11499,100 @@ impl GraphStore {
             if key.len() != 40 {
                 continue;
             }
-            let staged = batch
-                .pending_receipts
-                .iter()
-                .any(|receipt| receipt.id.0[..] == id[..]);
-            if staged {
+            // A receipt this batch rewrites stays live, so the floor must not pass it.
+            if <[u8; 32]>::try_from(id.as_ref()).is_ok_and(|id| staged.contains(&id)) {
+                advance = false;
                 continue;
             }
             if let Some(value) = self.receipts.get(&id)? {
                 let receipt: MutationReceipt = postcard::from_bytes(value.as_ref())?;
                 expired = expired.max(Some(receipt.admission_sequence));
             }
-            batch.inner.remove(&self.receipt_order, key.clone());
-            batch.inner.remove(&self.receipts, id);
-            batch.receipt_trim.receipts.floor = Some(key.to_vec());
+            batch.remove(&self.receipt_order, key.clone());
+            batch.remove(&self.receipts, id);
+            if advance {
+                next = Some(key.to_vec());
+            }
             removed += 1;
             if removed == excess {
                 break;
             }
         }
-        batch.receipt_trim.receipts.removed += removed;
+        // This batch's own keys are not visible to the scan; the floor must stay below them.
+        if let Some(lowest) = lowest {
+            lower_floor(&mut next, &lowest);
+        }
         if let Some(expired) = expired {
             let stored = self
                 .receipt_order
                 .get(RECEIPT_EXPIRED_KEY)?
                 .and_then(|value| decode_index_count(value.as_ref()))
                 .unwrap_or(0);
-            batch.inner.insert(
+            batch.insert(
                 &self.receipt_order,
                 RECEIPT_EXPIRED_KEY,
                 stored.max(expired).to_be_bytes(),
             );
         }
-        Ok(())
+        Ok((live - removed, next))
+    }
+
+    /// Removes the oldest committed batch receipts past retention, returning the new
+    /// count and floor. Mappings this batch writes are not committed yet.
+    fn trim_batch_mappings(
+        &self,
+        batch: &mut fjall::OwnedWriteBatch,
+        trim: &RetentionTrim<'_>,
+    ) -> Result<(usize, Option<Vec<u8>>)> {
+        let mut staged = HashSet::new();
+        let mut added = 0usize;
+        let mut floor = trim.state.batch_floor.clone();
+        let lowest = trim
+            .mappings
+            .iter()
+            .map(|(_, sequence)| batch_order_key(*sequence))
+            .min();
+        for (key, sequence) in trim.mappings {
+            if staged.insert(key) && !self.receipts.contains_key(key)? {
+                added += 1;
+            }
+            lower_floor(&mut floor, &batch_order_key(*sequence));
+        }
+        let live = trim.state.batches + added;
+        let excess = live.saturating_sub(RECEIPT_RETENTION).min(QV_BUILD_ROWS);
+        if excess == 0 {
+            return Ok((live, floor));
+        }
+        let start = floor.clone().map_or(
+            Bound::Included(BATCH_ORDER_PREFIX.to_vec()),
+            Bound::Included,
+        );
+        let mut next = floor;
+        let mut removed = 0usize;
+        for guard in self
+            .receipt_order
+            .range::<Vec<u8>, _>((start, Bound::Unbounded))
+        {
+            let (order, mapping_key) = guard.into_inner()?;
+            if !order.starts_with(&BATCH_ORDER_PREFIX) {
+                break;
+            }
+            if let Some(value) = self.receipts.get(&mapping_key)? {
+                let mapping: StoredBatchReceipt = postcard::from_bytes(value.as_ref())?;
+                batch.remove(&self.receipts, batch_reverse_key(&mapping.id));
+            }
+            batch.remove(&self.receipt_order, order.clone());
+            batch.remove(&self.receipts, mapping_key);
+            next = Some(order.to_vec());
+            removed += 1;
+            if removed == excess {
+                break;
+            }
+        }
+        if let Some(lowest) = lowest {
+            lower_floor(&mut next, &lowest);
+        }
+        Ok((live - removed, next))
     }
 
     /// Stage first acceptance evidence in the caller's source batch.
@@ -11512,8 +11621,6 @@ impl GraphStore {
         }
         #[cfg(test)]
         self.receipt_writes.fetch_add(1, Ordering::Relaxed);
-        self.trim_receipts(batch)?;
-        batch.receipt_trim.receipts.added += 1;
         batch
             .inner
             .insert(&self.receipts, stored.id.0, postcard::to_allocvec(&stored)?);
@@ -11539,6 +11646,7 @@ impl GraphStore {
         event_id: [u8; 32],
     ) -> Result<MutationReceipt> {
         let _guard = self.receipt_guard(id);
+        let _retention = self.retention_guard();
         let mut receipt = self
             .mutation_receipt(id)?
             .ok_or(StoreError::ReceiptConflict)?;
@@ -11582,44 +11690,6 @@ impl GraphStore {
         })
     }
 
-    /// Makes room for one more batch receipt by removing the oldest past retention.
-    fn trim_batch_receipts(&self, batch: &mut WriteBatch) -> Result<()> {
-        let change = &batch.receipt_trim.batches;
-        let excess = self.batch_receipt_count.excess(change);
-        if excess == 0 {
-            return Ok(());
-        }
-        let start = self.batch_receipt_count.start(change).map_or(
-            Bound::Included(BATCH_ORDER_PREFIX.to_vec()),
-            Bound::Excluded,
-        );
-        let mut removed = 0usize;
-        for guard in self
-            .receipt_order
-            .range::<Vec<u8>, _>((start, Bound::Unbounded))
-        {
-            let (order, mapping_key) = guard.into_inner()?;
-            if !order.starts_with(&BATCH_ORDER_PREFIX) {
-                break;
-            }
-            if let Some(value) = self.receipts.get(&mapping_key)? {
-                let mapping: StoredBatchReceipt = postcard::from_bytes(value.as_ref())?;
-                batch
-                    .inner
-                    .remove(&self.receipts, batch_reverse_key(&mapping.id));
-            }
-            batch.inner.remove(&self.receipt_order, order.clone());
-            batch.inner.remove(&self.receipts, mapping_key);
-            batch.receipt_trim.batches.floor = Some(order.to_vec());
-            removed += 1;
-            if removed == excess {
-                break;
-            }
-        }
-        batch.receipt_trim.batches.removed += removed;
-        Ok(())
-    }
-
     pub(crate) fn stage_batch_receipt(
         &self,
         batch: &mut WriteBatch,
@@ -11649,8 +11719,10 @@ impl GraphStore {
         {
             return Err(StoreError::ReceiptConflict);
         }
-        self.trim_batch_receipts(batch)?;
-        batch.receipt_trim.batches.added += 1;
+        batch
+            .receipt_trim
+            .mappings
+            .push((key.to_vec(), mapping.sequence));
         batch
             .inner
             .insert(&self.receipts, key, postcard::to_allocvec(&mapping)?);
@@ -11759,6 +11831,8 @@ impl GraphStore {
     /// Compare-fenced update for post-commit persistence and repair settlement.
     pub(crate) fn update_receipt(&self, receipt: &MutationReceipt) -> Result<MutationReceipt> {
         let _guard = self.receipt_guard(&receipt.id);
+        // Trims run under this lock, so a trimmed receipt is never written back.
+        let mut retention = self.retention_guard();
         let previous = self
             .mutation_receipt(&receipt.id)?
             .ok_or(StoreError::ReceiptConflict)?;
@@ -11788,6 +11862,7 @@ impl GraphStore {
         #[cfg(test)]
         self.receipt_writes.fetch_add(1, Ordering::Relaxed);
         self.commit_fjall_batch(batch)?;
+        lower_floor(&mut retention.receipt_floor, &receipt_order_key(&stored));
         Ok(stored)
     }
 
@@ -11883,6 +11958,9 @@ impl GraphStore {
                 entry.key.clone(),
                 encode_dirty_tokens(narrowed),
             );
+            // A stale acknowledgement can move queued work below the cached head.
+            self.queue_floor
+                .fetch_min(narrowed.oldest, Ordering::SeqCst);
             batch.insert(
                 &self.search_queue,
                 search_order_key(QueueCursor {
@@ -11901,8 +11979,8 @@ impl GraphStore {
         Ok(true)
     }
 
-    /// Oldest queued search work. Callers hold the queue lock, under which tokens are
-    /// minted in increasing order, so no work can appear below a head once seen.
+    /// Oldest queued search work. Callers hold the queue lock; every queue insert lowers
+    /// the floor to its token first, so the scan never starts past queued work.
     fn queue_head(&self) -> Result<Option<QueueCursor>> {
         let floor = self.queue_floor.load(Ordering::SeqCst);
         let mut start = [0u8; 9];
@@ -11959,6 +12037,7 @@ impl GraphStore {
             },
         };
         batch.insert(&self.graphs, bytes, encode_dirty_tokens(tokens));
+        self.queue_floor.fetch_min(tokens.oldest, Ordering::SeqCst);
         batch.insert(
             &self.search_queue,
             search_order_key(key.cursor(tokens.oldest)),
@@ -12018,10 +12097,24 @@ impl GraphStore {
                 search_token.to_be_bytes(),
             );
         }
+        let mut retention = (!pending_receipts.is_empty() || !receipt_trim.mappings.is_empty())
+            .then(|| self.retention_guard());
+        let settled = match &retention {
+            Some(state) => Some(self.stage_retention(
+                &mut batch,
+                RetentionTrim {
+                    state,
+                    pending: &pending_receipts,
+                    mappings: &receipt_trim.mappings,
+                },
+            )?),
+            None => None,
+        };
         let result = self.commit_fjall_batch(batch);
         if result.is_ok() {
-            self.receipt_count.settle(receipt_trim.receipts);
-            self.batch_receipt_count.settle(receipt_trim.batches);
+            if let (Some(state), Some(settled)) = (retention.as_mut(), settled) {
+                **state = settled;
+            }
             if let Some(token) = search_token {
                 self.dirty_committed.fetch_max(token, Ordering::SeqCst);
             }
@@ -12319,6 +12412,132 @@ mod tests {
         assert!(store.mutation_receipt(&third.id).unwrap().is_some());
     }
 
+    fn retained_receipts(store: &GraphStore) -> usize {
+        store
+            .receipts
+            .iter()
+            .map(|guard| guard.key().unwrap())
+            .filter(|key| key.len() == 32)
+            .count()
+    }
+
+    fn staged_receipt(store: &GraphStore, graph: &GraphId, nanos: usize) -> WriteBatch {
+        let mut batch = store.new_batch();
+        assert!(
+            store
+                .stage_receipt(&mut batch, &pending_receipt(graph, nanos))
+                .unwrap()
+                .is_none()
+        );
+        batch
+    }
+
+    /// Batches staged before either commits still settle retention exactly.
+    #[test]
+    fn overlapping_trims_recover() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = GraphStore::open(dir.path()).unwrap();
+        let graph = GraphId::new("urn:test:receipt-overlap");
+        for nanos in 0..RECEIPT_RETENTION {
+            commit_receipt(&store, &graph, nanos);
+        }
+        for pair in 0..4 {
+            let nanos = RECEIPT_RETENTION + pair * 2;
+            let first = staged_receipt(&store, &graph, nanos);
+            let second = staged_receipt(&store, &graph, nanos + 1);
+            store.commit(first).unwrap();
+            store.commit(second).unwrap();
+        }
+        commit_receipt(&store, &graph, RECEIPT_RETENTION + 8);
+        assert_eq!(retained_receipts(&store), RECEIPT_RETENTION);
+    }
+
+    /// A receipt committed with an older timestamp than earlier trims expires first.
+    #[test]
+    fn delayed_receipt_expires() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = GraphStore::open(dir.path()).unwrap();
+        let graph = GraphId::new("urn:test:receipt-delayed");
+        let delayed = staged_receipt(&store, &graph, 0);
+        for nanos in 1..=RECEIPT_RETENTION + 1 {
+            commit_receipt(&store, &graph, nanos);
+        }
+        let id = delayed.pending_receipts[0].id;
+        store.commit(delayed).unwrap();
+        for nanos in 2..=4 {
+            commit_receipt(&store, &graph, RECEIPT_RETENTION + nanos);
+        }
+        assert!(store.mutation_receipt(&id).unwrap().is_none());
+        assert_eq!(retained_receipts(&store), RECEIPT_RETENTION);
+    }
+
+    /// Search indexing between a source commit and its diagnostics commit derives the
+    /// orphans from its own snapshot instead of failing the subject.
+    #[test]
+    fn search_orphans_recomputed() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = GraphStore::open(dir.path()).unwrap();
+        let graph = GraphId::new("urn:test:search-orphans");
+        store.create_graph(&graph).unwrap();
+        let graph_tid = store
+            .lookup_term(&EncodedTerm::from_named_node(&graph.0))
+            .unwrap()
+            .unwrap();
+        let snapshot = store.search_snapshot();
+        assert!(matches!(
+            snapshot.orphaned_ids(graph_tid, usize::MAX),
+            Err(StoreError::InvalidSearchState(_))
+        ));
+        let scope = OrphanScope {
+            graph: graph_tid,
+            byte_limit: usize::MAX,
+        };
+        assert!(
+            store
+                .search_orphan_ids(&snapshot, scope)
+                .unwrap()
+                .is_empty()
+        );
+    }
+
+    /// A stale queue clear moving work below a checked head keeps coverage fenced.
+    #[test]
+    fn stale_clear_coverage() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = GraphStore::open(dir.path()).unwrap();
+        let graph = GraphId::new("urn:test:queue-stale");
+        store.create_graph(&graph).unwrap();
+        for _ in 0..3 {
+            store.ensure_reindex(&graph).unwrap();
+            store.clear_fts_queue().unwrap();
+        }
+        let (target, _) = store.ensure_reindex(&graph).unwrap();
+        let coverage = |covered: u64| {
+            let index_id = [7; 16];
+            let manifest = store.search_manifest_digest(index_id).unwrap();
+            SearchCoverage {
+                format: SEARCH_META_FORMAT,
+                index_id,
+                index_revision: 0,
+                covered,
+                rebuild: None,
+                manifest_count: manifest.count,
+                manifest_hash: manifest.hash,
+                manifest_epoch: manifest.epoch,
+            }
+        };
+        store
+            .advance_search_coverage(&coverage(target - 1))
+            .unwrap();
+        store.clear_graph_queue(&graph, target - 2).unwrap();
+        assert!(matches!(
+            store.advance_search_coverage(&coverage(target)),
+            Err(StoreError::InvalidSearchState(
+                "search-coverage-debt-remains"
+            ))
+        ));
+    }
+
     /// Receipts that never settle still leave retention; reopen orders untracked ones again.
     #[test]
     fn pending_receipts_expire() {
diff --git a/src/lib.rs b/src/lib.rs
index 805c70b..0ce610b 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1208,9 +1208,12 @@ fn run_search_worker(receiver: mpsc::Receiver<SearchWorkerMessage>, ctx: SearchW
                 if pending.is_empty() && progress.covered == 0 && certified == Some(target) {
                     continue;
                 }
-                let completed = ctx
-                    .search
-                    .complete_coverage(&ctx.store, target)
+                let completed = ctx.search.complete_coverage(&ctx.store, target);
+                if completed.as_ref().is_err_and(coverage_debt) {
+                    active = true;
+                    continue;
+                }
+                let completed = completed
                     .map_err(MaintenanceFailure::search)
                     .and_then(|()| ctx.store.persist().map_err(MaintenanceFailure::store));
                 if completed.is_ok() {
@@ -1316,9 +1319,24 @@ fn flush_search_queue(store: &GraphStore, search: &SearchIndex) -> Result<()> {
             break;
         }
     }
-    search.complete_coverage(store, target)?;
-    store.persist()?;
-    Ok(())
+    match search.complete_coverage(store, target) {
+        Err(error) if coverage_debt(&error) => flush_search_queue(store, search),
+        result => {
+            result?;
+            store.persist()?;
+            Ok(())
+        }
+    }
+}
+
+/// A concurrent write narrowed queued work below the drained target; drain it again.
+fn coverage_debt(error: &search::SearchError) -> bool {
+    matches!(
+        error,
+        search::SearchError::Store(store::StoreError::InvalidSearchState(
+            "search-coverage-debt-remains"
+        ))
+    )
 }
 
 /// Name what a flush could not cover, with the first entry's diagnostic.

From 8cd40f0839f89effb5412dadeb8c58d9072b382d Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 11:19:40 +0200
Subject: [PATCH 10/16] test: reject empty or oversized write latency samples

---
 tests/perf/write_latency.rs | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/tests/perf/write_latency.rs b/tests/perf/write_latency.rs
index 0df5440..4916c14 100644
--- a/tests/perf/write_latency.rs
+++ b/tests/perf/write_latency.rs
@@ -67,6 +67,10 @@ mod tests {
         let graphs = env_usize("CRAQLE_PERF_WRITE_GRAPHS", DEFAULT_GRAPHS);
         let samples = env_usize("CRAQLE_PERF_WRITE_SAMPLES", DEFAULT_SAMPLES);
         let entities = env_usize("CRAQLE_PERF_WRITE_ENTITIES", 0);
+        assert!(
+            graphs > 0 && (1..=graphs).contains(&samples),
+            "the profile needs at least one graph and between 1 and {graphs} samples"
+        );
         let dir = tempfile::tempdir().unwrap();
         let node = CraqleNode::open_with_options(
             dir.path(),

From 24a8935e8d928ce3d80ebac4853121fc5df48c80 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 12:23:09 +0200
Subject: [PATCH 11/16] chore: compile search-only code only with the search
 feature

---
 src/internal/search/queue.rs |  56 ++++++++++++++++-
 src/internal/store.rs        | 113 +++++++++++++++++++++++++++++++----
 src/lib.rs                   |  21 ++++++-
 src/search_stub.rs           |  27 +--------
 4 files changed, 180 insertions(+), 37 deletions(-)

diff --git a/src/internal/search/queue.rs b/src/internal/search/queue.rs
index de8d335..4d64913 100644
--- a/src/internal/search/queue.rs
+++ b/src/internal/search/queue.rs
@@ -5,8 +5,12 @@
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 
-use crate::core::{EncodedTerm, GraphId};
-use crate::store::{EncodedQuad, TermId};
+#[cfg(feature = "search")]
+use crate::core::EncodedTerm;
+use crate::core::GraphId;
+#[cfg(feature = "search")]
+use crate::store::EncodedQuad;
+use crate::store::TermId;
 
 pub(crate) const SEARCH_META_FORMAT: u16 = 3;
 
@@ -16,6 +20,7 @@ pub(crate) const SEARCH_META_FORMAT: u16 = 3;
 pub(crate) struct GenerationId(pub(crate) u64);
 
 #[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+#[cfg(feature = "search")]
 pub(crate) struct GraphGeneration {
     pub(crate) graph: GraphId,
     pub(crate) active: Option<GenerationId>,
@@ -23,6 +28,7 @@ pub(crate) struct GraphGeneration {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct ManifestScan {
     pub(crate) index_id: [u8; 16],
     pub(crate) after: Option<TermId>,
@@ -31,6 +37,7 @@ pub(crate) struct ManifestScan {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct ManifestRow {
     pub(crate) graph: GraphId,
     pub(crate) active: Option<GenerationId>,
@@ -42,6 +49,7 @@ pub(crate) struct ManifestRow {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct ManifestPage {
     pub(crate) entries: Vec<ManifestRow>,
     pub(crate) next: Option<TermId>,
@@ -51,6 +59,7 @@ pub(crate) struct ManifestPage {
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+#[cfg(feature = "search")]
 pub(crate) struct StageJob {
     pub(crate) graph: GraphId,
     pub(crate) target: u64,
@@ -64,6 +73,7 @@ pub(crate) struct StageJob {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct StageRequest {
     pub(crate) index_id: [u8; 16],
     pub(crate) graph: GraphId,
@@ -72,12 +82,14 @@ pub(crate) struct StageRequest {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct GenerationRequest {
     pub(crate) index_id: [u8; 16],
     pub(crate) graph: GraphId,
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct DeleteGeneration {
     pub(crate) index_id: [u8; 16],
     pub(crate) graph: GraphId,
@@ -85,12 +97,14 @@ pub(crate) struct DeleteGeneration {
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+#[cfg(feature = "search")]
 pub(crate) struct CleanupJob {
     pub(crate) graph: GraphId,
     pub(crate) generation: GenerationId,
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct CleanupScan {
     pub(crate) after: Option<GenerationId>,
     pub(crate) row_limit: usize,
@@ -98,6 +112,7 @@ pub(crate) struct CleanupScan {
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
+#[cfg(feature = "search")]
 pub(crate) struct OversizedCleanup {
     pub(crate) generation: GenerationId,
     pub(crate) bytes: usize,
@@ -105,6 +120,7 @@ pub(crate) struct OversizedCleanup {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct CleanupPage {
     pub(crate) entries: Vec<CleanupJob>,
     pub(crate) next: Option<GenerationId>,
@@ -115,6 +131,7 @@ pub(crate) struct CleanupPage {
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
+#[cfg(feature = "search")]
 pub(crate) struct GenerationSwitch {
     pub(crate) graph: GraphId,
     pub(crate) previous: Option<GenerationId>,
@@ -145,6 +162,7 @@ pub(crate) struct QueueCursor {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct QueueScan {
     pub(crate) max_token: Option<u64>,
     pub(crate) after: Option<QueueCursor>,
@@ -153,6 +171,7 @@ pub(crate) struct QueueScan {
 }
 
 #[derive(Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct QueuePage<T> {
     pub(crate) entries: Vec<T>,
     pub(crate) next: Option<QueueCursor>,
@@ -163,6 +182,7 @@ pub(crate) struct QueuePage<T> {
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
+#[cfg(feature = "search")]
 pub(crate) struct OversizedEntry {
     pub(crate) id: QueueId,
     pub(crate) owed_from: u64,
@@ -194,12 +214,14 @@ pub(crate) struct SearchCoverage {
 }
 
 #[derive(Clone, Copy, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct RebuildRequest {
     pub(crate) index_id: [u8; 16],
     pub(crate) index_revision: u64,
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct RebuildScan {
     pub(crate) index_id: [u8; 16],
     pub(crate) row_limit: usize,
@@ -207,6 +229,7 @@ pub(crate) struct RebuildScan {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct RebuildPage {
     pub(crate) remaining: bool,
     pub(crate) rows: usize,
@@ -216,6 +239,7 @@ pub(crate) struct RebuildPage {
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[cfg(feature = "search")]
 pub(crate) struct ManifestDigest {
     pub(crate) count: u64,
     pub(crate) hash: [u8; 32],
@@ -223,6 +247,7 @@ pub(crate) struct ManifestDigest {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct StageFailure {
     pub(crate) index_id: [u8; 16],
     pub(crate) state: RetryState,
@@ -230,6 +255,7 @@ pub(crate) struct StageFailure {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct GraphScan {
     pub(crate) graph: TermId,
     pub(crate) after: Option<[u8; 64]>,
@@ -238,6 +264,7 @@ pub(crate) struct GraphScan {
 }
 
 #[derive(Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct QuadPage {
     pub(crate) entries: Vec<EncodedQuad>,
     pub(crate) rows: usize,
@@ -246,6 +273,7 @@ pub(crate) struct QuadPage {
 }
 
 #[derive(Clone, Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct SubjectScan {
     pub(crate) graph: TermId,
     pub(crate) subject: TermId,
@@ -255,6 +283,7 @@ pub(crate) struct SubjectScan {
 }
 
 #[derive(Debug)]
+#[cfg(feature = "search")]
 pub(crate) struct SubjectPage {
     pub(crate) entries: Vec<(EncodedTerm, EncodedTerm)>,
     pub(crate) next: Option<(TermId, TermId)>,
@@ -265,12 +294,17 @@ pub(crate) struct SubjectPage {
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[cfg(feature = "search")]
 pub(crate) struct OversizedSource {
     pub(crate) bytes: usize,
     pub(crate) limit: usize,
 }
 
 /// Bounds one drain pass over the durable FTS queues.
+#[cfg_attr(
+    not(feature = "search"),
+    allow(dead_code, reason = "the search stub ignores drain bounds")
+)]
 pub(crate) struct QueueBound {
     /// Maximum eligible entries one drain may hand back.
     pub(crate) chunk: usize,
@@ -301,6 +335,10 @@ impl DrainControl {
     }
 }
 
+#[cfg_attr(
+    not(feature = "search"),
+    allow(dead_code, reason = "the search stub ignores drain bounds")
+)]
 pub(crate) struct DrainRequest {
     pub(crate) bound: QueueBound,
     pub(crate) control: DrainControl,
@@ -315,6 +353,10 @@ pub(crate) struct DirtyTokens {
 
 /// What one drain pass covered and what it still owes.
 #[derive(Debug, Default)]
+#[cfg_attr(
+    not(feature = "search"),
+    allow(dead_code, reason = "the search stub ignores drain bounds")
+)]
 pub(crate) struct DrainProgress {
     /// Queue entries committed to the index and then acknowledged.
     pub(crate) covered: usize,
@@ -355,6 +397,11 @@ pub(crate) struct CleanupFailure {
 
 /// A queued subject: one dirty entry in the per-`(graph, subject)` queue.
 #[derive(Clone, Debug)]
+#[cfg(any(test, feature = "search"))]
+#[cfg_attr(
+    not(feature = "search"),
+    allow(dead_code, reason = "tests without search only count queued rows")
+)]
 pub(crate) struct DirtySubject {
     pub(crate) graph: GraphId,
     pub(crate) subject: TermId,
@@ -363,6 +410,11 @@ pub(crate) struct DirtySubject {
 
 /// A queued whole-graph entry: a reindex or a search-delete.
 #[derive(Clone, Debug)]
+#[cfg(any(test, feature = "search"))]
+#[cfg_attr(
+    not(feature = "search"),
+    allow(dead_code, reason = "tests without search only count queued rows")
+)]
 pub(crate) struct DirtyGraph {
     pub(crate) graph: GraphId,
     pub(crate) tokens: DirtyTokens,
diff --git a/src/internal/store.rs b/src/internal/store.rs
index 2b7b5e4..67054dd 100644
--- a/src/internal/store.rs
+++ b/src/internal/store.rs
@@ -27,13 +27,18 @@ use crate::core::{
 use crate::core::{CrateRenderHints as RenderHints, EventId};
 use crate::memory::{MemoryBudget, MemoryLease};
 use crate::qv_gate::QvCommitGate;
+#[cfg(feature = "search")]
 use crate::search::queue::{
-    CleanupJob, CleanupPage, CleanupScan, DeleteGeneration, DirtyGraph, DirtySubject, DirtyTokens,
-    GenerationId, GenerationRequest, GenerationSwitch, GraphGeneration, GraphScan, ManifestDigest,
-    ManifestPage, ManifestRow, ManifestScan, OversizedCleanup, OversizedEntry, OversizedSource,
-    QuadPage, QueueCursor, QueueId, QueueKind, QueuePage, QueueScan, RebuildPage, RebuildRequest,
-    RebuildScan, RetryState, SEARCH_META_FORMAT, SearchCoverage, StageFailure, StageJob,
-    StageRequest, SubjectPage, SubjectScan,
+    CleanupJob, CleanupPage, CleanupScan, DeleteGeneration, GenerationId, GenerationRequest,
+    GenerationSwitch, GraphGeneration, GraphScan, ManifestDigest, ManifestPage, ManifestRow,
+    ManifestScan, OversizedCleanup, OversizedEntry, OversizedSource, QuadPage, QueueId, QueuePage,
+    QueueScan, RebuildPage, RebuildRequest, RebuildScan, StageFailure, StageJob, StageRequest,
+    SubjectPage, SubjectScan,
+};
+#[cfg(any(test, feature = "search"))]
+use crate::search::queue::{DirtyGraph, DirtySubject};
+use crate::search::queue::{
+    DirtyTokens, QueueCursor, QueueKind, RetryState, SEARCH_META_FORMAT, SearchCoverage,
 };
 use crate::sync::{
     BackupProof, MutationId, MutationLookup, MutationReceipt, MutationStatus, RepairAudit,
@@ -204,17 +209,25 @@ const GRAPH_REINDEX_PREFIX: u8 = b'R';
 const GRAPH_DELETE_PREFIX: u8 = b'X';
 const SEARCH_ORDER_PREFIX: u8 = b'O';
 const SEARCH_FAILURE_PREFIX: u8 = b'F';
+#[cfg(feature = "search")]
 const SEARCH_GENERATION_PREFIX: u8 = b'G';
+#[cfg(feature = "search")]
 const SEARCH_STAGE_PREFIX: u8 = b'S';
+#[cfg(feature = "search")]
 const SEARCH_CLEANUP_PREFIX: u8 = b'L';
+#[cfg(feature = "search")]
 const SEARCH_NEXT_KEY: &[u8] = b"N";
 const SEARCH_COVERAGE_KEY: &[u8] = b"C";
 const SEARCH_HEAD_KEY: &[u8] = b"H";
 const SEARCH_SCHEMA_KEY: &[u8] = b"V";
+#[cfg(feature = "search")]
 const SEARCH_MANIFEST_KEY: &[u8] = b"M";
+#[cfg(feature = "search")]
 const SEARCH_REBUILD_KEY: &[u8] = b"E";
 const SEARCH_COVERAGE_MAGIC: [u8; 2] = *b"SC";
+#[cfg(feature = "search")]
 const SEARCH_GENERATION_MAGIC: [u8; 2] = *b"SG";
+#[cfg(feature = "search")]
 const SEARCH_STAGE_MAGIC: [u8; 2] = *b"SS";
 const SEARCH_ORDER_KEY: &[u8] = b"Q";
 const SEARCH_ORDER_FORMAT: u16 = 1;
@@ -325,6 +338,7 @@ struct CacheBudget {
     subjects: usize,
     objects: usize,
     planner: usize,
+    #[cfg(feature = "shacl-core")]
     shacl: usize,
 }
 
@@ -343,6 +357,7 @@ impl CacheBudget {
                 subjects: SUBJECT_CACHE_BYTES,
                 objects: ORDER_CACHE_BYTES,
                 planner: PLANNER_CACHE_BYTES,
+                #[cfg(feature = "shacl-core")]
                 shacl: 0,
             };
         };
@@ -356,6 +371,7 @@ impl CacheBudget {
             subjects: scaled_ceiling(SUBJECT_CACHE_BYTES, allowed, total),
             objects: scaled_ceiling(ORDER_CACHE_BYTES, allowed, total),
             planner: scaled_ceiling(PLANNER_CACHE_BYTES, allowed, total),
+            #[cfg(feature = "shacl-core")]
             shacl: 0,
         }
     }
@@ -377,6 +393,7 @@ impl CacheBudget {
             subjects: scaled_ceiling(SUBJECT_CACHE_BYTES, cache_allowed, total),
             objects: scaled_ceiling(ORDER_CACHE_BYTES, cache_allowed, total),
             planner: scaled_ceiling(PLANNER_CACHE_BYTES, cache_allowed, total),
+            #[cfg(feature = "shacl-core")]
             shacl,
         }
     }
@@ -930,6 +947,7 @@ struct AckedEntry {
     covered: u64,
 }
 
+#[cfg(feature = "search")]
 struct OrderedQueuePage {
     rows: Vec<(QueueCursor, DirtyTokens)>,
     next: Option<QueueCursor>,
@@ -956,6 +974,7 @@ struct SearchOrderMigration {
 }
 
 #[derive(serde::Serialize, serde::Deserialize)]
+#[cfg(feature = "search")]
 struct StoredGeneration {
     format: u16,
     index_id: [u8; 16],
@@ -963,6 +982,7 @@ struct StoredGeneration {
 }
 
 #[derive(serde::Serialize, serde::Deserialize)]
+#[cfg(feature = "search")]
 struct StoredStage {
     format: u16,
     index_id: [u8; 16],
@@ -978,6 +998,7 @@ struct LegacySearchCoverage {
 }
 
 #[derive(serde::Serialize, serde::Deserialize)]
+#[cfg(feature = "search")]
 struct StoredManifest {
     format: u16,
     index_id: [u8; 16],
@@ -987,6 +1008,7 @@ struct StoredManifest {
 }
 
 #[derive(serde::Serialize, serde::Deserialize)]
+#[cfg(feature = "search")]
 struct SearchRebuildScan {
     format: u16,
     index_id: [u8; 16],
@@ -995,6 +1017,7 @@ struct SearchRebuildScan {
     done: bool,
 }
 
+#[cfg(feature = "search")]
 struct ManifestChange<'a> {
     index_id: [u8; 16],
     graph: &'a GraphId,
@@ -1382,7 +1405,7 @@ pub struct GraphStore {
     policy_failure: Mutex<Option<GraphId>>,
     /// Set by a test to stall inside a held [`GraphStore::fts_queue_guard`],
     /// between an acknowledgement's token read and its commit.
-    #[cfg(test)]
+    #[cfg(all(test, feature = "search"))]
     fts_ack_stall: Mutex<Option<std::time::Duration>>,
     /// Set by a test to stall a rebuild between its durable scan and the
     /// install; `rebuild_stalled` publishes that the window has been entered.
@@ -1600,6 +1623,7 @@ fn queue_kind_tag(kind: QueueKind) -> u8 {
     }
 }
 
+#[cfg(feature = "search")]
 fn decode_queue_kind(tag: u8) -> Result<QueueKind> {
     match tag {
         0 => Ok(QueueKind::Delete),
@@ -1621,6 +1645,7 @@ fn search_order_key(cursor: QueueCursor) -> [u8; 42] {
     key
 }
 
+#[cfg(feature = "search")]
 fn decode_search_order(bytes: &[u8]) -> Result<QueueCursor> {
     if bytes.len() != 42 || bytes[0] != SEARCH_ORDER_PREFIX {
         return Err(StoreError::InvalidSearchState("queue-order-key-invalid"));
@@ -1646,6 +1671,7 @@ fn search_failure_key(cursor: QueueCursor) -> [u8; 34] {
     key
 }
 
+#[cfg(feature = "search")]
 fn search_generation_key(graph: &GraphId) -> [u8; 17] {
     let term = hash_term(&EncodedTerm::from_named_node(&graph.0));
     let mut key = [0u8; 17];
@@ -1654,12 +1680,14 @@ fn search_generation_key(graph: &GraphId) -> [u8; 17] {
     key
 }
 
+#[cfg(feature = "search")]
 fn search_stage_key(graph: &GraphId) -> [u8; 17] {
     let mut key = search_generation_key(graph);
     key[0] = SEARCH_STAGE_PREFIX;
     key
 }
 
+#[cfg(feature = "search")]
 fn search_cleanup_key(generation: GenerationId) -> [u8; 9] {
     let mut key = [0u8; 9];
     key[0] = SEARCH_CLEANUP_PREFIX;
@@ -1675,6 +1703,7 @@ fn encode_search_coverage(coverage: &SearchCoverage) -> Result<Vec<u8>> {
     Ok(bytes)
 }
 
+#[cfg(feature = "search")]
 fn encode_search_generation(stored: &StoredGeneration) -> Result<Vec<u8>> {
     let mut bytes = Vec::new();
     bytes.extend_from_slice(&SEARCH_GENERATION_MAGIC);
@@ -1684,6 +1713,7 @@ fn encode_search_generation(stored: &StoredGeneration) -> Result<Vec<u8>> {
     Ok(bytes)
 }
 
+#[cfg(feature = "search")]
 fn encode_search_stage(stored: &StoredStage) -> Result<Vec<u8>> {
     let mut bytes = Vec::new();
     bytes.extend_from_slice(&SEARCH_STAGE_MAGIC);
@@ -1693,6 +1723,7 @@ fn encode_search_stage(stored: &StoredStage) -> Result<Vec<u8>> {
     Ok(bytes)
 }
 
+#[cfg(feature = "search")]
 fn decode_search_generation(bytes: &[u8]) -> Result<StoredGeneration> {
     if !bytes.starts_with(&SEARCH_GENERATION_MAGIC) {
         return Ok(postcard::from_bytes(bytes)?);
@@ -1716,6 +1747,7 @@ fn decode_search_generation(bytes: &[u8]) -> Result<StoredGeneration> {
     })
 }
 
+#[cfg(feature = "search")]
 fn decode_search_stage(bytes: &[u8]) -> Result<StoredStage> {
     if !bytes.starts_with(&SEARCH_STAGE_MAGIC) {
         return Ok(postcard::from_bytes(bytes)?);
@@ -2499,12 +2531,14 @@ pub(crate) struct QvRead<'a> {
 }
 
 /// The graph and byte budget of one search orphan lookup.
+#[cfg(feature = "search")]
 pub(crate) struct OrphanScope {
     pub(crate) graph: TermId,
     pub(crate) byte_limit: usize,
 }
 
 #[derive(Clone)]
+#[cfg(feature = "search")]
 pub(crate) struct SearchSnapshot {
     snapshot: Snapshot,
     terms: Keyspace,
@@ -2513,6 +2547,7 @@ pub(crate) struct SearchSnapshot {
 }
 
 #[derive(Clone)]
+#[cfg(feature = "search")]
 pub(crate) struct SearchManifestSnapshot {
     snapshot: Snapshot,
     terms: Keyspace,
@@ -2521,6 +2556,7 @@ pub(crate) struct SearchManifestSnapshot {
     search_queue: Keyspace,
 }
 
+#[cfg(feature = "search")]
 impl SearchManifestSnapshot {
     pub(crate) fn digest(&self, index_id: [u8; 16]) -> Result<ManifestDigest> {
         let manifest = self
@@ -2558,6 +2594,7 @@ impl SearchManifestSnapshot {
     }
 }
 
+#[cfg(feature = "search")]
 impl SearchSnapshot {
     pub(crate) fn scan_graph(&self, scan: &GraphScan) -> Result<QuadPage> {
         scan_graph_snapshot(&self.snapshot, &self.quads, scan)
@@ -2708,6 +2745,7 @@ impl SearchSnapshot {
     }
 }
 
+#[cfg(feature = "search")]
 fn snapshot_term(snapshot: &Snapshot, terms: &Keyspace, id: TermId) -> Result<EncodedTerm> {
     let value = snapshot
         .get(terms, id.to_be_bytes())?
@@ -2728,6 +2766,7 @@ fn resume_range(prefix: &[u8], after: Option<&[u8]>) -> (Bound<Vec<u8>>, Bound<V
     (lower, Unbounded)
 }
 
+#[cfg(feature = "search")]
 fn scan_graph_snapshot(
     snapshot: &Snapshot,
     quads: &Keyspace,
@@ -3640,6 +3679,7 @@ impl GraphStore {
         StoreReadSnapshot::from(&self.db.snapshot())
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn search_snapshot(&self) -> SearchSnapshot {
         SearchSnapshot {
             snapshot: self.db.snapshot(),
@@ -3649,6 +3689,7 @@ impl GraphStore {
         }
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn search_manifest_snapshot(&self) -> Result<SearchManifestSnapshot> {
         let _queue = self.fts_queue_guard();
         Ok(SearchManifestSnapshot {
@@ -5818,7 +5859,7 @@ impl GraphStore {
     }
 
     /// Stall between an acknowledgement's token read and its commit. Test-only.
-    #[cfg(test)]
+    #[cfg(all(test, feature = "search"))]
     fn stall_search_ack(&self) {
         let stall = *self
             .fts_ack_stall
@@ -5830,7 +5871,7 @@ impl GraphStore {
     }
 
     /// Widen the acknowledgement's check-and-remove window. Test-only.
-    #[cfg(test)]
+    #[cfg(all(test, feature = "search"))]
     pub(crate) fn set_ack_stall(&self, delay: std::time::Duration) {
         *self
             .fts_ack_stall
@@ -6590,6 +6631,7 @@ impl GraphStore {
 
     /// Orphan ids at a search snapshot. Diagnostics commit after their source batch, so
     /// a snapshot between the two recomputes them instead of failing the subject.
+    #[cfg(feature = "search")]
     pub(crate) fn search_orphan_ids(
         &self,
         snapshot: &SearchSnapshot,
@@ -7632,7 +7674,7 @@ impl GraphStore {
             commit_failure: std::sync::atomic::AtomicBool::new(false),
             #[cfg(test)]
             policy_failure: Mutex::new(None),
-            #[cfg(test)]
+            #[cfg(all(test, feature = "search"))]
             fts_ack_stall: Mutex::new(None),
             #[cfg(test)]
             rebuild_stall: Mutex::new(None),
@@ -9781,6 +9823,7 @@ impl GraphStore {
         Ok(())
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn ensure_reindex(&self, graph: &GraphId) -> Result<(u64, bool)> {
         let graph_id = self
             .graph_id_for(graph)?
@@ -9885,6 +9928,7 @@ impl GraphStore {
         Ok(result)
     }
 
+    #[cfg(feature = "search")]
     fn ordered_queue(&self, kind: QueueKind, scan: &QueueScan) -> Result<OrderedQueuePage> {
         let after = scan.after.map(search_order_key);
         let mut page = OrderedQueuePage {
@@ -9943,6 +9987,7 @@ impl GraphStore {
         Ok(page)
     }
 
+    #[cfg(feature = "search")]
     fn queue_id(&self, cursor: QueueCursor) -> Result<QueueId> {
         let graph = self
             .decode_term(cursor.graph)?
@@ -9956,6 +10001,7 @@ impl GraphStore {
         })
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn scan_fts_subjects(&self, scan: &QueueScan) -> Result<QueuePage<DirtySubject>> {
         let page = self.ordered_queue(QueueKind::Subject, scan)?;
         let mut entries = Vec::with_capacity(page.rows.len());
@@ -9989,14 +10035,17 @@ impl GraphStore {
         })
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn scan_fts_reindexes(&self, scan: &QueueScan) -> Result<QueuePage<DirtyGraph>> {
         self.scan_fts_graphs(QueueKind::Reindex, scan)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn scan_fts_deletes(&self, scan: &QueueScan) -> Result<QueuePage<DirtyGraph>> {
         self.scan_fts_graphs(QueueKind::Delete, scan)
     }
 
+    #[cfg(feature = "search")]
     fn scan_fts_graphs(&self, kind: QueueKind, scan: &QueueScan) -> Result<QueuePage<DirtyGraph>> {
         let page = self.ordered_queue(kind, scan)?;
         let mut entries = Vec::with_capacity(page.rows.len());
@@ -10026,6 +10075,7 @@ impl GraphStore {
         })
     }
 
+    #[cfg(feature = "search")]
     fn queue_id_cursor(&self, id: &QueueId) -> Result<Option<QueueCursor>> {
         let Some(graph) = self.graph_id_for(&id.graph)? else {
             return Ok(None);
@@ -10038,6 +10088,7 @@ impl GraphStore {
         }))
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn fts_failure(&self, id: &QueueId) -> Result<Option<RetryState>> {
         let Some(cursor) = self.queue_id_cursor(id)? else {
             return Ok(None);
@@ -10048,6 +10099,7 @@ impl GraphStore {
             .transpose()
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn set_fts_failure(&self, state: &RetryState) -> Result<()> {
         let cursor = self
             .queue_id_cursor(&state.id)?
@@ -10063,6 +10115,7 @@ impl GraphStore {
         self.commit_fjall_batch(batch)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn clear_fts_failure(&self, id: &QueueId) -> Result<()> {
         let Some(cursor) = self.queue_id_cursor(id)? else {
             return Ok(());
@@ -10072,6 +10125,7 @@ impl GraphStore {
         self.commit_fjall_batch(batch)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn quarantine_search_failure(&self, failure: &StageFailure) -> Result<()> {
         if !matches!(failure.state.id.kind, QueueKind::Reindex)
             || failure.state.id.subject.is_some()
@@ -10142,6 +10196,7 @@ impl GraphStore {
         self.commit_fjall_batch(batch)
     }
 
+    #[cfg(feature = "search")]
     fn next_search_generation(&self, batch: &mut fjall::OwnedWriteBatch) -> Result<GenerationId> {
         let next = self
             .search_meta
@@ -10157,6 +10212,7 @@ impl GraphStore {
         Ok(GenerationId(next))
     }
 
+    #[cfg(feature = "search")]
     fn generation_hash(graph: &GraphId, generation: GenerationId) -> [u8; 32] {
         let mut hash = blake3::Hasher::new();
         hash.update(&(graph.as_str().len() as u64).to_be_bytes());
@@ -10165,6 +10221,7 @@ impl GraphStore {
         *hash.finalize().as_bytes()
     }
 
+    #[cfg(feature = "search")]
     fn stored_manifest(&self) -> Result<Option<StoredManifest>> {
         Ok(self
             .search_meta
@@ -10172,6 +10229,7 @@ impl GraphStore {
             .and_then(|value| postcard::from_bytes(value.as_ref()).ok()))
     }
 
+    #[cfg(feature = "search")]
     fn stage_manifest(
         &self,
         batch: &mut fjall::OwnedWriteBatch,
@@ -10235,6 +10293,7 @@ impl GraphStore {
         Ok(())
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn search_manifest_digest(&self, index_id: [u8; 16]) -> Result<ManifestDigest> {
         Ok(self
             .stored_manifest()?
@@ -10253,6 +10312,7 @@ impl GraphStore {
             }))
     }
 
+    #[cfg(feature = "search")]
     fn queue_search_cleanup(
         &self,
         batch: &mut fjall::OwnedWriteBatch,
@@ -10266,6 +10326,7 @@ impl GraphStore {
         Ok(())
     }
 
+    #[cfg(feature = "search")]
     fn stored_generation(&self, graph: &GraphId) -> Result<Option<StoredGeneration>> {
         let Some(value) = self.search_meta.get(search_generation_key(graph))? else {
             return Ok(None);
@@ -10283,6 +10344,7 @@ impl GraphStore {
         Ok(stored)
     }
 
+    #[cfg(feature = "search")]
     fn stored_search_stage(&self, graph: &GraphId) -> Result<Option<StoredStage>> {
         let Some(value) = self.search_meta.get(search_stage_key(graph))? else {
             return Ok(None);
@@ -10300,6 +10362,7 @@ impl GraphStore {
         Ok(stored)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn scan_search_manifest(
         &self,
         view: &SearchManifestSnapshot,
@@ -10443,6 +10506,7 @@ impl GraphStore {
         })
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn search_generation(&self, req: &GenerationRequest) -> Result<GraphGeneration> {
         Ok(self
             .stored_generation(&req.graph)?
@@ -10455,6 +10519,7 @@ impl GraphStore {
             }))
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn ensure_search_generation(
         &self,
         req: &GenerationRequest,
@@ -10523,6 +10588,7 @@ impl GraphStore {
     }
 
     /// Graph generations of every in-flight search stage of one index.
+    #[cfg(feature = "search")]
     pub(crate) fn staged_search_generations(
         &self,
         index_id: [u8; 16],
@@ -10541,6 +10607,7 @@ impl GraphStore {
         Ok(staged)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn search_stage(&self, req: &GenerationRequest) -> Result<Option<StageJob>> {
         Ok(self
             .stored_search_stage(&req.graph)?
@@ -10548,6 +10615,7 @@ impl GraphStore {
             .map(|stored| stored.job))
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn begin_search_stage(&self, req: &StageRequest) -> Result<StageJob> {
         let _queue = self.fts_queue_guard();
         if self.graph_tombstoned(&req.graph)? {
@@ -10595,6 +10663,7 @@ impl GraphStore {
         Ok(job)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn advance_search_stage(&self, job: &StageJob) -> Result<()> {
         let _queue = self.fts_queue_guard();
         let mut stored = self
@@ -10626,6 +10695,7 @@ impl GraphStore {
         self.commit_fjall_batch(batch)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn finish_search_stage(&self, job: &StageJob) -> Result<()> {
         let _queue = self.fts_queue_guard();
         let mut stored = self
@@ -10652,6 +10722,7 @@ impl GraphStore {
         self.commit_fjall_batch(batch)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn switch_search_stage(&self, job: &StageJob) -> Result<GenerationSwitch> {
         let _queue = self.fts_queue_guard();
         let stored = self
@@ -10726,6 +10797,7 @@ impl GraphStore {
         })
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn delete_search_graph(&self, req: &DeleteGeneration) -> Result<GenerationSwitch> {
         let _queue = self.fts_queue_guard();
         let current = self.stored_generation(&req.graph)?;
@@ -10782,6 +10854,7 @@ impl GraphStore {
         })
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn scan_search_cleanup(&self, scan: &CleanupScan) -> Result<CleanupPage> {
         let mut entries = Vec::with_capacity(scan.row_limit);
         let mut next = scan.after;
@@ -10846,6 +10919,7 @@ impl GraphStore {
         })
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn ack_search_cleanup(&self, jobs: &[CleanupJob]) -> Result<()> {
         let _queue = self.fts_queue_guard();
         let mut batch = self.buffered_batch();
@@ -10941,6 +11015,7 @@ impl GraphStore {
         Ok(rebuild)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn bind_search_rebuild(&self, req: &RebuildRequest) -> Result<u64> {
         let _queue = self.fts_queue_guard();
         if let Some(coverage) = self.search_coverage()?
@@ -10998,6 +11073,7 @@ impl GraphStore {
         Ok(rebuild)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn queue_search_rebuild(&self, scan: &RebuildScan) -> Result<RebuildPage> {
         let _queue = self.fts_queue_guard();
         let coverage = self
@@ -11102,6 +11178,7 @@ impl GraphStore {
         })
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn finish_search_rebuild(&self, coverage: &SearchCoverage) -> Result<()> {
         let _queue = self.fts_queue_guard();
         let current = self
@@ -11170,6 +11247,7 @@ impl GraphStore {
         self.commit_fjall_batch(batch)
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn advance_search_coverage(&self, coverage: &SearchCoverage) -> Result<()> {
         let _queue = self.fts_queue_guard();
         if coverage.format != SEARCH_META_FORMAT {
@@ -11211,6 +11289,7 @@ impl GraphStore {
     }
 
     /// Acknowledges covered subjects while preserving later redirty events.
+    #[cfg(feature = "search")]
     pub(crate) fn acknowledge_fts_queue(&self, queued: &[DirtySubject]) -> Result<()> {
         if queued.is_empty() {
             return Ok(());
@@ -11239,6 +11318,7 @@ impl GraphStore {
         Ok(())
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn acknowledge_reindex(&self, queued: &[DirtyGraph]) -> Result<()> {
         if queued.is_empty() {
             return Ok(());
@@ -11265,6 +11345,7 @@ impl GraphStore {
         Ok(())
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn acknowledge_deletes(&self, queued: &[DirtyGraph]) -> Result<()> {
         if queued.is_empty() {
             return Ok(());
@@ -11291,6 +11372,7 @@ impl GraphStore {
         Ok(())
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn acknowledge_reindexed(&self, queued: &[DirtyGraph]) -> Result<()> {
         if queued.is_empty() {
             return Ok(());
@@ -11325,6 +11407,7 @@ impl GraphStore {
         Ok(())
     }
 
+    #[cfg(feature = "search")]
     pub(crate) fn acknowledge_deleted(&self, queued: &[DirtyGraph]) -> Result<()> {
         if queued.is_empty() {
             return Ok(());
@@ -11981,6 +12064,7 @@ impl GraphStore {
 
     /// Oldest queued search work. Callers hold the queue lock; every queue insert lowers
     /// the floor to its token first, so the scan never starts past queued work.
+    #[cfg(feature = "search")]
     fn queue_head(&self) -> Result<Option<QueueCursor>> {
         let floor = self.queue_floor.load(Ordering::SeqCst);
         let mut start = [0u8; 9];
@@ -12474,6 +12558,7 @@ mod tests {
     /// Search indexing between a source commit and its diagnostics commit derives the
     /// orphans from its own snapshot instead of failing the subject.
     #[test]
+    #[cfg(feature = "search")]
     fn search_orphans_recomputed() {
         let dir = tempfile::tempdir().unwrap();
         let store = GraphStore::open(dir.path()).unwrap();
@@ -12502,6 +12587,7 @@ mod tests {
 
     /// A stale queue clear moving work below a checked head keeps coverage fenced.
     #[test]
+    #[cfg(feature = "search")]
     fn stale_clear_coverage() {
         let dir = tempfile::tempdir().unwrap();
         let store = GraphStore::open(dir.path()).unwrap();
@@ -13255,6 +13341,7 @@ mod tests {
     }
 
     #[test]
+    #[cfg(feature = "search")]
     fn search_generations_unique() {
         let (_dir, store) = setup_store();
         let first = GraphId::new("urn:test:search-generation-one");
@@ -13284,6 +13371,7 @@ mod tests {
     }
 
     #[test]
+    #[cfg(feature = "search")]
     fn stage_sessions_recover() {
         let (_dir, store) = setup_store();
         let graph = GraphId::new("urn:test:search-stage-session");
@@ -15699,6 +15787,7 @@ mod tests {
     }
 
     #[test]
+    #[cfg(feature = "search")]
     fn dirty_subjects_deduplicate() {
         let (_dir, store) = setup_store();
         let graph = GraphId::new("urn:test:graph");
@@ -15729,6 +15818,7 @@ mod tests {
 
     /// Redirtying preserves the oldest token promised to an earlier flush.
     #[test]
+    #[cfg(feature = "search")]
     fn enqueue_keeps_oldest() {
         let (_dir, store) = setup_store();
         let graph = GraphId::new("urn:test:graph");
@@ -15791,6 +15881,7 @@ mod tests {
 
     /// An enqueue concurrent with acknowledgement must retain its newer debt.
     #[test]
+    #[cfg(feature = "search")]
     fn acknowledgement_preserves_enqueue() {
         let (_dir, store) = setup_store();
         let store = Arc::new(store);
@@ -15827,6 +15918,7 @@ mod tests {
     }
 
     #[test]
+    #[cfg(feature = "search")]
     fn reindex_queue_roundtrips() {
         let (_dir, store) = setup_store();
         let graph = GraphId::new("urn:test:graph");
@@ -16275,6 +16367,7 @@ mod tests {
 
     /// A pre-restart reindex cannot acknowledge post-restart subject debt.
     #[test]
+    #[cfg(feature = "search")]
     fn tokens_survive_restart() {
         let dir = tempfile::tempdir().unwrap();
         let graph = GraphId::new("urn:test:fts-token-restart");
diff --git a/src/lib.rs b/src/lib.rs
index 0ce610b..d6a9ab2 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -778,6 +778,10 @@ const SEARCH_WAIT_POLL: Duration = Duration::from_millis(5);
 type SearchReply = std::result::Result<SearchReceipt, MaintenanceFailure>;
 
 #[derive(Clone, Debug)]
+#[cfg_attr(
+    not(feature = "search"),
+    allow(dead_code, reason = "flushes need search")
+)]
 struct MaintenanceFailure {
     kind: CraqleErrorKind,
     source: Arc<dyn std::error::Error + Send + Sync>,
@@ -805,6 +809,7 @@ impl MaintenanceFailure {
         }
     }
 
+    #[cfg(feature = "search")]
     fn into_error(self, target: u64) -> CraqleError {
         CraqleError::SearchMaintenance {
             target,
@@ -834,11 +839,13 @@ impl Drop for FlushPermit {
     }
 }
 
+#[cfg(feature = "search")]
 struct FlushWaiter {
     control: search::queue::DrainControl,
     permit: Arc<FlushPermit>,
 }
 
+#[cfg(feature = "search")]
 impl Drop for FlushWaiter {
     fn drop(&mut self) {
         self.control.cancel();
@@ -847,6 +854,10 @@ impl Drop for FlushWaiter {
 }
 
 #[derive(Debug)]
+#[cfg_attr(
+    not(feature = "search"),
+    allow(dead_code, reason = "flushes need search")
+)]
 struct FlushRequest {
     requested: u64,
     target: u64,
@@ -857,6 +868,10 @@ struct FlushRequest {
 }
 
 #[derive(Debug)]
+#[cfg_attr(
+    not(feature = "search"),
+    allow(dead_code, reason = "flushes need search")
+)]
 enum SearchWorkerMessage {
     Wake,
     Flush(FlushRequest),
@@ -877,6 +892,10 @@ impl SearchWorkerMessage {
     }
 }
 
+#[cfg_attr(
+    not(feature = "search"),
+    allow(dead_code, reason = "flushes need search")
+)]
 struct SearchUpdateWorker {
     sender: mpsc::SyncSender<SearchWorkerMessage>,
     store: Arc<GraphStore>,
@@ -1279,7 +1298,7 @@ fn panic_message(payload: &(dyn std::any::Any + Send)) -> String {
 }
 
 // Tests synchronously drain the same fixed coverage boundary as the worker.
-#[cfg(test)]
+#[cfg(all(test, feature = "search"))]
 fn flush_search_queue(store: &GraphStore, search: &SearchIndex) -> Result<()> {
     #[cfg(test)]
     if search.take_drain_panic() {
diff --git a/src/search_stub.rs b/src/search_stub.rs
index fd3f95c..c205639 100644
--- a/src/search_stub.rs
+++ b/src/search_stub.rs
@@ -8,7 +8,7 @@ pub(crate) mod queue;
 use std::path::Path;
 
 use crate::core::GraphId;
-pub(crate) use crate::search::queue::{DrainRequest, QueueBound};
+pub(crate) use crate::search::queue::DrainRequest;
 use crate::store::GraphStore;
 
 #[derive(Debug, thiserror::Error)]
@@ -63,21 +63,13 @@ pub(crate) struct FilterQuery<'a, E> {
     pub subject: Option<&'a str>,
     pub allows: &'a dyn Fn(&str) -> std::result::Result<bool, E>,
     pub check: &'a dyn Fn() -> std::result::Result<(), E>,
+    #[allow(dead_code, reason = "the disabled search ignores candidate filters")]
     pub candidates: Option<&'a CandidateSource<'a, E>>,
 }
 
 pub(crate) type CandidateSource<'a, E> =
     dyn Fn() -> std::result::Result<Option<Vec<String>>, E> + 'a;
 
-pub(crate) fn stable_hit_key(graph_id: &str, subject_iri: &str) -> [u8; 32] {
-    let mut hasher = blake3::Hasher::new();
-    hasher.update(&(graph_id.len() as u64).to_be_bytes());
-    hasher.update(graph_id.as_bytes());
-    hasher.update(&(subject_iri.len() as u64).to_be_bytes());
-    hasher.update(subject_iri.as_bytes());
-    *hasher.finalize().as_bytes()
-}
-
 #[derive(Debug, Default)]
 pub struct SearchIndex;
 
@@ -207,20 +199,7 @@ impl SearchIndex {
         Ok(())
     }
 
-    /// Retain coalesced queue debt for a later search-enabled build.
-    pub fn process_queued_updates(&self, store: &GraphStore, bound: QueueBound) -> Result<usize> {
-        Ok(self
-            .drain_queues(
-                store,
-                DrainRequest {
-                    bound,
-                    control: crate::search::queue::DrainControl::default(),
-                },
-            )?
-            .covered)
-    }
-
-    pub fn drain_queues(
+    pub(crate) fn drain_queues(
         &self,
         _store: &GraphStore,
         _request: DrainRequest,

From 6daf9c84a1719eb3de8ab5ea37dfec179fa98fc1 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 13:05:37 +0200
Subject: [PATCH 12/16] fix: read validation parent lookups through the query
 index

---
 src/internal/rdf_read.rs         | 34 ++++++++++++++++++++-
 src/internal/validation_delta.rs | 52 +++++++++++++++++++++++++++++---
 2 files changed, 81 insertions(+), 5 deletions(-)

diff --git a/src/internal/rdf_read.rs b/src/internal/rdf_read.rs
index 44faa80..3e9bec2 100644
--- a/src/internal/rdf_read.rs
+++ b/src/internal/rdf_read.rs
@@ -10,7 +10,7 @@ use std::sync::Arc;
 use crate::core::{EncodedTerm, GraphId};
 use crate::query::context::{GraphVisibility, QueryReadMode, ReadAccessPath, ReadContext};
 use crate::query::cursor::{
-    DenseCursor, DenseInput, DenseResolver, IndexScan, QueryCursor, RawIndexPattern,
+    DenseCursor, DenseInput, DenseResolver, IndexScan, QueryCursor, RawIndexPattern, RawQuadCursor,
 };
 use crate::store::{
     EncodedQuad, GraphStore, IndexCursorOrder, QueryIndexAdmission, QvRead, QvStat, Result,
@@ -473,6 +473,38 @@ impl<'store> StoreReadView<'store> {
         orphaned_for_graph(self.store, &self.snapshot, context, graph)
     }
 
+    /// Durable rows of a named-graph pattern. A lookup by predicate or object with an open
+    /// subject uses the trusted query index; the source order would scan the whole graph.
+    pub(crate) fn durable_cursor(
+        &self,
+        context: &ReadContext<'_>,
+        pattern: QuadPattern,
+    ) -> Result<RawQuadCursor> {
+        let indexed = pattern.subject.is_none()
+            && (pattern.predicate.is_some() || pattern.object.is_some())
+            && matches!(self.read_mode, QueryReadMode::Auto);
+        let Some(graph) = pattern.graph.filter(|_| indexed) else {
+            return Ok(self.snapshot.raw_quad_cursor(self.store, pattern));
+        };
+        if !self.qv_admission(context)?.trusted {
+            return Ok(self.snapshot.raw_quad_cursor(self.store, pattern));
+        }
+        let path = Self::auto_access_path(GraphSelector::Named(graph), pattern);
+        let costs = context.costs();
+        Ok(self
+            .snapshot
+            .query_index_cursor(
+                self.store,
+                &IndexScan {
+                    order: Self::qv_order(path),
+                    pattern,
+                    query_id_limit: None,
+                    costs: &costs,
+                },
+            )?
+            .track_costs(costs))
+    }
+
     fn qv_order(path: ReadAccessPath) -> IndexCursorOrder {
         match path {
             ReadAccessPath::QvGspo => IndexCursorOrder::Gspo,
diff --git a/src/internal/validation_delta.rs b/src/internal/validation_delta.rs
index f4e5096..7e1cae3 100644
--- a/src/internal/validation_delta.rs
+++ b/src/internal/validation_delta.rs
@@ -459,10 +459,7 @@ impl RdfReadView for DeltaReadView<'_, '_> {
         if !self.index.is_empty() {
             context.increment_index_seeks();
         }
-        let base = self
-            .base
-            .snapshot()
-            .raw_quad_cursor(self.base.store(), pattern);
+        let base = self.base.durable_cursor(context, pattern)?;
         Ok(QueryCursor::delta(
             self.base.store(),
             self.base.snapshot(),
@@ -1171,6 +1168,53 @@ mod tests {
         assert!(cursor.next().is_none());
     }
 
+    /// Parent lookups during validation read the query index instead of scanning the graph.
+    #[test]
+    fn inverse_walks_indexed() {
+        let (_directory, store) = setup_store();
+        let graph = GraphId::new("urn:test:indexed-walks");
+        for index in 0..256 {
+            add_quad(
+                &store,
+                &graph,
+                &format!("urn:test:subject-{index}"),
+                "urn:test:name",
+                &format!("urn:test:value-{index}"),
+            );
+        }
+        add_quad(
+            &store,
+            &graph,
+            "urn:test:root",
+            "urn:test:hasPart",
+            "urn:test:child",
+        );
+        store.rebuild_query_indexes().unwrap();
+        let index = DeltaIndex::build(&store, &graph, &[]).unwrap();
+        let view = DeltaReadView::new(StoreReadView::new(&store), &index);
+        let context = ReadContext::for_validation(QueryCancellation::new(), &graph);
+        let (root, predicate, child) = ids(
+            &view,
+            &context,
+            "urn:test:root",
+            "urn:test:hasPart",
+            "urn:test:child",
+        );
+        let parents: Vec<_> = view
+            .inverse_predicate(
+                &context,
+                GraphSelector::Named(view.graph()),
+                predicate,
+                child,
+            )
+            .unwrap()
+            .collect::<Result<_>>()
+            .unwrap();
+        assert_eq!(1, parents.len());
+        assert_eq!(root, parents[0].subject);
+        assert!(context.snapshot().candidate_quads < 8);
+    }
+
     #[test]
     fn walks_include_overlay() {
         let (_directory, store) = setup_store();

From bebaa11f3a7e36dbb2b3656d8f6823f66d08d3c2 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 13:05:37 +0200
Subject: [PATCH 13/16] test: profile single entity patches in the write
 latency profile

---
 tests/perf/write_latency.rs | 22 ++++++++++++++++++++++
 1 file changed, 22 insertions(+)

diff --git a/tests/perf/write_latency.rs b/tests/perf/write_latency.rs
index 4916c14..21367ec 100644
--- a/tests/perf/write_latency.rs
+++ b/tests/perf/write_latency.rs
@@ -157,6 +157,28 @@ mod tests {
                 apply(graph, sample, jsonld);
             }),
         );
+        report(
+            "patch entity",
+            timed(samples, |sample| {
+                let request = PatchEntityRequest {
+                    entity: CreateEntityRequest {
+                        graph: documents[sample].0.clone(),
+                        entity_id: format!("./data/patched-{sample}.txt"),
+                        entity_type: "File".to_string(),
+                        name: format!("Patched file {sample}"),
+                        additional_triples: Vec::new(),
+                    },
+                    replaced_predicates: Vec::new(),
+                };
+                node.patch_data_with(
+                    &AllowAllAuthorizer,
+                    request,
+                    CraqleRequestDurability::WalAlreadyDurable,
+                    None,
+                )
+                .unwrap();
+            }),
+        );
         report(
             "set policy",
             timed(samples, |sample| {

From d877dd44bd0280fedff3e50610842a6d651b7d17 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 14:22:22 +0200
Subject: [PATCH 14/16] fix: bound the search orphan fallback by bytes and
 cancellation

---
 src/internal/search/mod.rs | 10 ++++
 src/internal/store.rs      | 99 ++++++++++++++++++++++++++++++++------
 2 files changed, 95 insertions(+), 14 deletions(-)

diff --git a/src/internal/search/mod.rs b/src/internal/search/mod.rs
index 2a54689..cd98069 100644
--- a/src/internal/search/mod.rs
+++ b/src/internal/search/mod.rs
@@ -712,6 +712,7 @@ struct PrepareSubject<'a> {
     subject: TermId,
     byte_limit: usize,
     generation: GenerationId,
+    control: &'a DrainControl,
 }
 
 struct OrphanInput<'a> {
@@ -719,6 +720,7 @@ struct OrphanInput<'a> {
     graph: &'a GraphId,
     graph_tid: TermId,
     byte_limit: usize,
+    control: &'a DrainControl,
 }
 
 struct StoreSyncCaches {
@@ -744,9 +746,11 @@ impl StoreSyncCaches {
             let scope = crate::store::OrphanScope {
                 graph: input.graph_tid,
                 byte_limit: artifact_limit,
+                control: input.control,
             };
             let orphaned = match input.store.search_orphan_ids(&self.snapshot, scope) {
                 Ok(orphaned) => orphaned,
+                Err(crate::store::StoreError::Cancelled) => return Err(SearchError::Cancelled),
                 Err(crate::store::StoreError::LimitExceeded {
                     resource: "search diagnostics rows",
                     limit,
@@ -2855,6 +2859,7 @@ impl SearchIndex {
                     subject: entry.subject,
                     byte_limit,
                     generation,
+                    control: &pass.control,
                 },
             ) {
                 Ok(op) => {
@@ -3604,6 +3609,7 @@ fn prepare_subject_op(
             graph: req.graph,
             graph_tid,
             byte_limit: text_limit,
+            control: req.control,
         })?
         .contains(&req.subject);
     if hidden {
@@ -4073,6 +4079,7 @@ mod tests {
                 subject: graph_tid,
                 byte_limit: node.search.work_bytes(),
                 generation: DIRECT_GENERATION,
+                control: &DrainControl::default(),
             },
             &mut caches,
         )
@@ -4126,6 +4133,7 @@ mod tests {
                 subject: graph_tid,
                 byte_limit: node.search.work_bytes(),
                 generation: DIRECT_GENERATION,
+                control: &DrainControl::default(),
             },
             &mut caches,
         );
@@ -4186,6 +4194,7 @@ mod tests {
                 subject,
                 byte_limit: node.search.work_bytes(),
                 generation: DIRECT_GENERATION,
+                control: &DrainControl::default(),
             },
             &mut caches,
         )
@@ -4205,6 +4214,7 @@ mod tests {
                 subject: absent,
                 byte_limit: node.search.work_bytes(),
                 generation: DIRECT_GENERATION,
+                control: &DrainControl::default(),
             },
             &mut caches,
         )
diff --git a/src/internal/store.rs b/src/internal/store.rs
index 67054dd..fa43a18 100644
--- a/src/internal/store.rs
+++ b/src/internal/store.rs
@@ -2530,11 +2530,12 @@ pub(crate) struct QvRead<'a> {
     pub(crate) costs: &'a crate::query::context::QueryCost,
 }
 
-/// The graph and byte budget of one search orphan lookup.
+/// The graph, byte budget and drain control of one search orphan lookup.
 #[cfg(feature = "search")]
-pub(crate) struct OrphanScope {
+pub(crate) struct OrphanScope<'a> {
     pub(crate) graph: TermId,
     pub(crate) byte_limit: usize,
+    pub(crate) control: &'a crate::search::queue::DrainControl,
 }
 
 #[derive(Clone)]
@@ -6635,21 +6636,75 @@ impl GraphStore {
     pub(crate) fn search_orphan_ids(
         &self,
         snapshot: &SearchSnapshot,
-        scope: OrphanScope,
+        scope: OrphanScope<'_>,
     ) -> Result<HashSet<TermId>> {
         match snapshot.orphaned_ids(scope.graph, scope.byte_limit) {
-            Err(StoreError::InvalidSearchState(
-                "search-diagnostics-stale" | "search-diagnostics-missing",
-            )) => self.snapshot_orphan_ids(
-                &snapshot.snapshot,
-                &crate::query::context::ReadContext::default(),
-                scope.graph,
-                &self.orphan_vocab()?,
-            ),
+            Err(
+                error @ StoreError::InvalidSearchState(
+                    "search-diagnostics-stale" | "search-diagnostics-missing",
+                ),
+            ) => self.bounded_orphan_ids(snapshot, &scope)?.ok_or(error),
             result => result,
         }
     }
 
+    /// Recomputes orphan ids within the scope's byte budget and drain control.
+    /// `None` means the graph did not fit, so the caller keeps its retryable error.
+    #[cfg(feature = "search")]
+    fn bounded_orphan_ids(
+        &self,
+        snapshot: &SearchSnapshot,
+        scope: &OrphanScope<'_>,
+    ) -> Result<Option<HashSet<TermId>>> {
+        let vocab = self.orphan_vocab()?;
+        let mut data_entities = HashSet::new();
+        let mut adjacency = HashMap::<TermId, Vec<TermId>>::new();
+        let mut bytes = 0usize;
+        for guard in snapshot
+            .snapshot
+            .prefix(&self.quads, scope.graph.to_be_bytes())
+        {
+            if scope.control.is_cancelled() {
+                return Err(StoreError::Cancelled);
+            }
+            let (key, value) = guard.into_inner()?;
+            bytes = bytes.saturating_add(key.len() + value.len());
+            if bytes > scope.byte_limit {
+                return Ok(None);
+            }
+            if dots_empty(value.as_ref()) {
+                continue;
+            }
+            let quad = Self::decode_quad_key(key.as_ref())?;
+            if vocab.has_part == Some(quad.predicate) {
+                adjacency.entry(quad.subject).or_default().push(quad.object);
+                if quad.subject != scope.graph {
+                    data_entities.insert(quad.subject);
+                }
+                if quad.object != scope.graph {
+                    data_entities.insert(quad.object);
+                }
+            }
+            if vocab.rdf_type == Some(quad.predicate)
+                && quad.subject != scope.graph
+                && vocab.data_types.contains(&Some(quad.object))
+            {
+                data_entities.insert(quad.subject);
+            }
+        }
+        let mut reachable = HashSet::from([scope.graph]);
+        let mut queue = VecDeque::from([scope.graph]);
+        while let Some(current) = queue.pop_front() {
+            for &neighbor in adjacency.get(&current).into_iter().flatten() {
+                if reachable.insert(neighbor) {
+                    queue.push_back(neighbor);
+                }
+            }
+        }
+        data_entities.retain(|entity| !reachable.contains(entity));
+        Ok(Some(data_entities))
+    }
+
     /// Resolves the vocabulary ids used by orphan detection.
     fn orphan_vocab(&self) -> Result<OrphanVocab> {
         let id = |named_node: oxrdf::NamedNode| {
@@ -12564,6 +12619,11 @@ mod tests {
         let store = GraphStore::open(dir.path()).unwrap();
         let graph = GraphId::new("urn:test:search-orphans");
         store.create_graph(&graph).unwrap();
+        commit_add(
+            &store,
+            &graph,
+            encode_quad(&store, &graph, ("urn:s", "urn:p", "urn:o")),
+        );
         let graph_tid = store
             .lookup_term(&EncodedTerm::from_named_node(&graph.0))
             .unwrap()
@@ -12573,16 +12633,27 @@ mod tests {
             snapshot.orphaned_ids(graph_tid, usize::MAX),
             Err(StoreError::InvalidSearchState(_))
         ));
-        let scope = OrphanScope {
+        let control = crate::search::queue::DrainControl::default();
+        let scope = |byte_limit| OrphanScope {
             graph: graph_tid,
-            byte_limit: usize::MAX,
+            byte_limit,
+            control: &control,
         };
         assert!(
             store
-                .search_orphan_ids(&snapshot, scope)
+                .search_orphan_ids(&snapshot, scope(usize::MAX))
                 .unwrap()
                 .is_empty()
         );
+        assert!(matches!(
+            store.search_orphan_ids(&snapshot, scope(1)),
+            Err(StoreError::InvalidSearchState(_))
+        ));
+        control.cancel();
+        assert!(matches!(
+            store.search_orphan_ids(&snapshot, scope(usize::MAX)),
+            Err(StoreError::Cancelled)
+        ));
     }
 
     /// A stale queue clear moving work below a checked head keeps coverage fenced.

From f35002813374b93cf837b6fc6ef3edbb6dc5f303 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 14:22:22 +0200
Subject: [PATCH 15/16] fix: remove a needless return in the disabled search
 flush

---
 src/lib.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/lib.rs b/src/lib.rs
index d6a9ab2..5e6d7a6 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -981,7 +981,7 @@ impl SearchUpdateWorker {
         #[cfg(not(feature = "search"))]
         {
             let _ = options;
-            return Err(search::SearchError::Disabled.into());
+            Err(search::SearchError::Disabled.into())
         }
         #[cfg(feature = "search")]
         {

From bf81fd34a8aa9edefa546c16cb4e9205095714a8 Mon Sep 17 00:00:00 2001
From: Sebastian Beyvers <sb@pus.de>
Date: Thu, 24 Sep 2026 15:04:37 +0200
Subject: [PATCH 16/16] fix: bound the orphan fallback by held rows, not
 scanned bytes

---
 src/internal/search/mod.rs |  1 +
 src/internal/store.rs      | 34 +++++++++++++++++++---------------
 2 files changed, 20 insertions(+), 15 deletions(-)

diff --git a/src/internal/search/mod.rs b/src/internal/search/mod.rs
index cd98069..549916c 100644
--- a/src/internal/search/mod.rs
+++ b/src/internal/search/mod.rs
@@ -746,6 +746,7 @@ impl StoreSyncCaches {
             let scope = crate::store::OrphanScope {
                 graph: input.graph_tid,
                 byte_limit: artifact_limit,
+                work_limit: input.byte_limit,
                 control: input.control,
             };
             let orphaned = match input.store.search_orphan_ids(&self.snapshot, scope) {
diff --git a/src/internal/store.rs b/src/internal/store.rs
index fa43a18..5d64439 100644
--- a/src/internal/store.rs
+++ b/src/internal/store.rs
@@ -2530,11 +2530,13 @@ pub(crate) struct QvRead<'a> {
     pub(crate) costs: &'a crate::query::context::QueryCost,
 }
 
-/// The graph, byte budget and drain control of one search orphan lookup.
+/// The graph, byte budgets and drain control of one search orphan lookup.
 #[cfg(feature = "search")]
 pub(crate) struct OrphanScope<'a> {
     pub(crate) graph: TermId,
     pub(crate) byte_limit: usize,
+    /// Bytes a stale-diagnostics recompute may hold while it scans the graph.
+    pub(crate) work_limit: usize,
     pub(crate) control: &'a crate::search::queue::DrainControl,
 }
 
@@ -6648,8 +6650,8 @@ impl GraphStore {
         }
     }
 
-    /// Recomputes orphan ids within the scope's byte budget and drain control.
-    /// `None` means the graph did not fit, so the caller keeps its retryable error.
+    /// Recomputes orphan ids within the scope's work budget and drain control.
+    /// `None` means the held rows did not fit, so the caller keeps its retryable error.
     #[cfg(feature = "search")]
     fn bounded_orphan_ids(
         &self,
@@ -6659,7 +6661,7 @@ impl GraphStore {
         let vocab = self.orphan_vocab()?;
         let mut data_entities = HashSet::new();
         let mut adjacency = HashMap::<TermId, Vec<TermId>>::new();
-        let mut bytes = 0usize;
+        let mut held = 0usize;
         for guard in snapshot
             .snapshot
             .prefix(&self.quads, scope.graph.to_be_bytes())
@@ -6668,14 +6670,18 @@ impl GraphStore {
                 return Err(StoreError::Cancelled);
             }
             let (key, value) = guard.into_inner()?;
-            bytes = bytes.saturating_add(key.len() + value.len());
-            if bytes > scope.byte_limit {
-                return Ok(None);
-            }
             if dots_empty(value.as_ref()) {
                 continue;
             }
             let quad = Self::decode_quad_key(key.as_ref())?;
+            let kept =
+                vocab.has_part == Some(quad.predicate) || vocab.rdf_type == Some(quad.predicate);
+            if kept {
+                held = held.saturating_add(4 * std::mem::size_of::<TermId>());
+                if held > scope.work_limit {
+                    return Ok(None);
+                }
+            }
             if vocab.has_part == Some(quad.predicate) {
                 adjacency.entry(quad.subject).or_default().push(quad.object);
                 if quad.subject != scope.graph {
@@ -12619,11 +12625,8 @@ mod tests {
         let store = GraphStore::open(dir.path()).unwrap();
         let graph = GraphId::new("urn:test:search-orphans");
         store.create_graph(&graph).unwrap();
-        commit_add(
-            &store,
-            &graph,
-            encode_quad(&store, &graph, ("urn:s", "urn:p", "urn:o")),
-        );
+        let part = (graph.as_str(), "http://schema.org/hasPart", "urn:o");
+        commit_add(&store, &graph, encode_quad(&store, &graph, part));
         let graph_tid = store
             .lookup_term(&EncodedTerm::from_named_node(&graph.0))
             .unwrap()
@@ -12634,9 +12637,10 @@ mod tests {
             Err(StoreError::InvalidSearchState(_))
         ));
         let control = crate::search::queue::DrainControl::default();
-        let scope = |byte_limit| OrphanScope {
+        let scope = |work_limit| OrphanScope {
             graph: graph_tid,
-            byte_limit,
+            byte_limit: usize::MAX,
+            work_limit,
             control: &control,
         };
         assert!(