From c7d1fde8a29569f7e9012ba97de2ff3456399463 Mon Sep 17 00:00:00 2001
From: xav-db <xav@helix-db.com>
Date: Sun, 4 Oct 2026 13:47:29 +0100
Subject: [PATCH 1/9] fix(db): never plan a vector node as its own HNSW
 neighbor

A link that released versions left without a reverse locator outlives
its target's delete, so re-embedding that node searched back to it
through the link and selected it as its own neighbor, failing planning
with ContainsOwner on every attempt. Both mutation searches and the
beam's entry resolution now skip the inserting node.

A neighbor row that already links its own node failed every mutation
that loaded it. The deployed-row adapter now drops the self-link in
memory; a mutation that changes the row stores it self-free. No codec
or stored format changes.
---
 .../vector/hnsw/mutation/locator_tests.rs     | 269 +++++++++++++++++-
 .../db/src/search/vector/hnsw/mutation/mod.rs |  32 ++-
 .../db/src/search/vector/hnsw/neighbor_set.rs |  36 ++-
 .../production_support/vector/mutation.rs     |  45 ++-
 4 files changed, 351 insertions(+), 31 deletions(-)

diff --git a/crates/db/src/search/vector/hnsw/mutation/locator_tests.rs b/crates/db/src/search/vector/hnsw/mutation/locator_tests.rs
index 821b5fc3c..bdc1f5c5a 100644
--- a/crates/db/src/search/vector/hnsw/mutation/locator_tests.rs
+++ b/crates/db/src/search/vector/hnsw/mutation/locator_tests.rs
@@ -16,6 +16,11 @@
 //! [`links_released_versions_left_without_locators_outlive_their_target`]
 //! starts from links committed without a locator, as released versions
 //! write them, and pins that no operation here repairs them.
+//!
+//! [`a_committed_self_link_never_fails_its_node`] and
+//! [`damaged_graphs_never_gain_a_self_link`] start from rows that link their
+//! own node, as damage from released versions can leave them, and pin that
+//! every operation still plans and none links a node to itself.
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
@@ -515,6 +520,43 @@ async fn one_off_caches_at_their_bound_keep_one_locator_per_link() {
     );
 }
 
+/// Every link from a node to itself.
+fn self_links(graph: &Graph) -> BTreeSet<Link> {
+    graph
+        .links
+        .iter()
+        .filter(|(_, source, target)| source == target)
+        .copied()
+        .collect()
+}
+
+/// Commits `node`'s row at `layer` in `graph` linking `node` too, as damage
+/// from released versions can leave a row: a self-link without a locator.
+async fn commit_self_link<D: Distance>(
+    db: &slatedb::Db,
+    index: &VectorIndex<D>,
+    graph: &Graph,
+    layer: u16,
+    node: NodeId,
+) {
+    let neighbors = graph
+        .links
+        .iter()
+        .filter(|(row_layer, source, _)| (*row_layer, *source) == (layer, node))
+        .map(|(_, _, target)| *target)
+        .chain([node])
+        .collect::<Vec<_>>();
+    let txn = db.begin(IsolationLevel::Snapshot).await.unwrap();
+    let measured = MeasuredVectorTransaction::new(&txn);
+    let rows = VectorWriteRows::new(&measured, index.row_keyspace());
+    match layer {
+        0 => rows.put_layer0_neighbors(node, &neighbors),
+        layer => rows.put_upper_neighbors(layer, node, &neighbors),
+    }
+    .unwrap();
+    txn.commit().await.unwrap();
+}
+
 /// Commits `links` without their locators, as released versions left them.
 async fn commit_without_locators<D: Distance>(
     db: &slatedb::Db,
@@ -647,15 +689,15 @@ async fn released_unlocated_run<D: Distance>() {
     commit_without_locators(&db, &index, &released).await;
     let unlocated = unlocated.union(&released).copied().collect::<BTreeSet<_>>();
     // The re-embedding's delete misses each one-way source, so its insert
-    // searches through one back to the node and links the node to itself.
-    let error = re_embed(&db, &index, &moved, &mut points, target)
+    // searches through one back to the node, which it skips.
+    re_embed(&db, &index, &moved, &mut points, target)
         .await
-        .unwrap_err();
-    assert!(
-        error
-            .to_string()
-            .contains(&format!("neighbor set contains its owner {target}")),
-        "{name}: re-embedding {target}: {error}"
+        .unwrap_or_else(|error| panic!("{name}: re-embedding {target}: {error}"));
+    let re_embedded = Graph::read(&index, db.snapshot().await.unwrap().as_ref()).await;
+    assert_eq!(
+        self_links(&re_embedded),
+        BTreeSet::new(),
+        "{name}: re-embedded {target}"
     );
 
     let mut session = VectorBuildSession::<D>::new(NonZeroU64::new(1 << 20).unwrap());
@@ -716,12 +758,215 @@ async fn released_unlocated_run<D: Distance>() {
 /// value changed, so each link the insert restored kept no locator. A delete
 /// finds a link's source only through that locator or the node's own rows, so
 /// nothing here repairs such a link. Once the node stops linking back, nothing
-/// reaches the link: re-embedding the node fails, as its insert searches
-/// through the link back to the node and links the node to itself, and deleting
-/// the node leaves the link naming a node without an item, which fails searches
-/// that reach it. Every other link and locator stays exact.
+/// reaches the link: re-embedding the node searches through the link back to
+/// the node, which the insert skips rather than linking the node to itself,
+/// and deleting the node leaves the link naming a node without an item, which
+/// fails searches that reach it. Every other link and locator stays exact.
 #[tokio::test]
 async fn links_released_versions_left_without_locators_outlive_their_target() {
     released_unlocated_run::<Euclidean>().await;
     released_unlocated_run::<Cosine>().await;
 }
+
+/// Self-links in every row of a node with upper-layer rows, then an insert
+/// that searches through it, a re-embedding of it, and its delete.
+async fn self_linked_node_run<D: Distance>() {
+    let mut rng = StdRng::seed_from_u64(1);
+    let name = format!("self-linked-node-{}", D::name());
+    let (db, index) = create::<D>(&name).await;
+    let mut points = build(&db, &index, &mut rng, 60).await;
+    let built = Graph::read(&index, db.snapshot().await.unwrap().as_ref()).await;
+    let node = built
+        .rows
+        .iter()
+        .rev()
+        .map(|(_, node)| *node)
+        .next()
+        .unwrap();
+    let damaged = built
+        .rows
+        .iter()
+        .filter(|(_, row)| *row == node)
+        .map(|(layer, row)| (*layer, *row, *row))
+        .collect::<BTreeSet<_>>();
+    assert!(damaged.len() > 1, "{name}: {node} has upper-layer rows");
+    for (layer, _, _) in &damaged {
+        commit_self_link(&db, &index, &built, *layer, node).await;
+    }
+    let graph = Graph::read(&index, db.snapshot().await.unwrap().as_ref()).await;
+    assert_eq!(self_links(&graph), damaged, "{name}: damage committed");
+    let live = points.keys().copied().collect::<BTreeSet<_>>();
+    graph.assert_faults(&live, &damaged, &BTreeSet::new(), &name);
+
+    // An insert at the node's own point loads every one of its rows.
+    let txn = db.begin(IsolationLevel::Snapshot).await.unwrap();
+    let measured = MeasuredVectorTransaction::new(&txn);
+    let layer = damaged.iter().map(|(layer, _, _)| *layer).max().unwrap();
+    index
+        .stage_upsert_at_layer(&measured, 61, &points[&node], layer)
+        .await
+        .unwrap_or_else(|error| panic!("{name}: inserting beside {node}: {error}"));
+    txn.commit().await.unwrap();
+    points.insert(61, points[&node]);
+    let inserted = Graph::read(&index, db.snapshot().await.unwrap().as_ref()).await;
+    assert!(
+        self_links(&inserted).is_subset(&damaged),
+        "{name}: the insert links no node to itself"
+    );
+
+    re_embed(&db, &index, &inserted, &mut points, node)
+        .await
+        .unwrap_or_else(|error| panic!("{name}: re-embedding {node}: {error}"));
+    let moved = Graph::read(&index, db.snapshot().await.unwrap().as_ref()).await;
+    // A row the re-embedding stages unchanged keeps its stored self-link.
+    let kept = self_links(&moved);
+    assert!(kept.is_subset(&damaged), "{name}: re-embedded {node}");
+    let live = points.keys().copied().collect::<BTreeSet<_>>();
+    moved.assert_faults(&live, &kept, &BTreeSet::new(), &name);
+
+    let txn = db.begin(IsolationLevel::Snapshot).await.unwrap();
+    let measured = MeasuredVectorTransaction::new(&txn);
+    index
+        .stage_delete(&measured, node)
+        .await
+        .unwrap_or_else(|error| panic!("{name}: deleting {node}: {error}"));
+    txn.commit().await.unwrap();
+    points.remove(&node);
+    check(
+        &db,
+        &index,
+        &points,
+        &mut rng,
+        &format!("{name}: deleted {node}"),
+    )
+    .await;
+}
+
+/// A node whose rows link it, as damage from released versions can leave
+/// them, still plans: an insert searching through it, its re-embedding, and
+/// its delete each succeed, none links a node to itself, and the delete
+/// removes every self-linked row with the node.
+#[tokio::test]
+async fn a_committed_self_link_never_fails_its_node() {
+    self_linked_node_run::<Euclidean>().await;
+    self_linked_node_run::<Cosine>().await;
+}
+
+/// Before each operation, commits a self-link into a random row and drops
+/// the locators of every link to a random live node, as released versions
+/// can leave them, then deletes or moves a node through `planner`, mostly
+/// one of the two just damaged. A deleted node an operation picks is
+/// inserted again instead. Every operation plans, and the self-links after
+/// it are among those before it.
+async fn damaged_run<D: Distance>(seed: u64, planner: Planner) {
+    const NODES: NodeId = 100;
+    const OPS: usize = 40;
+    let mut rng = StdRng::seed_from_u64(seed);
+    let name = format!("self-link-damage-{}-{planner:?}-{seed}", D::name());
+    let (db, index) = create::<D>(&name).await;
+    let mut points = build(&db, &index, &mut rng, NODES).await;
+    let mut session = VectorBuildSession::<D>::new(NonZeroU64::new(1 << 20).unwrap());
+    for op in 0..OPS {
+        let graph = Graph::read(&index, db.snapshot().await.unwrap().as_ref()).await;
+        let rows = graph.rows.iter().copied().collect::<Vec<_>>();
+        let (layer, self_linked) = rows[rng.random_range(0..rows.len())];
+        commit_self_link(&db, &index, &graph, layer, self_linked).await;
+        let live = points.keys().copied().collect::<Vec<_>>();
+        let unlocated = live[rng.random_range(0..live.len())];
+        commit_without_locators(
+            &db,
+            &index,
+            &graph
+                .links
+                .iter()
+                .filter(|(_, _, target)| *target == unlocated)
+                .copied()
+                .collect(),
+        )
+        .await;
+        let before = self_links(&Graph::read(&index, db.snapshot().await.unwrap().as_ref()).await);
+
+        let node = match rng.random_range(0..4_u8) {
+            0 => self_linked,
+            1 | 2 => unlocated,
+            _ => rng.random_range(1..=NODES),
+        };
+        let current = points.get(&node).copied();
+        let delete = current.is_some() && rng.random_range(0..3_u8) == 0;
+        let point = loop {
+            let point = random_point(&mut rng);
+            if Some(point) != current {
+                break point;
+            }
+        };
+        let layer = random_layer(&mut rng);
+        let context = format!(
+            "{name}: op {op} ({} {node}, self-link at ({layer}, {self_linked}), \
+             unlocated links to {unlocated})",
+            if delete { "delete" } else { "upsert" }
+        );
+        let txn = db.begin(IsolationLevel::Snapshot).await.unwrap();
+        let measured = MeasuredVectorTransaction::new(&txn);
+        match (planner, delete) {
+            (Planner::Session, true) => {
+                index
+                    .stage_delete_with_build_session(&measured, node, &mut session)
+                    .await
+            }
+            (Planner::Session, false) => {
+                index
+                    .stage_upsert_at_layer_with_session(
+                        &measured,
+                        node,
+                        &point,
+                        layer,
+                        &mut session,
+                    )
+                    .await
+            }
+            (Planner::OneOff, true) => index.stage_delete(&measured, node).await,
+            (Planner::OneOff, false) => {
+                index
+                    .stage_upsert_at_layer(&measured, node, &point, layer)
+                    .await
+            }
+        }
+        .unwrap_or_else(|error| panic!("{context}: {error}"));
+        if matches!(planner, Planner::Session) {
+            session.flush_all(&measured).unwrap();
+            session.enforce_limits(&measured).unwrap();
+            session.admit_entity();
+        }
+        txn.commit().await.unwrap();
+        if delete {
+            points.remove(&node);
+        } else {
+            points.insert(node, point);
+        }
+        let after = Graph::read(&index, db.snapshot().await.unwrap().as_ref()).await;
+        let created = self_links(&after)
+            .difference(&before)
+            .copied()
+            .collect::<Vec<_>>();
+        assert!(created.is_empty(), "{context}: new self-links {created:?}");
+        assert!(
+            !delete || after.rows.iter().all(|(_, row)| *row != node),
+            "{context}: the delete removes every row of the node"
+        );
+        assert_eq!(
+            after.items,
+            points.keys().copied().collect::<BTreeSet<_>>(),
+            "{context}: items"
+        );
+    }
+}
+
+/// Random deletes and moves over a graph that keeps gaining self-links and
+/// links without locators always plan and never link a node to itself.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn damaged_graphs_never_gain_a_self_link() {
+    for planner in [Planner::Session, Planner::OneOff] {
+        damaged_run::<Euclidean>(0, planner).await;
+        damaged_run::<Cosine>(0, planner).await;
+    }
+}
diff --git a/crates/db/src/search/vector/hnsw/mutation/mod.rs b/crates/db/src/search/vector/hnsw/mutation/mod.rs
index 2074d2bdb..885bd9ce7 100644
--- a/crates/db/src/search/vector/hnsw/mutation/mod.rs
+++ b/crates/db/src/search/vector/hnsw/mutation/mod.rs
@@ -511,7 +511,9 @@ impl<D: Distance> VectorIndex<D> {
     ///
     /// Missing items fall through to the writable candidate index and return an
     /// owned item, or `None` when insertion must continue with an empty candidate
-    /// set. Any candidate cleanup remains staged in the caller's measured
+    /// set. The inserting node is never a root, even when stale metadata names
+    /// it: its item is already staged, so it would be its own nearest neighbor.
+    /// Any candidate cleanup remains staged in the caller's measured
     /// transaction; this method never mutates a resident snapshot.
     pub(in crate::search::vector) async fn resolve_beam_entry_point_for_insert(
         &self,
@@ -521,9 +523,10 @@ impl<D: Distance> VectorIndex<D> {
         inserting_node_id: NodeId,
         mutation_cache: &mut MutationOpCache<D>,
     ) -> Result<Option<(NodeId, Item<'static, D>)>, HelixDbError> {
-        if let Some(item) = self
-            .get_item_for_layer_cached(txn, layer, entry_point, mutation_cache)
-            .await?
+        if entry_point != inserting_node_id
+            && let Some(item) = self
+                .get_item_for_layer_cached(txn, layer, entry_point, mutation_cache)
+                .await?
         {
             return Ok(Some((entry_point, item.as_ref().clone())));
         }
@@ -532,6 +535,7 @@ impl<D: Distance> VectorIndex<D> {
             .find_best_entry_candidate_cached(txn, mutation_cache)
             .await?
             && replacement_entry_point != entry_point
+            && replacement_entry_point != inserting_node_id
             && let Some(item) = self
                 .get_item_for_layer_cached(txn, layer, replacement_entry_point, mutation_cache)
                 .await?
@@ -888,6 +892,7 @@ impl<D: Distance> VectorIndex<D> {
                         item,
                         current_entry_point,
                         layer,
+                        node_id,
                         mutation_cache,
                     )
                     .await?;
@@ -976,6 +981,12 @@ impl<D: Distance> VectorIndex<D> {
     /// neighbor/item cache so staged rows are authoritative and speculative
     /// layer-0 reads remain bounded. Missing entry points are resolved through
     /// the write-side recovery contract before expansion begins.
+    ///
+    /// `inserting_node_id` is never a candidate. Its item is staged before the
+    /// search, and a row may still link it: a link released versions left
+    /// without a reverse locator outlives the node's delete (see
+    /// [`Self::stage_delete_with_metadata`]). Admitting it would select the
+    /// node as its own neighbor.
     #[allow(clippy::too_many_arguments)]
     pub(in crate::search::vector) async fn search_layer_beam(
         &self,
@@ -1008,6 +1019,7 @@ impl<D: Distance> VectorIndex<D> {
         else {
             return Ok(Vec::new());
         };
+        visited.insert(inserting_node_id);
         let entry_distance = D::distance(query, &entry_item);
         candidates.push(Reverse(Candidate::try_new(
             resolved_entry_point,
@@ -1081,15 +1093,20 @@ impl<D: Distance> VectorIndex<D> {
     }
 
     /// Greedily descends through reusable item and neighbor cache state.
+    ///
+    /// Like [`Self::search_layer_beam`], it never moves to
+    /// `inserting_node_id`, whose item is already staged.
     async fn search_layer_greedy_for_mutation(
         &self,
         txn: &MeasuredVectorTransaction<'_>,
         query: &Item<'_, D>,
         entry_point: NodeId,
         layer: u16,
+        inserting_node_id: NodeId,
         mutation_cache: &mut MutationOpCache<D>,
     ) -> Result<NodeId, HelixDbError> {
         let mut visited = foldhash::HashSet::default();
+        visited.insert(inserting_node_id);
         let mut current = entry_point;
         let Some(current_item) = self
             .get_item_for_layer_cached(txn, layer, current, mutation_cache)
@@ -1715,9 +1732,10 @@ impl<D: Distance> VectorIndex<D> {
     /// link's locator or the node's own rows, so a link without a locator
     /// from a source the node does not link back survives the delete. A
     /// reinsertion of the node in the same cache then searches through it
-    /// back to the node and fails; otherwise the link dangles. Released
-    /// versions v3.1.0 through v3.4.2 (Docker images through v0.0.9) left links
-    /// without locators on re-embeddings, and nothing here repairs them.
+    /// back to the node, which the search skips, and the link names the node
+    /// again; otherwise the link dangles. Released versions v3.1.0 through
+    /// v3.4.2 (Docker images through v0.0.9) left links without locators on
+    /// re-embeddings, and nothing here repairs them.
     pub(in crate::search::vector) async fn stage_delete_with_metadata(
         &self,
         txn: &MeasuredVectorTransaction<'_>,
diff --git a/crates/db/src/search/vector/hnsw/neighbor_set.rs b/crates/db/src/search/vector/hnsw/neighbor_set.rs
index c7bb8ecc3..236ec34e0 100644
--- a/crates/db/src/search/vector/hnsw/neighbor_set.rs
+++ b/crates/db/src/search/vector/hnsw/neighbor_set.rs
@@ -3,9 +3,10 @@
 //! Persisted neighbor rows retain their deployed byte codecs. The vector core
 //! converts them into [`NeighborSet`] before mutation or difference work, so a
 //! set is always sorted, unique, self-free, and within its layer degree limit.
-//! Historical upper-layer rows may be ordered by distance rather than node ID;
-//! [`NeighborSet::try_from_deployed`] sorts that decoded compatibility input in
-//! memory without rewriting the row. New core state uses the strict canonical
+//! Historical upper-layer rows may be ordered by distance rather than node ID,
+//! and damaged rows may link their own node; [`NeighborSet::try_from_deployed`]
+//! sorts that decoded compatibility input and drops the self-link in memory
+//! without rewriting the row. New core state uses the strict canonical
 //! constructor and encodes through the unchanged existing codecs.
 
 use std::num::NonZeroUsize;
@@ -112,14 +113,20 @@ impl NeighborSet {
 
     /// Adapts a decoded deployed row into canonical runtime order.
     ///
-    /// Existing upper-neighbor rows may retain distance order. Sorting here is
-    /// runtime-only and preserves restart compatibility; duplicates, self-links,
-    /// and degree violations still fail closed as corrupt graph state.
+    /// Existing upper-neighbor rows may retain distance order, and a row
+    /// damaged by a released version may link its own node. Sorting and
+    /// dropping the self-link here are runtime-only: traversal never follows
+    /// a self-link, so the row means the same without it, and a mutation that
+    /// changes the row stores it self-free. Failing instead would fail every
+    /// mutation that loads the row, including the ones that would change it.
+    /// Duplicates and degree violations still fail closed as corrupt graph
+    /// state.
     pub(crate) fn try_from_deployed(
         owner: NodeId,
         degree_limit: NeighborDegreeLimit,
         mut nodes: Vec<NodeId>,
     ) -> Result<Self, NeighborSetError> {
+        nodes.retain(|node| *node != owner);
         nodes.sort_unstable();
         Self::try_from_canonical(owner, degree_limit, nodes)
     }
@@ -270,13 +277,28 @@ mod tests {
     }
 
     #[test]
-    fn deployed_adapter_canonicalizes_order_but_not_corruption() {
+    fn deployed_adapter_canonicalizes_order_and_self_links_but_not_corruption() {
         let set = NeighborSet::try_from_deployed(9, limit(3), vec![3, 1, 2]).unwrap();
         assert_eq!(set.as_slice(), &[1, 2, 3]);
+        // A full row that also links its owner fits once the self-link drops.
+        let set = NeighborSet::try_from_deployed(9, limit(3), vec![3, 9, 1, 2]).unwrap();
+        assert_eq!(set.as_slice(), &[1, 2, 3]);
+        assert!(!set.contains(9));
+        assert_eq!(
+            NeighborSet::try_from_deployed(9, limit(3), vec![9]).unwrap(),
+            NeighborSet::empty(9, limit(3))
+        );
         assert_eq!(
             NeighborSet::try_from_deployed(9, limit(3), vec![2, 1, 2]),
             Err(NeighborSetError::Duplicate(2))
         );
+        assert_eq!(
+            NeighborSet::try_from_deployed(9, limit(3), vec![4, 3, 9, 1, 2]),
+            Err(NeighborSetError::DegreeExceeded {
+                limit: 3,
+                actual: 4
+            })
+        );
     }
 
     #[test]
diff --git a/crates/db/tests/production_support/vector/mutation.rs b/crates/db/tests/production_support/vector/mutation.rs
index d417106a9..240911423 100644
--- a/crates/db/tests/production_support/vector/mutation.rs
+++ b/crates/db/tests/production_support/vector/mutation.rs
@@ -444,7 +444,14 @@ async fn run_neighbor_write_contracts(db: &Db) {
     let mut mutation_cache = MutationOpCache::<Cosine>::default();
     assert_eq!(
         index
-            .search_layer_greedy_for_mutation(&measured, &item, 999_999, 0, &mut mutation_cache,)
+            .search_layer_greedy_for_mutation(
+                &measured,
+                &item,
+                999_999,
+                0,
+                600,
+                &mut mutation_cache,
+            )
             .await
             .unwrap(),
         999_999
@@ -578,18 +585,46 @@ async fn run_neighbor_write_contracts(db: &Db) {
         .await
         .unwrap();
 
+    // A stored self-link, as damage leaves one, loads without it; a row over
+    // its degree still fails closed.
     let rows = VectorWriteRows::new(&measured, index.row_keyspace());
-    rows.put_layer0_neighbors(701, &[701]).unwrap();
+    rows.put_layer0_neighbors(701, &[2, 701]).unwrap();
+    let mut self_linked = MutationOpCache::<Cosine>::with_degree_limits(8, 4).unwrap();
+    assert_eq!(
+        index
+            .load_neighbors_for_mutation(&measured, 0, 701, &mut self_linked)
+            .await
+            .unwrap(),
+        [2]
+    );
+    rows.put_layer0_neighbors(702, &[702]).unwrap();
+    assert_eq!(
+        index
+            .prefetch_layer0_neighbors_for_mutation(&measured, &[702], &mut self_linked)
+            .await
+            .unwrap(),
+        1
+    );
+    assert_eq!(
+        self_linked
+            .neighbor(MutationOpCache::<Cosine>::node_row_id(0, 702))
+            .unwrap()
+            .current(),
+        &neighbors(702, Vec::new())
+    );
+
+    let over_degree = (1..=9).collect::<Vec<NodeId>>();
+    rows.put_layer0_neighbors(704, &over_degree).unwrap();
     let mut malformed_load = MutationOpCache::<Cosine>::with_degree_limits(8, 4).unwrap();
     assert!(index
-        .load_neighbors_for_mutation(&measured, 0, 701, &mut malformed_load)
+        .load_neighbors_for_mutation(&measured, 0, 704, &mut malformed_load)
         .await
         .is_err());
 
-    rows.put_layer0_neighbors(702, &[702]).unwrap();
+    rows.put_layer0_neighbors(705, &over_degree).unwrap();
     let mut malformed_prefetch = MutationOpCache::<Cosine>::with_degree_limits(8, 4).unwrap();
     assert!(index
-        .prefetch_layer0_neighbors_for_mutation(&measured, &[702], &mut malformed_prefetch)
+        .prefetch_layer0_neighbors_for_mutation(&measured, &[705], &mut malformed_prefetch)
         .await
         .is_err());
 

From 7b19fc9fccb9d494ec86a18e315dab775e17f6d9 Mon Sep 17 00:00:00 2001
From: xav-db <xav@helix-db.com>
Date: Sun, 4 Oct 2026 13:54:58 +0100
Subject: [PATCH 2/9] fix(db): hold back only the entity whose index planning
 fails

Planning one queued operation could fail the same way on every attempt,
for example on damaged graph rows. Publication retried the whole batch
forever, so its generation stalled, the queue filled, writes failed
with backpressure, and no entity was reported as blocked.

Publication failures are now classified by FailureKind. A deterministic
failure planning one entity holds that entity back through the existing
held-entity state: the vector planner reports the failing effect's
position, and a failed text epoch halves its entity ceiling until one
entity fails alone. The rest of the generation keeps publishing, the
held operations stay queued, and a newer write to the entity, or a
restart, plans it again. Transient failures still retry the batch with
backoff, and a deterministic failure outside one entity's planning
still retries the generation. No stored format changes.
---
 .../index_lifecycle/queue/isolation_tests.rs  | 469 ++++++++++++++++++
 crates/db/src/index_lifecycle/queue/mod.rs    |  26 +-
 .../src/index_lifecycle/queue/publication.rs  | 410 +++++++++++++--
 .../queue/publication_tests.rs                |  55 +-
 crates/db/src/index_lifecycle/vector.rs       |  43 +-
 .../src/index_lifecycle/vector/publication.rs |  33 +-
 crates/db/src/lib.rs                          |  11 +-
 .../production_support/queue_publication.rs   |  38 +-
 ...production_text_correctness_regressions.rs |  63 ++-
 .../helix-db/query-guides/troubleshooting.mdx |   6 +
 10 files changed, 1015 insertions(+), 139 deletions(-)
 create mode 100644 crates/db/src/index_lifecycle/queue/isolation_tests.rs

diff --git a/crates/db/src/index_lifecycle/queue/isolation_tests.rs b/crates/db/src/index_lifecycle/queue/isolation_tests.rs
new file mode 100644
index 000000000..a7ed0e20a
--- /dev/null
+++ b/crates/db/src/index_lifecycle/queue/isolation_tests.rs
@@ -0,0 +1,469 @@
+//! Deterministic planning failures hold back only their own entity, while
+//! transient failures still retry the whole batch.
+//!
+//! Failures are injected through [`PublicationHooks::planning_failures`],
+//! keyed by the operation an entity's selection ends at, so a newer write to
+//! the entity supersedes one. A corrupt injection fails the real planner: a
+//! vector replacement of the wrong dimension, or a text effect naming the
+//! other element kind.
+//!
+//! [`PublicationHooks::planning_failures`]: super::publication::test_hooks::PublicationHooks
+
+use std::num::NonZeroU64;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+
+use helix_ast::{
+    batch, graph::NodeRef, query::QueryRequest, query::SearchConsistency, traversal,
+    value::PropertyInput,
+};
+use slatedb::object_store::memory::InMemory;
+use slatedb::object_store::ObjectStore;
+
+use super::overlay_tests::{text_search, vector_search, write};
+use super::publication::test_hooks::InjectedPlanningFailure;
+use super::publication::{FailureKind, PublicationOutcome};
+use super::publication_tests::{install_vector, publisher};
+use super::tests::{add_doc, install_vector_and_text, open, queue, queued, target};
+use super::QueueTarget;
+use crate::config::{IndexOperationQueueTuning, TextIndexDefinition};
+use crate::encoding::v2::values::indexes::operation_queue::{QueueFamily, QueuedOperationId};
+use crate::error::HelixDbError;
+use crate::index_lifecycle::ValidatedDynamicIndexDefinition;
+use crate::HelixDB;
+
+/// Paused publication, and eventual searches that overlay no queued work, so
+/// they see exactly what is published.
+fn tuning() -> IndexOperationQueueTuning {
+    IndexOperationQueueTuning::default().with_eventual_search_budget_for_tests(0)
+}
+
+async fn install_text(db: &HelixDB) {
+    db.install_index_for_tests(
+        ValidatedDynamicIndexDefinition::try_from(
+            TextIndexDefinition::new_node("Doc", "body").unwrap(),
+        )
+        .unwrap(),
+    )
+    .await
+    .unwrap();
+}
+
+async fn set(db: &HelixDB, id: u64, property: &'static str, value: PropertyInput) {
+    write(db, || {
+        QueryRequest::write(
+            batch::write_batch().var_as(
+                "updated",
+                traversal::g()
+                    .n(NodeRef::from(id))
+                    .set_property(property, value.clone()),
+            ),
+        )
+    })
+    .await;
+}
+
+/// Entity IDs of `family`'s queued operations, in queue order.
+async fn queued_ids(db: &HelixDB, family: QueueFamily) -> Vec<u64> {
+    queue(db, family).await.map_or_else(Vec::new, |queue| {
+        queue
+            .operations()
+            .iter()
+            .map(|operation| operation.entity().id.get())
+            .collect()
+    })
+}
+
+/// The newest queued operation of entity `id` in `family`'s queue.
+async fn newest(db: &HelixDB, family: QueueFamily, id: u64) -> QueuedOperationId {
+    queue(db, family)
+        .await
+        .unwrap()
+        .operations()
+        .iter()
+        .rev()
+        .find(|operation| operation.entity().id.get() == id)
+        .unwrap()
+        .id()
+}
+
+fn inject(db: &HelixDB, operation: QueuedOperationId, failure: Option<InjectedPlanningFailure>) {
+    let mut failures = publisher(db).hooks().planning_failures.lock();
+    match failure {
+        Some(failure) => failures.insert(operation, failure),
+        None => failures.remove(&operation),
+    };
+}
+
+/// Publishes `target` until only held-back work is left, returning every
+/// outcome.
+async fn settle(db: &HelixDB, target: QueueTarget) -> Vec<PublicationOutcome> {
+    let mut outcomes = Vec::new();
+    for _ in 0..32 {
+        let outcome = publisher(db).publish_once(target).await.unwrap();
+        outcomes.push(outcome);
+        match outcome {
+            PublicationOutcome::Stalled | PublicationOutcome::Empty => return outcomes,
+            PublicationOutcome::Published { .. }
+            | PublicationOutcome::Trimmed
+            | PublicationOutcome::Blocked => {}
+            PublicationOutcome::Discarded { .. }
+            | PublicationOutcome::Deferred
+            | PublicationOutcome::Retry => panic!("publication did not progress: {outcomes:?}"),
+        }
+    }
+    panic!("publication did not settle: {outcomes:?}")
+}
+
+/// The published entity nearest `point`.
+async fn published_nearest(db: &HelixDB, point: [f32; 2]) -> Option<u64> {
+    vector_search(db, point, 1, None, SearchConsistency::Eventual)
+        .await
+        .first()
+        .map(|(id, _)| *id)
+}
+
+async fn published_text(db: &HelixDB, term: &str) -> Vec<u64> {
+    text_search(db, term, 10, None, SearchConsistency::Eventual)
+        .await
+        .into_iter()
+        .map(|(id, _)| id)
+        .collect()
+}
+
+fn held(db: &HelixDB) -> Vec<u64> {
+    db.blocked_index_entities()
+        .into_iter()
+        .map(|entity| entity.id.get())
+        .collect()
+}
+
+#[test]
+fn failures_are_classified_by_what_retrying_can_change() {
+    for (error, kind) in [
+        (HelixDbError::DatabaseClosed, FailureKind::Fatal),
+        (
+            HelixDbError::WriterFencedCommitOutcomeUnknown,
+            FailureKind::Fatal,
+        ),
+        (
+            HelixDbError::ObjectStore(slatedb::object_store::Error::Generic {
+                store: "test",
+                source: "outage".into(),
+            }),
+            FailureKind::Transient,
+        ),
+        (
+            HelixDbError::TransactionConflict("conflict".to_string()),
+            FailureKind::Transient,
+        ),
+        (
+            HelixDbError::QueryCancelledByReaderRetirement,
+            FailureKind::Transient,
+        ),
+        (
+            HelixDbError::InvariantViolation("neighbor set contains its owner 7".to_string()),
+            FailureKind::Deterministic,
+        ),
+        (
+            HelixDbError::IndexCatalogCorruption("no metadata".to_string()),
+            FailureKind::Deterministic,
+        ),
+        (
+            HelixDbError::InvalidDimension {
+                expected: 2,
+                got: 3,
+            },
+            FailureKind::Deterministic,
+        ),
+    ] {
+        assert_eq!(FailureKind::of(&error), kind, "{error}");
+    }
+}
+
+/// A queued vector update that fails to plan on every attempt is held back
+/// alone: every other entity keeps publishing, so writes to them are never
+/// refused even with the member cap at two, until a newer write to it
+/// publishes.
+#[tokio::test]
+async fn a_vector_entity_that_fails_to_plan_is_held_back_alone() {
+    let db = open(
+        "isolate-vector",
+        Arc::new(InMemory::new()),
+        queued(tuning().with_max_members(NonZeroU64::new(2).unwrap())),
+    )
+    .await;
+    install_vector(&db, None).await;
+    let mut published = Vec::new();
+    for x in 0..4_u8 {
+        published.push(add_doc(&db, vec![f32::from(x), 0.0], "doc").await.unwrap());
+        let target = target(&db, QueueFamily::Vector).await;
+        assert_eq!(
+            settle(&db, target).await,
+            [
+                PublicationOutcome::Published {
+                    operations: 1,
+                    entities: 1
+                },
+                PublicationOutcome::Empty
+            ]
+        );
+    }
+    let target = target(&db, QueueFamily::Vector).await;
+    let failing = published[1];
+    set(&db, failing, "embedding", vec![5.0_f32, 5.0].into()).await;
+    let corrupt = newest(&db, QueueFamily::Vector, failing).await;
+    inject(&db, corrupt, Some(InjectedPlanningFailure::Corrupt));
+
+    for (round, x) in (10..14_u8).enumerate() {
+        // The held update and this insert fill the member cap, so the next
+        // insert is admitted only once this one publishes.
+        let inserted = add_doc(&db, vec![f32::from(x), 0.0], "doc")
+            .await
+            .unwrap_or_else(|error| panic!("round {round}: a write was refused: {error}"));
+        let outcomes = settle(&db, target).await;
+        let expected = [
+            PublicationOutcome::Published {
+                operations: 1,
+                entities: 1,
+            },
+            PublicationOutcome::Stalled,
+        ];
+        if round == 0 {
+            // The first attempt plans the update first and holds it back.
+            assert_eq!(outcomes[0], PublicationOutcome::Blocked);
+            assert_eq!(outcomes[1..], expected);
+        } else {
+            assert_eq!(outcomes, expected, "round {round}: the update stays held");
+        }
+        assert_eq!(queued_ids(&db, QueueFamily::Vector).await, [failing]);
+        assert_eq!(held(&db), [failing]);
+        assert_eq!(db.blocked_index_entity_count(), 1);
+        assert_eq!(
+            published_nearest(&db, [f32::from(x), 0.0]).await,
+            Some(inserted)
+        );
+    }
+    let stats = db.index_operation_queue_stats();
+    assert_eq!(stats.blocked_entities, 1);
+    assert_eq!(stats.blocked_attempts, 1, "the update was planned once");
+    assert_eq!(stats.publication_error_retries, 0);
+    // The held update is queued, never acknowledged or dropped.
+    assert_eq!(published_nearest(&db, [1.0, 0.0]).await, Some(failing));
+    assert_eq!(
+        vector_search(&db, [5.0, 5.0], 1, None, SearchConsistency::Strong).await[0].0,
+        failing,
+        "strong search serves the held update"
+    );
+
+    // A newer write supersedes the corrupt one, and its repair publishes both.
+    set(&db, failing, "embedding", vec![6.0_f32, 6.0].into()).await;
+    assert_eq!(
+        publisher(&db).publish_once(target).await.unwrap(),
+        PublicationOutcome::Published {
+            operations: 2,
+            entities: 1
+        }
+    );
+    assert!(held(&db).is_empty());
+    assert_eq!(db.index_operation_queue_stats().blocked_entities, 0);
+    assert!(queued_ids(&db, QueueFamily::Vector).await.is_empty());
+    assert_eq!(published_nearest(&db, [6.0, 6.0]).await, Some(failing));
+    db.close().await.unwrap();
+}
+
+/// A text epoch fails as a whole, so its entity ceiling halves until the
+/// failing entity publishes alone and is held back; every other entity
+/// publishes, and a newer write to it publishes it.
+#[tokio::test]
+async fn a_text_entity_that_fails_to_plan_is_held_back_alone() {
+    let db = open("isolate-text", Arc::new(InMemory::new()), queued(tuning())).await;
+    install_text(&db).await;
+    let mut ids = Vec::new();
+    for word in ["apple", "banana", "cherry", "damson"] {
+        ids.push(add_doc(&db, vec![0.0, 0.0], word).await.unwrap());
+    }
+    let target = target(&db, QueueFamily::Text).await;
+    let corrupt = newest(&db, QueueFamily::Text, ids[2]).await;
+    inject(&db, corrupt, Some(InjectedPlanningFailure::Corrupt));
+    let published = |entities| PublicationOutcome::Published {
+        operations: entities,
+        entities,
+    };
+    assert_eq!(
+        settle(&db, target).await,
+        [
+            PublicationOutcome::Trimmed,
+            published(2),
+            PublicationOutcome::Trimmed,
+            PublicationOutcome::Blocked,
+            published(1),
+            PublicationOutcome::Stalled,
+        ]
+    );
+    assert_eq!(queued_ids(&db, QueueFamily::Text).await, [ids[2]]);
+    assert_eq!(held(&db), [ids[2]]);
+    for (id, word) in ids.iter().zip(["apple", "banana", "cherry", "damson"]) {
+        let expected = if *id == ids[2] { vec![] } else { vec![*id] };
+        assert_eq!(published_text(&db, word).await, expected, "{word}");
+    }
+    assert_eq!(db.index_operation_queue_stats().blocked_attempts, 1);
+
+    set(&db, ids[2], "body", "cherry repaired".to_string().into()).await;
+    assert_eq!(
+        publisher(&db).publish_once(target).await.unwrap(),
+        PublicationOutcome::Published {
+            operations: 2,
+            entities: 1
+        }
+    );
+    assert!(held(&db).is_empty());
+    assert!(queued_ids(&db, QueueFamily::Text).await.is_empty());
+    assert_eq!(published_text(&db, "repaired").await, [ids[2]]);
+    db.close().await.unwrap();
+}
+
+/// A transient planning failure retries the whole batch after backoff and
+/// holds nothing back, for both families; the batch publishes once the
+/// failure clears.
+#[tokio::test]
+async fn a_transient_planning_failure_retries_the_whole_batch() {
+    let db = open(
+        "isolate-transient",
+        Arc::new(InMemory::new()),
+        queued(tuning()),
+    )
+    .await;
+    install_vector_and_text(&db).await;
+    let mut ids = Vec::new();
+    for x in 0..3_u8 {
+        ids.push(add_doc(&db, vec![f32::from(x), 0.0], "word").await.unwrap());
+    }
+    for family in [QueueFamily::Vector, QueueFamily::Text] {
+        let target = target(&db, family).await;
+        let unavailable = newest(&db, family, ids[1]).await;
+        inject(&db, unavailable, Some(InjectedPlanningFailure::Unavailable));
+        for attempt in 1..=2 {
+            assert_eq!(
+                publisher(&db).publish_once(target).await.unwrap(),
+                PublicationOutcome::Retry,
+                "{family:?} attempt {attempt}"
+            );
+            assert_eq!(queued_ids(&db, family).await, ids, "{family:?}");
+            assert!(held(&db).is_empty(), "{family:?}");
+        }
+        inject(&db, unavailable, None);
+        assert_eq!(
+            publisher(&db).publish_once(target).await.unwrap(),
+            PublicationOutcome::Published {
+                operations: 3,
+                entities: 3
+            },
+            "{family:?}"
+        );
+    }
+    let stats = db.index_operation_queue_stats();
+    assert_eq!(stats.publication_error_retries, 4);
+    assert_eq!(stats.blocked_attempts, 0);
+    assert_eq!(stats.blocked_entities, 0);
+    db.close().await.unwrap();
+}
+
+/// Holding an entity back is process memory: after a restart the failing
+/// update is planned again, held back again in one attempt, and the
+/// generation keeps publishing around it.
+#[tokio::test]
+async fn an_entity_held_back_before_a_restart_is_held_back_again_after_it() {
+    let store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+    let db = open("isolate-restart", Arc::clone(&store), queued(tuning())).await;
+    install_vector(&db, None).await;
+    let kept = add_doc(&db, vec![0.0, 0.0], "doc").await.unwrap();
+    let failing = add_doc(&db, vec![1.0, 0.0], "doc").await.unwrap();
+    let target = target(&db, QueueFamily::Vector).await;
+    settle(&db, target).await;
+    set(&db, failing, "embedding", vec![5.0_f32, 5.0].into()).await;
+    let corrupt = newest(&db, QueueFamily::Vector, failing).await;
+    inject(&db, corrupt, Some(InjectedPlanningFailure::Corrupt));
+    assert_eq!(
+        settle(&db, target).await,
+        [PublicationOutcome::Blocked, PublicationOutcome::Stalled]
+    );
+    assert_eq!(held(&db), [failing]);
+    db.close().await.unwrap();
+
+    let db = open("isolate-restart", store, queued(tuning())).await;
+    assert!(held(&db).is_empty(), "nothing durable records the hold");
+    assert_eq!(queued_ids(&db, QueueFamily::Vector).await, [failing]);
+    inject(&db, corrupt, Some(InjectedPlanningFailure::Corrupt));
+    let inserted = add_doc(&db, vec![2.0, 0.0], "doc").await.unwrap();
+    assert_eq!(
+        settle(&db, target).await,
+        [
+            PublicationOutcome::Blocked,
+            PublicationOutcome::Published {
+                operations: 1,
+                entities: 1
+            },
+            PublicationOutcome::Stalled,
+        ]
+    );
+    assert_eq!(held(&db), [failing]);
+    assert_eq!(queued_ids(&db, QueueFamily::Vector).await, [failing]);
+    assert_eq!(published_nearest(&db, [2.0, 0.0]).await, Some(inserted));
+    assert_eq!(published_nearest(&db, [0.0, 0.0]).await, Some(kept));
+
+    set(&db, failing, "embedding", vec![6.0_f32, 6.0].into()).await;
+    assert_eq!(
+        publisher(&db).publish_once(target).await.unwrap(),
+        PublicationOutcome::Published {
+            operations: 2,
+            entities: 1
+        }
+    );
+    assert!(held(&db).is_empty());
+    assert_eq!(published_nearest(&db, [6.0, 6.0]).await, Some(failing));
+    db.close().await.unwrap();
+}
+
+/// The publisher's error path classifies a failure outside one entity's
+/// planning too: a corrupt queue value has no entity to hold back, so the
+/// generation retries.
+#[tokio::test]
+async fn a_deterministic_failure_outside_planning_holds_nothing_back() {
+    let db = open(
+        "isolate-outside-planning",
+        Arc::new(InMemory::new()),
+        queued(tuning()),
+    )
+    .await;
+    install_vector(&db, None).await;
+    add_doc(&db, vec![0.0, 0.0], "doc").await.unwrap();
+    let target = target(&db, QueueFamily::Vector).await;
+    let storage = db.inner_db();
+    let stored = storage.get(target.key()).await.unwrap().unwrap();
+    storage
+        .put(target.key(), b"not an operation queue")
+        .await
+        .unwrap();
+    assert_eq!(
+        publisher(&db).publish_once(target).await.unwrap(),
+        PublicationOutcome::Retry
+    );
+    assert!(held(&db).is_empty());
+    assert_eq!(
+        publisher(&db)
+            .metrics()
+            .error_retries
+            .load(Ordering::Relaxed),
+        1
+    );
+    storage.put(target.key(), &stored).await.unwrap();
+    assert_eq!(
+        publisher(&db).publish_once(target).await.unwrap(),
+        PublicationOutcome::Published {
+            operations: 1,
+            entities: 1
+        }
+    );
+    db.close().await.unwrap();
+}
diff --git a/crates/db/src/index_lifecycle/queue/mod.rs b/crates/db/src/index_lifecycle/queue/mod.rs
index 26720e9a9..02405a5bf 100644
--- a/crates/db/src/index_lifecycle/queue/mod.rs
+++ b/crates/db/src/index_lifecycle/queue/mod.rs
@@ -86,10 +86,12 @@ pub struct IndexOperationQueueStats {
     /// Publication commits with an unknown outcome.
     pub uncertain_commits: u64,
     /// Attempts retried with fewer entities or operations after exceeding an
-    /// output budget.
+    /// output budget, or with fewer text entities after planning a text epoch
+    /// failed deterministically.
     pub output_retries: u64,
-    /// Attempts where one operation's effect and acknowledgement alone
-    /// exceeded an output budget, holding its entity back.
+    /// Attempts that held an entity back: one operation's effect and
+    /// acknowledgement alone exceeded an output budget, or planning the
+    /// entity failed deterministically.
     pub blocked_attempts: u64,
     /// Entities held back right now (a gauge, unlike the publication
     /// counters): see [`crate::HelixDB::blocked_index_entities`].
@@ -136,7 +138,8 @@ pub struct IndexOperationQueueStats {
     /// Attempts that must rediscover and retry: commit conflicts, uncertain
     /// commits, retryable errors, and ownership changes after classification.
     pub publication_retries: u64,
-    /// Retries caused by a retryable storage or decoding error.
+    /// Retries caused by an error: a transient one, such as storage I/O, or a
+    /// deterministic one that no single entity's planning raised.
     pub publication_error_retries: u64,
     /// Attempts deferred because a hidden build owns the generation.
     pub deferred_attempts: u64,
@@ -144,7 +147,9 @@ pub struct IndexOperationQueueStats {
 
 /// One entity whose queued vector/text work is held back because one of its
 /// operations alone can never fit a publication under the current limits,
-/// for example after they were lowered.
+/// for example after they were lowered, or because planning its change fails
+/// deterministically, for example on damaged index rows or a corrupt queued
+/// payload.
 ///
 /// It blocks only its own publication: the rest of its generation keeps
 /// publishing. Its operations stay queued, so strong searches keep serving
@@ -162,9 +167,12 @@ pub struct IndexOperationQueueStats {
 /// the entity within their budget can.
 ///
 /// A rewrite or delete repairs it only once one publication fits the change
-/// from its published state: removing a document published under larger
-/// limits, or relinking a deleted vector's neighbors, can exceed the lowered
-/// limits too. Raising the limits again lets it publish.
+/// from its published state and planning that change succeeds: removing a
+/// document published under larger limits, or relinking a deleted vector's
+/// neighbors, can exceed the lowered limits too. Raising the limits again
+/// lets it publish. An entity held back after its planning failed is planned
+/// again only on a later write to it or once the writer restarts, so after
+/// repairing what failed, write it again or restart the writer.
 ///
 /// Its queued text still counts toward the pending text a strong text search
 /// may analyze in its partition (one text publication's analysis budget),
@@ -245,6 +253,8 @@ impl QueueTarget {
 #[cfg(test)]
 mod codec_storage_tests;
 #[cfg(test)]
+mod isolation_tests;
+#[cfg(test)]
 mod layout_tests;
 #[cfg(test)]
 mod lifecycle_tests;
diff --git a/crates/db/src/index_lifecycle/queue/publication.rs b/crates/db/src/index_lifecycle/queue/publication.rs
index 67f8361bc..e4b0322eb 100644
--- a/crates/db/src/index_lifecycle/queue/publication.rs
+++ b/crates/db/src/index_lifecycle/queue/publication.rs
@@ -25,6 +25,9 @@
 //! 6. Stage the longest prefix of entities whose exact output fits beside the
 //!    acknowledgement: vectors through the build planner
 //!    ([`crate::index_lifecycle::vector::publication`]), text as one epoch.
+//!    When planning fails, [`FailureKind`] decides: an entity whose planning
+//!    fails deterministically is held back like one that cannot fit, and the
+//!    rest publish from the next attempt; anything else retries the batch.
 //! 7. Stage one acknowledgement naming exactly the published IDs.
 //! 8. Commit through the vector cache's commit fence, release accounting and
 //!    retain the rest of the queue, then retire emptied partition caches and
@@ -103,12 +106,14 @@ pub(crate) enum PublicationOutcome {
     Deferred,
     /// A serializable conflict or uncertain commit; rediscover and retry.
     Retry,
-    /// Exact output crossed a budget before anything fit; retry immediately
-    /// with fewer text entities, or fewer operations.
+    /// Exact output crossed a budget before anything fit, or planning a text
+    /// epoch failed deterministically; retry immediately with fewer text
+    /// entities, or fewer operations.
     Trimmed,
     /// One operation's effect and acknowledgement cannot fit an output budget,
-    /// so its entity is held back until a later operation supersedes it;
-    /// retry immediately with the generation's other entities.
+    /// or planning one entity failed deterministically, so its entity is held
+    /// back until a later operation supersedes it; retry immediately with the
+    /// generation's other entities.
     Blocked,
     /// Every queued entity is held back after blocking; nothing was attempted.
     /// The generation waits for a new operation rather than retrying on a
@@ -160,15 +165,16 @@ struct TargetSchedule {
     /// Consecutive attempts without progress.
     failures: u32,
     /// Entities held back because a lone operation of theirs could not fit a
-    /// publication, or draining after their repair. Process memory only: a
-    /// restarted publisher rediscovers blocked entities by blocking again,
-    /// and publishes a draining entity's remaining operations in regular
-    /// batches (see [`HeldEntity`]).
+    /// publication or failed to plan, or draining after their repair. Process
+    /// memory only: a restarted publisher rediscovers blocked entities by
+    /// blocking again, and publishes a draining entity's remaining operations
+    /// in regular batches (see [`HeldEntity`]).
     held: HashMap<IndexEntity, HeldEntity>,
 }
 
 /// One entity held back after one of its operations could not fit a
-/// publication.
+/// publication, or failed to plan deterministically
+/// ([`QueuePublisher::isolate`]).
 ///
 /// A held entity is selected only alone, as a repair, once the rotation
 /// reaches it ahead of every entity that is not held back; a batch the
@@ -354,11 +360,23 @@ impl Drop for AttemptClaim<'_> {
 pub(crate) mod test_hooks {
     //! Deterministic interleaving and failure seams for publication tests.
 
+    use std::collections::HashMap;
     use std::sync::atomic::AtomicBool;
+    use std::sync::Arc;
 
     use parking_lot::Mutex;
     use tokio::sync::oneshot;
 
+    use super::SelectedEntity;
+    use crate::encoding::v2::values::indexes::operation_queue::{
+        QueuedOperationId, QueuedVectorReplacement,
+    };
+    use crate::error::{HelixDbError, Result};
+    use crate::index_lifecycle::text::active_batch::QueuedTextEffect;
+    use crate::index_lifecycle::vector::publication::QueuedVectorEffect;
+    use crate::index_lifecycle::work::TextPartition;
+    use crate::index_lifecycle::IndexElementKind;
+
     /// Test-only controls installed on one publisher.
     #[derive(Debug, Default)]
     pub(crate) struct PublicationHooks {
@@ -380,6 +398,108 @@ pub(crate) mod test_hooks {
         /// Row-batch fetch policy of the last staged vector attempt's
         /// mutation indexes.
         pub(crate) batch_reads: Mutex<Option<crate::batch_reads::BatchReads>>,
+        /// Failures injected into planning an entity whose newest selected
+        /// operation is the key, so a newer operation supersedes one.
+        pub(crate) planning_failures: Mutex<HashMap<QueuedOperationId, InjectedPlanningFailure>>,
+    }
+
+    /// A failure injected into planning one entity.
+    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+    pub(crate) enum InjectedPlanningFailure {
+        /// The entity plans as a corrupt payload would, so planning fails the
+        /// same way on every attempt: a vector replacement of the wrong
+        /// dimension, or a text effect naming the other element kind.
+        Corrupt,
+        /// Planning the entity fails with an object-store error, as an
+        /// outage would.
+        Unavailable,
+    }
+
+    impl PublicationHooks {
+        /// The failure injected into planning each of `selection`'s
+        /// entities, in order.
+        pub(crate) fn planning_failures(
+            &self,
+            selection: &[SelectedEntity<'_>],
+        ) -> Vec<Option<InjectedPlanningFailure>> {
+            let failures = self.planning_failures.lock();
+            selection
+                .iter()
+                .map(|selected| {
+                    selected
+                        .taken()
+                        .last()
+                        .and_then(|newest| failures.get(&newest.id()).copied())
+                })
+                .collect()
+        }
+    }
+
+    /// Gives every effect `injected` corrupts a replacement of the wrong
+    /// dimension.
+    pub(crate) fn corrupt_vector_effects(
+        effects: Vec<QueuedVectorEffect>,
+        injected: &[Option<InjectedPlanningFailure>],
+    ) -> Vec<QueuedVectorEffect> {
+        effects
+            .into_iter()
+            .zip(injected)
+            .map(|(effect, injected)| match injected {
+                Some(InjectedPlanningFailure::Corrupt) => QueuedVectorEffect {
+                    replacement: Some(
+                        QueuedVectorReplacement::try_new(
+                            TextPartition::Unpartitioned,
+                            Arc::from([0.5_f32; 3]),
+                        )
+                        .expect("a finite vector is a valid replacement"),
+                    ),
+                    ..effect
+                },
+                Some(InjectedPlanningFailure::Unavailable) | None => effect,
+            })
+            .collect()
+    }
+
+    /// Names the other element kind in every effect `injected` corrupts.
+    pub(crate) fn corrupt_text_effects(
+        effects: Vec<QueuedTextEffect>,
+        injected: &[Option<InjectedPlanningFailure>],
+    ) -> Vec<QueuedTextEffect> {
+        effects
+            .into_iter()
+            .zip(injected)
+            .map(|(mut effect, injected)| {
+                if *injected == Some(InjectedPlanningFailure::Corrupt) {
+                    effect.entity.kind = match effect.entity.kind {
+                        IndexElementKind::Node => IndexElementKind::Edge,
+                        IndexElementKind::Edge => IndexElementKind::Node,
+                    };
+                }
+                effect
+            })
+            .collect()
+    }
+
+    /// Fails planning with an object-store error at the first entity
+    /// `injected` makes unavailable, if any.
+    pub(crate) fn fail_unavailable<T>(
+        planned: Result<T>,
+        injected: &[Option<InjectedPlanningFailure>],
+        failed: impl FnOnce(usize, HelixDbError) -> Result<T>,
+    ) -> Result<T> {
+        let Some(position) = injected
+            .iter()
+            .position(|injected| *injected == Some(InjectedPlanningFailure::Unavailable))
+        else {
+            return planned;
+        };
+        failed(
+            position,
+            HelixDbError::ObjectStore(slatedb::object_store::Error::Generic {
+                store: "injected",
+                source: "injected planning outage".into(),
+            }),
+        )
     }
 }
 
@@ -487,9 +607,11 @@ impl QueuePublisher {
     /// attempt is its queue's only acknowledger, and reconciliation and
     /// discard rely on that. A call while another attempt holds the target
     /// returns [`PublicationOutcome::Retry`] without reading the queue or
-    /// touching its schedule. Retryable storage outcomes are classified here;
-    /// only errors that make the writer unusable (closed or fenced storage)
-    /// are returned.
+    /// touching its schedule. Failures are classified by [`FailureKind`]: one
+    /// entity's deterministic planning failure holds that entity back within
+    /// the attempt, any other failure that leaves the writer usable retries
+    /// the generation after backoff, and only errors that make the writer
+    /// unusable (closed or fenced storage) are returned.
     pub(crate) async fn publish_once(&self, target: QueueTarget) -> Result<PublicationOutcome> {
         if !self.attempts.lock().insert(target) {
             return Ok(PublicationOutcome::Retry);
@@ -505,17 +627,22 @@ impl QueuePublisher {
         let started = Instant::now();
         let outcome = match self.try_publish(target).await {
             Ok(outcome) => outcome,
-            Err(error) if is_fatal(&error) => return Err(error),
-            Err(error) => {
-                self.metrics.error_retries.fetch_add(1, Ordering::Relaxed);
-                tracing::warn!(
-                    %error,
-                    index_id = target.index_id.get(),
-                    generation = target.generation.get(),
-                    "queued index publication failed; retrying after backoff"
-                );
-                PublicationOutcome::Retry
-            }
+            Err(error) => match FailureKind::of(&error) {
+                FailureKind::Fatal => return Err(error),
+                // A deterministic failure that no one entity's planning
+                // raised has no entity to hold back.
+                kind @ (FailureKind::Transient | FailureKind::Deterministic) => {
+                    self.metrics.error_retries.fetch_add(1, Ordering::Relaxed);
+                    tracing::warn!(
+                        %error,
+                        ?kind,
+                        index_id = target.index_id.get(),
+                        generation = target.generation.get(),
+                        "queued index publication failed; retrying after backoff"
+                    );
+                    PublicationOutcome::Retry
+                }
+            },
         };
         self.metrics.attempts.fetch_add(1, Ordering::Relaxed);
         self.metrics.attempt_micros.fetch_add(
@@ -734,7 +861,7 @@ impl QueuePublisher {
             }
             Err(error) => {
                 let error = HelixDbError::from_storage_commit(error);
-                if is_fatal(&error) {
+                if FailureKind::of(&error) == FailureKind::Fatal {
                     return Err(error);
                 }
                 self.metrics
@@ -854,10 +981,24 @@ impl QueuePublisher {
                 let _ = release.await;
             }
         }
-        let effects = selection
+        // Collapsing reads only the decoded queue, so it fails the same way
+        // on every attempt.
+        let effects = match selection
             .iter()
-            .map(|selected| collapse_vector(selected))
-            .collect::<Result<Vec<_>>>()?;
+            .map(|selected| collapse_vector(selected).map_err(|error| (selected, error)))
+            .collect::<std::result::Result<Vec<_>, _>>()
+        {
+            Ok(effects) => effects,
+            Err((failed, error)) => {
+                let outcome = self.isolate(target, std::slice::from_ref(failed), &error);
+                self.store.retained().retain(target, stored, &[]);
+                return Ok(outcome);
+            }
+        };
+        #[cfg(test)]
+        let injected = self.hooks.planning_failures(&selection);
+        #[cfg(test)]
+        let effects = test_hooks::corrupt_vector_effects(effects, &injected);
         // Admission reserves the whole selection's acknowledgement, which
         // bounds that of any prefix.
         let reserved = self.store.acknowledgement_output(
@@ -871,7 +1012,7 @@ impl QueuePublisher {
         let cache_writes = vector::VectorCacheWriteSet::default();
         let commit =
             NonZeroU64::MIN.saturating_add(self.vector_commits.fetch_add(1, Ordering::Relaxed));
-        let (staged, retained) = match stage_active_effects(
+        let staged = stage_active_effects(
             &self.db,
             &transaction,
             ownership,
@@ -884,8 +1025,12 @@ impl QueuePublisher {
             last_vector_commit,
             commit,
         )
-        .await
-        {
+        .await;
+        #[cfg(test)]
+        let staged = test_hooks::fail_unavailable(staged, &injected, |position, error| {
+            Ok(StagedEffects::Failed { position, error })
+        });
+        let (staged, retained) = match staged {
             Ok(StagedEffects::Prefix { staged, retained }) => (staged.get(), retained),
             Ok(StagedEffects::NoneFits) => {
                 let outcome = self.shrink(target, &selection, 0);
@@ -905,15 +1050,27 @@ impl QueuePublisher {
                 }
                 return Ok(outcome);
             }
+            Ok(StagedEffects::Failed { position, error })
+                if FailureKind::of(&error) == FailureKind::Deterministic =>
+            {
+                let outcome =
+                    self.isolate(target, std::slice::from_ref(&selection[position]), &error);
+                // Nothing committed: the next selection, without the held
+                // entity, reuses the queue.
+                self.store.retained().retain(target, stored, &[]);
+                return Ok(outcome);
+            }
             // Planning proved the transaction cannot commit.
-            Err(error) if error.is_transaction_conflict() => {
+            Ok(StagedEffects::Failed { error, .. }) | Err(error)
+                if error.is_transaction_conflict() =>
+            {
                 self.metrics
                     .commit_conflicts
                     .fetch_add(1, Ordering::Relaxed);
                 self.store.retained().retain(target, stored, &[]);
                 return Ok(PublicationOutcome::Retry);
             }
-            Err(error) => return Err(error),
+            Ok(StagedEffects::Failed { error, .. }) | Err(error) => return Err(error),
         };
         // Exactly the staged prefix is acknowledged; later entities stay
         // queued for the next attempt.
@@ -971,7 +1128,7 @@ impl QueuePublisher {
             }
             Err(error) => {
                 let error = HelixDbError::from_storage_commit(error);
-                if is_fatal(&error) {
+                if FailureKind::of(&error) == FailureKind::Fatal {
                     return Err(error);
                 }
                 self.record_uncertain_commit(target, selection, &acknowledged);
@@ -1194,6 +1351,64 @@ impl QueuePublisher {
         outcome
     }
 
+    /// Narrows `target`'s next attempt toward the entity of `failed`, a
+    /// selection whose planning failed deterministically with `error`.
+    ///
+    /// A failed selection of one entity holds it back as [`HeldEntity`]
+    /// describes: it waits for an operation newer than every state of it
+    /// known not to publish, which a later write supplies and its repair then
+    /// plans, and the rotation moves past it, so the rest of its generation
+    /// keeps publishing. Nothing is acknowledged or dropped, and nothing
+    /// durable records the hold: a restarted publisher plans the entity again
+    /// and holds it back again if it still fails. A text epoch is planned as
+    /// a whole, so its failure does not name an entity; a failed epoch of
+    /// several entities halves the text entity ceiling instead, until a
+    /// failing epoch is one entity.
+    fn isolate(
+        &self,
+        target: QueueTarget,
+        failed: &[SelectedEntity<'_>],
+        error: &HelixDbError,
+    ) -> PublicationOutcome {
+        let mut schedules = self.schedules.lock();
+        let schedule = schedules
+            .entry(target)
+            .or_insert_with(|| TargetSchedule::new(self.limits.max_entities().get()));
+        let [only] = failed else {
+            debug_assert!(failed.len() > 1, "an attempt selects at least one entity");
+            schedule.entity_limit = failed.len() / 2;
+            self.metrics.output_retries.fetch_add(1, Ordering::Relaxed);
+            return PublicationOutcome::Trimmed;
+        };
+        // A narrowed repair failing says nothing of the wider states its
+        // first attempt already found not to fit.
+        let through = match schedule.held.get(&only.entity) {
+            Some(HeldEntity::Repairing { tried, .. }) => *tried,
+            Some(HeldEntity::Waiting { .. } | HeldEntity::Draining { .. }) | None => only
+                .taken()
+                .last()
+                .expect("a selected entity has operations")
+                .id(),
+        };
+        schedule
+            .held
+            .insert(only.entity, HeldEntity::Waiting { through });
+        schedule.cursor = Some(only.entity);
+        self.metrics
+            .blocked_attempts
+            .fetch_add(1, Ordering::Relaxed);
+        tracing::error!(
+            %error,
+            scope = ?target.scope,
+            entity = ?only.entity,
+            index_id = target.index_id.get(),
+            generation = target.generation.get(),
+            "planning a queued index operation failed deterministically; its entity is held \
+             back until a later write supersedes it"
+        );
+        PublicationOutcome::Blocked
+    }
+
     /// Records a publication commit of `selection` whose outcome is unknown.
     ///
     /// The acknowledgement may have committed, so its capacity stays charged
@@ -1288,10 +1503,23 @@ impl QueuePublisher {
             // Ownership changed after classification; classify again.
             return Ok(PublicationOutcome::Retry);
         };
-        let effects = selection
+        // As for vectors, collapsing fails the same way on every attempt.
+        let effects = match selection
             .iter()
-            .map(collapse_text)
-            .collect::<Result<Vec<_>>>()?;
+            .map(|selected| collapse_text(selected).map_err(|error| (selected, error)))
+            .collect::<std::result::Result<Vec<_>, _>>()
+        {
+            Ok(effects) => effects,
+            Err((failed, error)) => {
+                let outcome = self.isolate(target, std::slice::from_ref(failed), &error);
+                self.store.retained().retain(target, stored, &[]);
+                return Ok(outcome);
+            }
+        };
+        #[cfg(test)]
+        let injected = self.hooks.planning_failures(&selection);
+        #[cfg(test)]
+        let effects = test_hooks::corrupt_text_effects(effects, &injected);
         let acknowledged = selection
             .iter()
             .flat_map(|selected| selected.operations.iter().map(|operation| operation.id()))
@@ -1299,15 +1527,17 @@ impl QueuePublisher {
         let acknowledgement = self
             .store
             .acknowledgement_output(target, &stored, &acknowledged)?;
-        let prepared = match crate::index_lifecycle::text::active_batch::prepare_queued_text_epoch(
+        let prepared = crate::index_lifecycle::text::active_batch::prepare_queued_text_epoch(
             &transaction,
             &handle,
             effects,
             self.text.limits,
             acknowledgement,
         )
-        .await
-        {
+        .await;
+        #[cfg(test)]
+        let prepared = test_hooks::fail_unavailable(prepared, &injected, |_, error| Err(error));
+        let prepared = match prepared {
             Ok(prepared) => prepared,
             // Producers and builds admit every document to half of each
             // per-entity budget and bound its lone split within the split
@@ -1332,6 +1562,13 @@ impl QueuePublisher {
                 }
                 return Ok(outcome);
             }
+            Err(error) if FailureKind::of(&error) == FailureKind::Deterministic => {
+                let outcome = self.isolate(target, &selection, &error);
+                // Nothing committed: the next, narrower epoch reuses the
+                // queue.
+                self.store.retained().retain(target, stored, &[]);
+                return Ok(outcome);
+            }
             Err(error) => return Err(error),
         };
         // Immutable split artifacts become durable before any committed
@@ -1375,7 +1612,7 @@ impl QueuePublisher {
             }
             Err(error) => {
                 let error = HelixDbError::from_storage_commit(error);
-                if is_fatal(&error) {
+                if FailureKind::of(&error) == FailureKind::Fatal {
                     return Err(error);
                 }
                 self.record_uncertain_commit(target, &selection, &acknowledged);
@@ -1666,13 +1903,88 @@ async fn load_generation_record(
 #[path = "../../../tests/production_support/queue_publication.rs"]
 pub(crate) mod production_contracts;
 
-/// Errors after which this writer can no longer make durable progress.
-fn is_fatal(error: &HelixDbError) -> bool {
-    matches!(
-        error,
-        HelixDbError::DatabaseClosed | HelixDbError::WriterFencedCommitOutcomeUnknown
-    ) || matches!(
-        error,
-        HelixDbError::Storage(error) if matches!(error.kind(), slatedb::ErrorKind::Closed(_))
-    )
+/// What retrying a failed publication can change, which decides how the
+/// failure is handled.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum FailureKind {
+    /// The writer can make no further durable progress (closed or fenced
+    /// storage): the error is returned and publication stops.
+    Fatal,
+    /// Storage or object-store I/O, a conflict, a lifecycle race, or a
+    /// cancellation: the same batch may publish later, so the generation
+    /// backs off and retries it whole, holding nothing back.
+    Transient,
+    /// The input itself cannot be planned (corrupt or invariant-violating
+    /// state, or a payload its index rejects), so retrying the same input
+    /// fails the same way. An entity whose planning fails like this is held
+    /// back ([`QueuePublisher::isolate`]) while the rest of its generation
+    /// publishes; a failure outside one entity's planning has no entity to
+    /// hold back and retries like a transient one.
+    Deterministic,
+}
+
+impl FailureKind {
+    /// Classifies `error` by its variant, never its message. Every variant
+    /// is named, so a new one cannot go unclassified.
+    pub(crate) fn of(error: &HelixDbError) -> Self {
+        match error {
+            HelixDbError::DatabaseClosed | HelixDbError::WriterFencedCommitOutcomeUnknown => {
+                Self::Fatal
+            }
+            HelixDbError::Storage(error)
+                if matches!(error.kind(), slatedb::ErrorKind::Closed(_)) =>
+            {
+                Self::Fatal
+            }
+            HelixDbError::Storage(_)
+            | HelixDbError::ObjectStore(_)
+            | HelixDbError::TransactionConflict(_)
+            | HelixDbError::RequestReadViewChanged
+            | HelixDbError::QueryDeadlineExceeded
+            | HelixDbError::QueryCancelledByReaderRetirement
+            | HelixDbError::StaleIndexGeneration { .. }
+            | HelixDbError::IndexBusy { .. }
+            | HelixDbError::IndexBackpressure { .. }
+            | HelixDbError::IndexBuildBlocked { .. }
+            | HelixDbError::IdentifierAllocationFailed { .. } => Self::Transient,
+            HelixDbError::Encoding(_)
+            | HelixDbError::InvalidNodeId(_)
+            | HelixDbError::NodeNotFound(_)
+            | HelixDbError::EdgeNotFound { .. }
+            | HelixDbError::Config(_)
+            | HelixDbError::IndexLifecycleUnavailable { .. }
+            | HelixDbError::SecondaryLifecycleSteppingRequiresDisabledMode
+            | HelixDbError::MigrationSteppingRequiresDisabledMode
+            | HelixDbError::ActiveTextMutationLimitExceeded { .. }
+            | HelixDbError::IndexOperationBatchTooLarge { .. }
+            | HelixDbError::InvalidIndexSourceData { .. }
+            | HelixDbError::InvalidIndexV2Model(_)
+            | HelixDbError::SecondaryIndexValue(_)
+            | HelixDbError::MigrationRequired { .. }
+            | HelixDbError::WriterMigrationRequired { .. }
+            | HelixDbError::UnsupportedIndexStorageVersion { .. }
+            | HelixDbError::IdentifierExhausted(_)
+            | HelixDbError::IndexCatalogCorruption(_)
+            | HelixDbError::InvalidVectorConfig(_)
+            | HelixDbError::InvalidVectorItem(_)
+            | HelixDbError::Query(_)
+            | HelixDbError::Planner(_)
+            | HelixDbError::InvalidQueryJson(_)
+            | HelixDbError::WriterModeRequired { .. }
+            | HelixDbError::ReaderModeRequired { .. }
+            | HelixDbError::IndexAlreadyExists(_)
+            | HelixDbError::IndexDefinitionConflict { .. }
+            | HelixDbError::IndexOperationNotFound { .. }
+            | HelixDbError::IndexOperationNotAbortable { .. }
+            | HelixDbError::IndexNotFound(_)
+            | HelixDbError::UniqueConstraintViolation { .. }
+            | HelixDbError::UnsupportedUniqueIndexValueType { .. }
+            | HelixDbError::InvalidDimension { .. }
+            | HelixDbError::InvalidVectorComponent { .. }
+            | HelixDbError::VectorComponentMagnitudeExceeded { .. }
+            | HelixDbError::ZeroNormCosineVector
+            | HelixDbError::LegacyZeroNormCosineVector { .. }
+            | HelixDbError::InvariantViolation(_) => Self::Deterministic,
+        }
+    }
 }
diff --git a/crates/db/src/index_lifecycle/queue/publication_tests.rs b/crates/db/src/index_lifecycle/queue/publication_tests.rs
index 6b57e97f3..81207c7c6 100644
--- a/crates/db/src/index_lifecycle/queue/publication_tests.rs
+++ b/crates/db/src/index_lifecycle/queue/publication_tests.rs
@@ -1554,27 +1554,32 @@ async fn an_active_namespace_without_metadata_fails_closed_and_writes_nothing()
         )
         .await
         .unwrap();
-    add_doc(&db, vec![2.0, 2.0], "c").await.unwrap();
+    let inserted = add_doc(&db, vec![2.0, 2.0], "c").await.unwrap();
     let keys = all_keys(&db).await;
     let rows = unpartitioned_vector_rows(&db).await;
-    let errors = publisher(&db)
-        .metrics()
-        .error_retries
-        .load(Ordering::Relaxed);
 
     // Only a build creates a missing namespace, so publication never
-    // recreates it over rows search can no longer reach.
+    // recreates it over rows search can no longer reach: planning the insert
+    // fails, which holds it back.
     assert_eq!(
         publisher(&db).publish_once(target).await.unwrap(),
-        PublicationOutcome::Retry
+        PublicationOutcome::Blocked
+    );
+    assert_eq!(
+        publisher(&db)
+            .blocked_entities()
+            .into_iter()
+            .map(|(_, entity)| entity.id.get())
+            .collect::<Vec<_>>(),
+        [inserted]
     );
     assert_eq!(
         publisher(&db)
             .metrics()
             .error_retries
             .load(Ordering::Relaxed),
-        errors + 1,
-        "a missing namespace is an error, not a conflict"
+        0,
+        "a missing namespace fails planning, not storage"
     );
     assert_eq!(
         all_keys(&db).await,
@@ -1616,31 +1621,36 @@ async fn contradict_metadata(db: &HelixDB, physical_index_id: u64) {
 }
 
 /// Asserts `target`'s next attempt fails closed on the contradicting metadata
-/// of namespace `physical_index_id`: an error rather than a conflict, which
-/// writes nothing, leaves its one queued operation, and retains no planning
-/// session.
+/// of namespace `physical_index_id`: planning fails, holding back `entity`,
+/// whose one queued operation stays queued, and the attempt writes nothing
+/// and retains no planning session.
 async fn assert_contradicting_metadata_fails_closed(
     db: &HelixDB,
     target: QueueTarget,
     physical_index_id: u64,
+    entity: u64,
 ) {
     let keys = all_keys(db).await;
     let rows = physical_rows(db, physical_index_id).await;
-    let errors = publisher(db)
-        .metrics()
-        .error_retries
-        .load(Ordering::Relaxed);
     assert_eq!(
         publisher(db).publish_once(target).await.unwrap(),
-        PublicationOutcome::Retry
+        PublicationOutcome::Blocked
+    );
+    assert_eq!(
+        publisher(db)
+            .blocked_entities()
+            .into_iter()
+            .map(|(_, held)| held.id.get())
+            .collect::<Vec<_>>(),
+        [entity]
     );
     assert_eq!(
         publisher(db)
             .metrics()
             .error_retries
             .load(Ordering::Relaxed),
-        errors + 1,
-        "contradicting metadata is an error, not a conflict"
+        0,
+        "contradicting metadata fails planning, not storage"
     );
     assert_eq!(
         all_keys(db).await,
@@ -1688,12 +1698,13 @@ async fn an_upsert_into_contradicting_metadata_fails_closed_and_writes_nothing()
         PublicationOutcome::Published { .. }
     ));
     contradict_metadata(&db, physical_index_id.get()).await;
-    add_doc(&db, vec![2.0, 2.0], "c").await.unwrap();
+    let inserted = add_doc(&db, vec![2.0, 2.0], "c").await.unwrap();
     assert!(publisher(&db)
         .planning_cache()
         .retained_publication(target)
         .is_some());
-    assert_contradicting_metadata_fails_closed(&db, target, physical_index_id.get()).await;
+    assert_contradicting_metadata_fails_closed(&db, target, physical_index_id.get(), inserted)
+        .await;
     db.close().await.unwrap();
 }
 
@@ -1718,7 +1729,7 @@ async fn a_removal_from_contradicting_metadata_fails_closed_and_writes_nothing()
     // The partition keeps another entity, so only the removal reads its
     // metadata; no reclamation runs.
     super::overlay_tests::delete(&db, removed).await;
-    assert_contradicting_metadata_fails_closed(&db, target, partition).await;
+    assert_contradicting_metadata_fails_closed(&db, target, partition, removed).await;
     db.close().await.unwrap();
 }
 
diff --git a/crates/db/src/index_lifecycle/vector.rs b/crates/db/src/index_lifecycle/vector.rs
index 3c3234459..d0ec1daeb 100644
--- a/crates/db/src/index_lifecycle/vector.rs
+++ b/crates/db/src/index_lifecycle/vector.rs
@@ -443,26 +443,29 @@ mod tests {
                     *generation,
                 ))
                 .await;
-            let publication::StagedEffects::Prefix { staged, .. } =
-                publication::stage_active_effects(
-                    db,
-                    transaction,
-                    &permit,
-                    handle,
-                    &effects,
-                    SearchIndexBackfillLimits::default().batch(),
-                    AcknowledgementOutput {
-                        operations: 0,
-                        bytes: 0,
-                    },
-                    &resources,
-                    cache_writes,
-                    None,
-                    std::num::NonZeroU64::MIN,
-                )
-                .await?
-            else {
-                panic!("every effect fits the default budget");
+            let staged = match publication::stage_active_effects(
+                db,
+                transaction,
+                &permit,
+                handle,
+                &effects,
+                SearchIndexBackfillLimits::default().batch(),
+                AcknowledgementOutput {
+                    operations: 0,
+                    bytes: 0,
+                },
+                &resources,
+                cache_writes,
+                None,
+                std::num::NonZeroU64::MIN,
+            )
+            .await?
+            {
+                publication::StagedEffects::Prefix { staged, .. } => staged,
+                publication::StagedEffects::NoneFits => {
+                    panic!("every effect fits the default budget")
+                }
+                publication::StagedEffects::Failed { error, .. } => return Err(error),
             };
             assert_eq!(staged, expected);
         }
diff --git a/crates/db/src/index_lifecycle/vector/publication.rs b/crates/db/src/index_lifecycle/vector/publication.rs
index f603cb579..6a5c0e074 100644
--- a/crates/db/src/index_lifecycle/vector/publication.rs
+++ b/crates/db/src/index_lifecycle/vector/publication.rs
@@ -94,6 +94,13 @@ pub(crate) enum StagedEffects {
     /// Not even the first effect fits beside the reserved output; nothing was
     /// staged.
     NoneFits,
+    /// Planning the effect at `position` failed with `error`, so the caller
+    /// can tell which entity failed. Every earlier effect is staged and the
+    /// failed one may be partly planned, so the transaction must not commit.
+    Failed {
+        position: usize,
+        error: HelixDbError,
+    },
 }
 
 /// Plans `effects` in order and stages the longest prefix that fits `limits`
@@ -106,6 +113,9 @@ pub(crate) enum StagedEffects {
 /// Planning reuses the session retained after the target's commit numbered
 /// `latest_commit` when that is still the retained one, and offers its own
 /// session for retention at `commit`, the number this attempt's commit takes.
+///
+/// An error planning one effect is [`StagedEffects::Failed`]; an error before
+/// any effect is planned is returned.
 #[allow(
     clippy::too_many_arguments,
     reason = "publication binds the exact storage, generation, budget, planner resources, cache effects, and session checkpoints"
@@ -227,7 +237,7 @@ async fn stage_with_distance<D: Distance>(
         .checkout_publication::<D>(permit, reuse.as_ref(), limits.max_input_bytes())
         .await;
     let mut staged = 0_usize;
-    for effect in effects {
+    for (position, effect) in effects.iter().enumerate() {
         let next = effect
             .replacement
             .as_ref()
@@ -235,13 +245,7 @@ async fn stage_with_distance<D: Distance>(
                 partition: replacement.partition().clone(),
                 vector: replacement.vector().to_vec(),
             });
-        let EntityPlanOutcome::Admitted {
-            vector_writes,
-            single_vector_output_bytes,
-            lifecycle_operations,
-            lifecycle_bytes,
-            ..
-        } = plan_and_apply::<D>(
+        let planned = match plan_and_apply::<D>(
             &planning,
             &recorder,
             transaction,
@@ -255,7 +259,18 @@ async fn stage_with_distance<D: Distance>(
             &accounting,
             &mut session,
         )
-        .await?
+        .await
+        {
+            Ok(planned) => planned,
+            Err(error) => return Ok(StagedEffects::Failed { position, error }),
+        };
+        let EntityPlanOutcome::Admitted {
+            vector_writes,
+            single_vector_output_bytes,
+            lifecycle_operations,
+            lifecycle_bytes,
+            ..
+        } = planned
         else {
             session.discard_entity();
             break;
diff --git a/crates/db/src/lib.rs b/crates/db/src/lib.rs
index fe9ecd2a3..671c86e11 100644
--- a/crates/db/src/lib.rs
+++ b/crates/db/src/lib.rs
@@ -3309,11 +3309,12 @@ impl HelixDB {
     /// publisher holds back, in ascending order.
     ///
     /// Each one has an operation that alone can never fit a publication under
-    /// the current limits; see [`BlockedIndexEntity`] for what that means for
-    /// writes and searches, including strong text searches that fail with
-    /// backpressure no publication clears. The list is the publisher's
-    /// process memory: it is empty on a reader and is rebuilt after a
-    /// restart as publication blocks again.
+    /// the current limits, or one whose planning failed deterministically;
+    /// see [`BlockedIndexEntity`] for what that means for writes and
+    /// searches, including strong text searches that fail with backpressure
+    /// no publication clears. The list is the publisher's process memory: it
+    /// is empty on a reader and is rebuilt after a restart as publication
+    /// blocks again.
     pub fn blocked_index_entities(&self) -> Vec<BlockedIndexEntity> {
         let Some(publisher) = &self.inner.index_queue_publisher else {
             return Vec::new();
diff --git a/crates/db/tests/production_support/queue_publication.rs b/crates/db/tests/production_support/queue_publication.rs
index 014e4f637..77aa46123 100644
--- a/crates/db/tests/production_support/queue_publication.rs
+++ b/crates/db/tests/production_support/queue_publication.rs
@@ -7,7 +7,7 @@
 //! fixture's indexes and is then reopened explicitly over the same store.
 //! Graph writes enqueue every operation through the production producer.
 //! Failure contracts stage one unobserved acknowledgement, corrupt one queue
-//! value, rewrite one namespace's metadata through the current codecs, hold
+//! value, rewrite or delete one namespace's metadata through the current codecs, hold
 //! the planning budget while a catalog change commits, fail WAL uploads, or
 //! fence the writer with a newer one; none introduces a row family or
 //! encoding.
@@ -951,9 +951,10 @@ enum Change {
     Insert([f32; 2]),
 }
 
-/// Proves publication fails closed, writing nothing, when a namespace's
-/// metadata disagrees with its definition or is missing, for both removals
-/// and upserts, and publishes once the metadata is restored.
+/// Proves publication fails closed when a namespace's metadata disagrees with
+/// its definition or is missing, for both removals and upserts: it writes
+/// nothing and holds back only the entity it planned, which a publisher
+/// restarted after the metadata is restored publishes.
 async fn inconsistent_namespace_metadata_fails_closed() {
     let db = open_explicit(
         "queue-publication-metadata",
@@ -1032,26 +1033,45 @@ async fn inconsistent_namespace_metadata_fails_closed() {
             publisher
                 .publish_once(target)
                 .await
-                .expect("inconsistent metadata retries"),
-            PublicationOutcome::Retry
+                .expect("inconsistent metadata holds its entity back"),
+            PublicationOutcome::Blocked
         );
         assert_eq!(
-            load(&publisher.metrics().error_retries),
+            load(&publisher.metrics().blocked_attempts),
             u64::try_from(errors + 1).expect("phase count fits u64"),
-            "inconsistent metadata is an error, not a conflict"
+            "inconsistent metadata fails planning, not storage"
         );
+        assert_eq!(load(&publisher.metrics().error_retries), 0);
+        assert_eq!(publisher.blocked_entity_count(), 1);
         assert_eq!(all_keys(&db).await, keys, "a failed attempt writes nothing");
         storage
             .put(&key, &original)
             .await
             .expect("metadata restores");
-        assert!(matches!(
+        assert_eq!(
             publisher
+                .publish_once(target)
+                .await
+                .expect("a held entity waits"),
+            PublicationOutcome::Stalled,
+            "only a write or a restart plans a held entity again"
+        );
+        let restarted = publisher_with_limits(&db, publisher.limits, publisher.text.limits);
+        assert!(matches!(
+            restarted
                 .publish_once(target)
                 .await
                 .expect("restored metadata publishes"),
             PublicationOutcome::Published { .. }
         ));
+        assert_eq!(
+            publisher
+                .publish_once(target)
+                .await
+                .expect("the drained queue reads empty"),
+            PublicationOutcome::Empty
+        );
+        assert_eq!(publisher.blocked_entity_count(), 0);
     }
     db.close().await.expect("metadata writer closes");
 }
diff --git a/crates/db/tests/production_text_correctness_regressions.rs b/crates/db/tests/production_text_correctness_regressions.rs
index 77fb4bcdd..4d6f4b5c9 100644
--- a/crates/db/tests/production_text_correctness_regressions.rs
+++ b/crates/db/tests/production_text_correctness_regressions.rs
@@ -448,10 +448,11 @@ async fn wait_published(db: &HelixDB) {
     }
 }
 
-/// Waits until automatic publication has failed more often than `before`.
-async fn wait_publication_failure(db: &HelixDB, before: u64) {
+/// Waits until automatic publication has held an entity back more often than
+/// `before`.
+async fn wait_publication_blocked(db: &HelixDB, before: u64) {
     let started = Instant::now();
-    while db.index_operation_queue_stats().publication_error_retries <= before {
+    while db.index_operation_queue_stats().blocked_attempts <= before {
         assert!(
             started.elapsed() < OPERATION_TIMEOUT,
             "queued publication was never attempted"
@@ -1342,15 +1343,29 @@ async fn active_node_insert_rejects_nonempty_root_without_corpus_statistics() {
         .migration_parity_v2_state()
         .await
         .expect("damaged node evidence reads");
-    let failures = db.index_operation_queue_stats().publication_error_retries;
+    let stats = db.index_operation_queue_stats();
     // Producers never read statistics; the damaged root fails closed when
-    // publication applies the queued insert.
-    insert_node(&db, "rejectednodeappendonly").await;
-    wait_publication_failure(&db, failures).await;
+    // publication plans the queued insert, a deterministic failure that holds
+    // the node back instead of retrying it.
+    let rejected = insert_node(&db, "rejectednodeappendonly").await;
+    wait_publication_blocked(&db, stats.blocked_attempts).await;
+    let held = db.index_operation_queue_stats();
     assert_eq!(
-        db.index_operation_queue_stats().pending_operations,
-        1,
-        "an operation that cannot be published is never acknowledged"
+        (
+            held.pending_operations,
+            held.blocked_entities,
+            held.publication_error_retries
+        ),
+        (1, 1, stats.publication_error_retries),
+        "an operation that cannot be published is held back, never acknowledged or retried"
+    );
+    assert_eq!(
+        db.blocked_index_entities()
+            .iter()
+            .map(|blocked| blocked.id.get())
+            .collect::<Vec<_>>(),
+        [rejected],
+        "only the damaged node is held back"
     );
     let mut after = db
         .migration_parity_v2_state()
@@ -1419,15 +1434,29 @@ async fn active_edge_insert_rejects_nonempty_root_without_corpus_statistics() {
         .migration_parity_v2_state()
         .await
         .expect("damaged edge evidence reads");
-    let failures = db.index_operation_queue_stats().publication_error_retries;
+    let stats = db.index_operation_queue_stats();
     // Producers never read statistics; the damaged root fails closed when
-    // publication applies the queued insert.
-    insert_edge(&db, from, to, "rejectededgeappendonly").await;
-    wait_publication_failure(&db, failures).await;
+    // publication plans the queued insert, a deterministic failure that holds
+    // the edge back instead of retrying it.
+    let rejected = insert_edge(&db, from, to, "rejectededgeappendonly").await;
+    wait_publication_blocked(&db, stats.blocked_attempts).await;
+    let held = db.index_operation_queue_stats();
     assert_eq!(
-        db.index_operation_queue_stats().pending_operations,
-        1,
-        "an operation that cannot be published is never acknowledged"
+        (
+            held.pending_operations,
+            held.blocked_entities,
+            held.publication_error_retries
+        ),
+        (1, 1, stats.publication_error_retries),
+        "an operation that cannot be published is held back, never acknowledged or retried"
+    );
+    assert_eq!(
+        db.blocked_index_entities()
+            .iter()
+            .map(|blocked| blocked.id.get())
+            .collect::<Vec<_>>(),
+        [rejected],
+        "only the damaged edge is held back"
     );
     let mut after = db
         .migration_parity_v2_state()
diff --git a/docs/database/helix-db/query-guides/troubleshooting.mdx b/docs/database/helix-db/query-guides/troubleshooting.mdx
index d9e21b71b..0f6176a6e 100644
--- a/docs/database/helix-db/query-guides/troubleshooting.mdx
+++ b/docs/database/helix-db/query-guides/troubleshooting.mdx
@@ -99,6 +99,12 @@ error naming each one when it is held back. Raise the limits again. Rewriting or
 deleting a document also clears it once replacing or removing its published version
 fits the lowered limits.
 
+The worker also holds back a single vector or text entry whose update fails the same
+way every time it is applied, for example on damaged index data, and keeps publishing
+the rest of the index. It counts under `blocked_index_entity_count` too, and the logged
+error names the entry and the failure. Writing the entry again retries it, and so does
+restarting the writer.
+
 ## Request returns HTTP 429 (Helix Cloud)
 
 `rate_limited` means the shared request bucket for the Cloud database has no

From bf11b6b2f5e63278db6301f6386fd1d941faa5b3 Mon Sep 17 00:00:00 2001
From: xav-db <xav@helix-db.com>
Date: Sun, 4 Oct 2026 13:56:57 +0100
Subject: [PATCH 3/9] ci: shift vector coverage line anchors past the self-link
 fix

The self-link fix moved lines in the HNSW mutation module, so the
line-anchored vector coverage dispositions and exclusions there now
name the same source lines at their new positions.
---
 .../db-production-coverage-dispositions.json  | 134 +++++++++---------
 .../db-production-coverage-exclusions.json    |   6 +-
 2 files changed, 70 insertions(+), 70 deletions(-)

diff --git a/scripts/db-production-coverage-dispositions.json b/scripts/db-production-coverage-dispositions.json
index ba4e624c3..977896fa9 100644
--- a/scripts/db-production-coverage-dispositions.json
+++ b/scripts/db-production-coverage-dispositions.json
@@ -92,85 +92,85 @@
       484,
       498,
       499,
-      526,
-      533,
-      537,
-      540,
+      529,
+      536,
       541,
-      554,
-      555,
-      605,
+      544,
+      545,
+      558,
+      559,
       609,
       613,
-      665,
+      617,
       669,
       673,
-      726,
-      740,
-      780,
-      799,
-      804,
-      893,
-      929,
-      938,
-      949,
+      677,
+      730,
+      744,
+      784,
+      803,
+      808,
+      898,
+      934,
+      943,
       954,
-      966,
-      1007,
-      1015,
-      1039,
-      1043,
-      1096,
-      1099,
-      1100,
-      1115,
-      1122,
-      1126,
-      1312,
-      1349,
+      959,
+      971,
+      1018,
+      1027,
+      1051,
+      1055,
+      1113,
+      1116,
+      1117,
+      1132,
+      1139,
+      1143,
+      1329,
       1366,
-      1473,
-      1479,
-      1480,
-      1495,
-      1569,
-      1583,
-      1592,
-      1601,
+      1383,
+      1490,
+      1496,
+      1497,
+      1512,
+      1586,
+      1600,
       1609,
-      1616,
-      1623,
+      1618,
+      1626,
       1633,
-      1663,
-      1692,
-      1730,
-      1734,
-      1822,
-      1862,
-      1932,
-      2022,
-      2043,
-      2049,
-      2084,
-      2086,
-      2091,
-      2100,
-      2110,
-      2125,
-      2127,
+      1640,
+      1650,
+      1680,
+      1709,
+      1748,
+      1752,
+      1840,
+      1880,
+      1950,
+      2040,
+      2061,
+      2067,
+      2102,
+      2104,
+      2109,
+      2118,
       2128,
-      2129,
-      2130,
-      2131,
-      2134,
-      2135,
-      2142,
+      2143,
+      2145,
+      2146,
+      2147,
+      2148,
       2149,
-      2158,
+      2152,
+      2153,
+      2160,
       2167,
-      2693,
-      3691,
-      3941
+      2176,
+      2185,
+      2711,
+      3709,
+      3959
     ],
     "classification": "named-test",
     "evidence": "vector_mutation_cache_exercises_closed_state_transitions",
diff --git a/scripts/db-production-coverage-exclusions.json b/scripts/db-production-coverage-exclusions.json
index b8e75ecda..9f96bcd64 100644
--- a/scripts/db-production-coverage-exclusions.json
+++ b/scripts/db-production-coverage-exclusions.json
@@ -601,19 +601,19 @@
   },
   {
     "path": "crates/db/src/search/vector/hnsw/mutation/mod.rs",
-    "line": 3720,
+    "line": 3738,
     "reason": "Only shrink_to evicts without a transaction, and it runs only on sessions the lifecycle driver retained. The driver retains a session only when it holds no unflushed rows, after flush_all has written every dirty neighbor for the step, so this arm cannot meet a dirty neighbor row.",
     "evidence": "vector_build_cache_reuses_only_exact_committed_checkpoints"
   },
   {
     "path": "crates/db/src/search/vector/hnsw/mutation/mod.rs",
-    "line": 3816,
+    "line": 3834,
     "reason": "This error value belongs to the platform-gated u64-to-usize payload accounting failure that cannot occur on supported 64-bit production targets.",
     "evidence": "vector_build_cache_reuses_only_exact_committed_checkpoints and db-vector-production-coverage architecture matrix"
   },
   {
     "path": "crates/db/src/search/vector/hnsw/mutation/mod.rs",
-    "line": 3819,
+    "line": 3837,
     "reason": "This closing edge belongs to the platform-gated u64-to-usize payload accounting failure that cannot occur on supported 64-bit production targets.",
     "evidence": "vector_build_cache_reuses_only_exact_committed_checkpoints and db-vector-production-coverage architecture matrix"
   },

From b32d108e94b9b5d29d12383bcca613dbc9f37bcf Mon Sep 17 00:00:00 2001
From: xav-db <xav@helix-db.com>
Date: Sun, 4 Oct 2026 15:08:19 +0100
Subject: [PATCH 4/9] test(db): pin each HNSW insert guard against the
 inserting node

Removing the greedy descent's guard or either beam-root guard left every
vector test passing. Stale metadata naming the inserting node now has an
end-to-end case, and each guard a direct case that first shows the same
traversal reaching the node when another node inserts.
---
 .../vector/hnsw/mutation/locator_tests.rs     | 140 ++++++++++++++++++
 1 file changed, 140 insertions(+)

diff --git a/crates/db/src/search/vector/hnsw/mutation/locator_tests.rs b/crates/db/src/search/vector/hnsw/mutation/locator_tests.rs
index bdc1f5c5a..83503c418 100644
--- a/crates/db/src/search/vector/hnsw/mutation/locator_tests.rs
+++ b/crates/db/src/search/vector/hnsw/mutation/locator_tests.rs
@@ -21,6 +21,11 @@
 //! [`damaged_graphs_never_gain_a_self_link`] start from rows that link their
 //! own node, as damage from released versions can leave them, and pin that
 //! every operation still plans and none links a node to itself.
+//!
+//! [`an_entry_point_naming_the_inserting_node_never_roots_its_insert`] and
+//! [`insert_traversals_never_reach_the_inserting_node`] pin each traversal
+//! guard against the inserting node, whose item an insert stages first: one
+//! through stale metadata, one per guard.
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
@@ -970,3 +975,138 @@ async fn damaged_graphs_never_gain_a_self_link() {
         damaged_run::<Cosine>(0, planner).await;
     }
 }
+
+/// Commits metadata naming [`GHOST`], which has no item, as the entry point,
+/// as a stale entry can, then inserts [`GHOST`] at layer 0.
+async fn stale_entry_point_run<D: Distance>() {
+    let mut rng = StdRng::seed_from_u64(2);
+    let name = format!("stale-entry-point-{}", D::name());
+    let (db, index) = create::<D>(&name).await;
+    let mut points = build(&db, &index, &mut rng, 60).await;
+    let txn = db.begin(IsolationLevel::Snapshot).await.unwrap();
+    let measured = MeasuredVectorTransaction::new(&txn);
+    let mut metadata = index.get_metadata(&measured).await.unwrap().unwrap();
+    assert!(metadata.max_layer > 0, "{name}: the build has upper layers");
+    metadata.entry_point = Some(GHOST);
+    index.update_metadata(&measured, &metadata).await.unwrap();
+    txn.commit().await.unwrap();
+
+    // The item the insert stages makes the entry point look live, so no
+    // repair replaces it before the traversals start there.
+    let point = random_point(&mut rng);
+    let txn = db.begin(IsolationLevel::Snapshot).await.unwrap();
+    let measured = MeasuredVectorTransaction::new(&txn);
+    index
+        .stage_upsert_at_layer(&measured, GHOST, &point, 0)
+        .await
+        .unwrap_or_else(|error| panic!("{name}: inserting {GHOST}: {error}"));
+    txn.commit().await.unwrap();
+    points.insert(GHOST, point);
+    let graph = Graph::read(&index, db.snapshot().await.unwrap().as_ref()).await;
+    assert_eq!(self_links(&graph), BTreeSet::new(), "{name}");
+    assert!(
+        graph
+            .links
+            .iter()
+            .any(|(layer, source, _)| (*layer, *source) == (0, GHOST)),
+        "{name}: {GHOST} gains neighbors"
+    );
+    graph.assert_valid(&points.keys().copied().collect(), &name);
+}
+
+/// Metadata naming the inserting node as its entry point, which the item the
+/// insert stages makes look live, never roots that insert's beam at the node:
+/// it links real neighbors and never itself.
+#[tokio::test]
+async fn an_entry_point_naming_the_inserting_node_never_roots_its_insert() {
+    stale_entry_point_run::<Euclidean>().await;
+    stale_entry_point_run::<Cosine>().await;
+}
+
+/// Runs each traversal an insert starts with on a built graph, once for
+/// another inserting node, which reaches `target`, and once for `target`.
+async fn inserting_node_run<D: Distance>() {
+    let mut rng = StdRng::seed_from_u64(3);
+    let name = format!("inserting-node-{}", D::name());
+    let (db, index) = create::<D>(&name).await;
+    build(&db, &index, &mut rng, 60).await;
+    let graph = Graph::read(&index, db.snapshot().await.unwrap().as_ref()).await;
+    let txn = db.begin(IsolationLevel::Snapshot).await.unwrap();
+    let measured = MeasuredVectorTransaction::new(&txn);
+    let mut cache = MutationOpCache::<D>::default();
+
+    // An upper-layer link the greedy descent toward its target's own item
+    // follows when another node inserts.
+    let mut reached = None;
+    for (layer, source, target) in graph.links.iter().copied().filter(|link| link.0 > 0) {
+        let query = index
+            .get_item_for_layer_cached(&measured, 0, target, &mut cache)
+            .await
+            .unwrap()
+            .unwrap();
+        let descended = index
+            .search_layer_greedy_for_mutation(&measured, &query, source, layer, GHOST, &mut cache)
+            .await
+            .unwrap();
+        if descended == target {
+            reached = Some((layer, source, target, query));
+            break;
+        }
+    }
+    let Some((layer, source, target, query)) = reached else {
+        panic!("{name}: a greedy descent reaches a linked node");
+    };
+    assert_ne!(
+        index
+            .search_layer_greedy_for_mutation(&measured, &query, source, layer, target, &mut cache)
+            .await
+            .unwrap(),
+        target,
+        "{name}: the descent never moves to the inserting node"
+    );
+
+    // An entry point naming the inserting node never roots its beam.
+    let root = async |entry_point, inserting_node_id, cache: &mut MutationOpCache<D>| {
+        index
+            .resolve_beam_entry_point_for_insert(
+                &measured,
+                entry_point,
+                0,
+                inserting_node_id,
+                cache,
+            )
+            .await
+            .unwrap()
+            .map(|(node, _)| node)
+    };
+    assert_eq!(root(target, GHOST, &mut cache).await, Some(target));
+    assert_ne!(
+        root(target, target, &mut cache).await,
+        Some(target),
+        "{name}: the inserting node is never its own root"
+    );
+
+    // Nor does the best entry candidate, which replaces a missing entry
+    // point: it is the only replacement tried.
+    let (best, _) = index
+        .find_best_entry_candidate(&measured)
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(root(GHOST, GHOST + 1, &mut cache).await, Some(best));
+    assert_eq!(
+        root(GHOST, best, &mut cache).await,
+        None,
+        "{name}: the best candidate is never the root of its own insert"
+    );
+}
+
+/// No traversal an insert starts reaches the inserting node, whose item the
+/// insert stages first: the greedy upper-layer descent never moves to it
+/// even when a row links it, and neither an entry point naming it nor the
+/// best entry candidate being it roots the beam.
+#[tokio::test]
+async fn insert_traversals_never_reach_the_inserting_node() {
+    inserting_node_run::<Euclidean>().await;
+    inserting_node_run::<Cosine>().await;
+}

From 22931dae900d6ff8985de3d01b738cfe071ee53a Mon Sep 17 00:00:00 2001
From: xav-db <xav@helix-db.com>
Date: Sun, 4 Oct 2026 15:08:19 +0100
Subject: [PATCH 5/9] test(db): pin which vector entity a mid-batch planning
 failure holds back

Every isolation case planned the failing entity first, so holding back
the batch's first entity instead of the failing one went unnoticed. The
failing update now follows two inserts that plan in the same batch.
---
 .../index_lifecycle/queue/isolation_tests.rs  | 57 +++++++++++++++++++
 1 file changed, 57 insertions(+)

diff --git a/crates/db/src/index_lifecycle/queue/isolation_tests.rs b/crates/db/src/index_lifecycle/queue/isolation_tests.rs
index a7ed0e20a..93c2dd2b9 100644
--- a/crates/db/src/index_lifecycle/queue/isolation_tests.rs
+++ b/crates/db/src/index_lifecycle/queue/isolation_tests.rs
@@ -272,6 +272,63 @@ async fn a_vector_entity_that_fails_to_plan_is_held_back_alone() {
     db.close().await.unwrap();
 }
 
+/// A failure after other entities of the batch planned holds back the
+/// failing entity, not the batch's first: the attempt commits nothing, and
+/// the entities planned before it publish from the next attempt.
+#[tokio::test]
+async fn a_vector_entity_failing_after_planned_ones_is_the_one_held_back() {
+    let db = open(
+        "isolate-vector-position",
+        Arc::new(InMemory::new()),
+        queued(tuning()),
+    )
+    .await;
+    install_vector(&db, None).await;
+    let failing = add_doc(&db, vec![0.0, 0.0], "doc").await.unwrap();
+    let target = target(&db, QueueFamily::Vector).await;
+    settle(&db, target).await;
+    let first = add_doc(&db, vec![1.0, 0.0], "doc").await.unwrap();
+    let second = add_doc(&db, vec![2.0, 0.0], "doc").await.unwrap();
+    set(&db, failing, "embedding", vec![5.0_f32, 5.0].into()).await;
+    assert_eq!(
+        queued_ids(&db, QueueFamily::Vector).await,
+        [first, second, failing],
+        "one batch plans the update after both inserts"
+    );
+    let corrupt = newest(&db, QueueFamily::Vector, failing).await;
+    inject(&db, corrupt, Some(InjectedPlanningFailure::Corrupt));
+    assert_eq!(
+        publisher(&db).publish_once(target).await.unwrap(),
+        PublicationOutcome::Blocked
+    );
+    assert_eq!(held(&db), [failing]);
+    assert_eq!(
+        queued_ids(&db, QueueFamily::Vector).await,
+        [first, second, failing],
+        "the failed attempt commits nothing"
+    );
+    assert_eq!(
+        settle(&db, target).await,
+        [
+            PublicationOutcome::Published {
+                operations: 2,
+                entities: 2
+            },
+            PublicationOutcome::Stalled,
+        ]
+    );
+    assert_eq!(held(&db), [failing]);
+    assert_eq!(queued_ids(&db, QueueFamily::Vector).await, [failing]);
+    assert_eq!(published_nearest(&db, [1.0, 0.0]).await, Some(first));
+    assert_eq!(published_nearest(&db, [2.0, 0.0]).await, Some(second));
+    assert_eq!(
+        published_nearest(&db, [0.0, 0.0]).await,
+        Some(failing),
+        "the held update is not published"
+    );
+    db.close().await.unwrap();
+}
+
 /// A text epoch fails as a whole, so its entity ceiling halves until the
 /// failing entity publishes alone and is held back; every other entity
 /// publishes, and a newer write to it publishes it.

From 3099a264a831047d4a813219667ca07fdd148f16 Mon Sep 17 00:00:00 2001
From: xav-db <xav@helix-db.com>
Date: Sun, 4 Oct 2026 15:08:29 +0100
Subject: [PATCH 6/9] fix(db): retry entities held back after failed planning

A failure outside an entity's own input, such as missing namespace
metadata, held back every queued entity one immediate attempt at a time,
and nothing planned them again until a write or a restart, which a held
delete never gets. A failure hold now repairs its entity again once
MAX_STALLED_WAIT passes, a stalled generation wakes at the first such
retry, and from the second hold in a row without a publication the next
attempt backs off. The docs now say a held entity's writes keep using
the index's shared queue capacity, and llms-full.txt is regenerated.
---
 .../index_lifecycle/queue/isolation_tests.rs  | 159 ++++++++++++++-
 crates/db/src/index_lifecycle/queue/mod.rs    |  14 +-
 .../src/index_lifecycle/queue/publication.rs  | 186 +++++++++++++-----
 .../production_support/queue_publication.rs   |  35 +++-
 .../helix-db/query-guides/troubleshooting.mdx |   7 +-
 docs/llms-full.txt                            |   9 +
 6 files changed, 347 insertions(+), 63 deletions(-)

diff --git a/crates/db/src/index_lifecycle/queue/isolation_tests.rs b/crates/db/src/index_lifecycle/queue/isolation_tests.rs
index 93c2dd2b9..acc077d15 100644
--- a/crates/db/src/index_lifecycle/queue/isolation_tests.rs
+++ b/crates/db/src/index_lifecycle/queue/isolation_tests.rs
@@ -9,9 +9,11 @@
 //!
 //! [`PublicationHooks::planning_failures`]: super::publication::test_hooks::PublicationHooks
 
+use std::collections::HashSet;
 use std::num::NonZeroU64;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
+use std::time::{Duration, Instant};
 
 use helix_ast::{
     batch, graph::NodeRef, query::QueryRequest, query::SearchConsistency, traversal,
@@ -22,11 +24,13 @@ use slatedb::object_store::ObjectStore;
 
 use super::overlay_tests::{text_search, vector_search, write};
 use super::publication::test_hooks::InjectedPlanningFailure;
-use super::publication::{FailureKind, PublicationOutcome};
+use super::publication::{FailureKind, NextTarget, PublicationOutcome, QueuePublisher};
 use super::publication_tests::{install_vector, publisher};
-use super::tests::{add_doc, install_vector_and_text, open, queue, queued, target};
+use super::tests::{
+    add_doc, install_vector_and_text, open, publisher_with_limits, queue, queued, target,
+};
 use super::QueueTarget;
-use crate::config::{IndexOperationQueueTuning, TextIndexDefinition};
+use crate::config::{DbConfig, IndexOperationQueueTuning, TextIndexDefinition};
 use crate::encoding::v2::values::indexes::operation_queue::{QueueFamily, QueuedOperationId};
 use crate::error::HelixDbError;
 use crate::index_lifecycle::ValidatedDynamicIndexDefinition;
@@ -88,7 +92,15 @@ async fn newest(db: &HelixDB, family: QueueFamily, id: u64) -> QueuedOperationId
 }
 
 fn inject(db: &HelixDB, operation: QueuedOperationId, failure: Option<InjectedPlanningFailure>) {
-    let mut failures = publisher(db).hooks().planning_failures.lock();
+    inject_into(publisher(db), operation, failure);
+}
+
+fn inject_into(
+    publisher: &QueuePublisher,
+    operation: QueuedOperationId,
+    failure: Option<InjectedPlanningFailure>,
+) {
+    let mut failures = publisher.hooks().planning_failures.lock();
     match failure {
         Some(failure) => failures.insert(operation, failure),
         None => failures.remove(&operation),
@@ -329,6 +341,145 @@ async fn a_vector_entity_failing_after_planned_ones_is_the_one_held_back() {
     db.close().await.unwrap();
 }
 
+/// An entity whose planning failed is planned again once its retry is due,
+/// without a write: a stalled generation waits only until that retry, an
+/// entity that still fails is held back again, and one whose failure was
+/// repaired publishes.
+#[tokio::test]
+async fn a_failed_entity_is_planned_again_once_its_retry_is_due() {
+    let db = open("isolate-retry", Arc::new(InMemory::new()), queued(tuning())).await;
+    install_vector(&db, None).await;
+    let failing = add_doc(&db, vec![1.0, 0.0], "doc").await.unwrap();
+    let target = target(&db, QueueFamily::Vector).await;
+    settle(&db, target).await;
+    // The writer's own publisher never schedules in tests; one with the same
+    // limits does.
+    let defaults = DbConfig::new().search_index_backfill();
+    let scheduler = publisher_with_limits(&db, defaults.batch(), defaults.active_text_mutation());
+    set(&db, failing, "embedding", vec![5.0_f32, 5.0].into()).await;
+    let corrupt = newest(&db, QueueFamily::Vector, failing).await;
+    inject_into(&scheduler, corrupt, Some(InjectedPlanningFailure::Corrupt));
+    let blocked =
+        |scheduler: &QueuePublisher| scheduler.metrics().blocked_attempts.load(Ordering::Relaxed);
+    assert_eq!(
+        scheduler.publish_once(target).await.unwrap(),
+        PublicationOutcome::Blocked
+    );
+    let stalled = Instant::now();
+    assert_eq!(
+        scheduler.publish_once(target).await.unwrap(),
+        PublicationOutcome::Stalled,
+        "the retry is not due yet"
+    );
+    let none = HashSet::new();
+    let NextTarget::Delayed(deadline) = scheduler.next_target(&none, stalled) else {
+        panic!("a held entity waits for its retry");
+    };
+    assert!(
+        deadline < stalled + Duration::from_secs(60),
+        "the stall ends when the retry is due, not a full wait after the stall"
+    );
+    assert_eq!(
+        scheduler.next_target(&none, deadline),
+        NextTarget::Ready(target)
+    );
+
+    // Still failing when due: planned once more and held back again.
+    scheduler.make_failed_retries_due();
+    assert_eq!(
+        scheduler.publish_once(target).await.unwrap(),
+        PublicationOutcome::Blocked
+    );
+    assert_eq!(blocked(&scheduler), 2);
+    assert_eq!(
+        scheduler.publish_once(target).await.unwrap(),
+        PublicationOutcome::Stalled
+    );
+    assert_eq!(blocked(&scheduler), 2, "a failed retry waits for the next");
+
+    // Once what failed is repaired, the due retry publishes it.
+    inject_into(&scheduler, corrupt, None);
+    scheduler.make_failed_retries_due();
+    assert_eq!(
+        scheduler.publish_once(target).await.unwrap(),
+        PublicationOutcome::Published {
+            operations: 1,
+            entities: 1
+        }
+    );
+    assert!(scheduler.blocked_entities().is_empty());
+    assert!(queued_ids(&db, QueueFamily::Vector).await.is_empty());
+    assert_eq!(published_nearest(&db, [5.0, 5.0]).await, Some(failing));
+    db.close().await.unwrap();
+}
+
+/// A second entity failing to plan with no publication between suggests a
+/// failure that is not the entities' own, such as a partition's missing
+/// metadata, so the next attempt backs off instead of holding back the queue
+/// one immediate attempt at a time; a publication ends the backoff.
+#[tokio::test]
+async fn consecutive_planning_failures_back_off() {
+    let db = open(
+        "isolate-backoff",
+        Arc::new(InMemory::new()),
+        queued(tuning()),
+    )
+    .await;
+    install_vector(&db, None).await;
+    let mut ids = Vec::new();
+    for x in 0..3_u8 {
+        ids.push(add_doc(&db, vec![f32::from(x), 0.0], "doc").await.unwrap());
+    }
+    let target = target(&db, QueueFamily::Vector).await;
+    let defaults = DbConfig::new().search_index_backfill();
+    let scheduler = publisher_with_limits(&db, defaults.batch(), defaults.active_text_mutation());
+    for id in &ids[..2] {
+        let corrupt = newest(&db, QueueFamily::Vector, *id).await;
+        inject_into(&scheduler, corrupt, Some(InjectedPlanningFailure::Corrupt));
+    }
+    let none = HashSet::new();
+    assert_eq!(
+        scheduler.publish_once(target).await.unwrap(),
+        PublicationOutcome::Blocked
+    );
+    assert_eq!(
+        scheduler.next_target(&none, Instant::now()),
+        NextTarget::Ready(target),
+        "one failure holds back only its entity"
+    );
+    assert_eq!(
+        scheduler.publish_once(target).await.unwrap(),
+        PublicationOutcome::Blocked
+    );
+    let NextTarget::Delayed(retry) = scheduler.next_target(&none, Instant::now()) else {
+        panic!("a second failure in a row backs off");
+    };
+    assert_eq!(
+        scheduler.next_target(&none, retry),
+        NextTarget::Ready(target)
+    );
+    assert_eq!(
+        scheduler.publish_once(target).await.unwrap(),
+        PublicationOutcome::Published {
+            operations: 1,
+            entities: 1
+        }
+    );
+    assert_eq!(
+        scheduler.next_target(&none, Instant::now()),
+        NextTarget::Ready(target),
+        "a publication ends the backoff"
+    );
+    let mut blocked = scheduler
+        .blocked_entities()
+        .into_iter()
+        .map(|(_, entity)| entity.id.get())
+        .collect::<Vec<_>>();
+    blocked.sort_unstable();
+    assert_eq!(blocked, ids[..2]);
+    db.close().await.unwrap();
+}
+
 /// A text epoch fails as a whole, so its entity ceiling halves until the
 /// failing entity publishes alone and is held back; every other entity
 /// publishes, and a newer write to it publishes it.
diff --git a/crates/db/src/index_lifecycle/queue/mod.rs b/crates/db/src/index_lifecycle/queue/mod.rs
index 02405a5bf..01a458af6 100644
--- a/crates/db/src/index_lifecycle/queue/mod.rs
+++ b/crates/db/src/index_lifecycle/queue/mod.rs
@@ -170,9 +170,17 @@ pub struct IndexOperationQueueStats {
 /// from its published state and planning that change succeeds: removing a
 /// document published under larger limits, or relinking a deleted vector's
 /// neighbors, can exceed the lowered limits too. Raising the limits again
-/// lets it publish. An entity held back after its planning failed is planned
-/// again only on a later write to it or once the writer restarts, so after
-/// repairing what failed, write it again or restart the writer.
+/// lets it publish. An entity held back after its planning failed is also
+/// planned again about once a minute without a write, so it publishes on its
+/// own once what failed is repaired, for example restored metadata; a held
+/// delete is never written again, and this is what publishes it.
+///
+/// Its queued operations, and each later write to it, keep counting toward
+/// its index's retained-byte limit until it publishes, a limit the whole
+/// index shares. An entity written over and over while its planning keeps
+/// failing, for example on damaged index rows of its own, therefore fills
+/// that limit, and then writes to every entity of the index fail with
+/// `index_backpressure`. Stop writing it until it publishes.
 ///
 /// Its queued text still counts toward the pending text a strong text search
 /// may analyze in its partition (one text publication's analysis budget),
diff --git a/crates/db/src/index_lifecycle/queue/publication.rs b/crates/db/src/index_lifecycle/queue/publication.rs
index e4b0322eb..d310284e3 100644
--- a/crates/db/src/index_lifecycle/queue/publication.rs
+++ b/crates/db/src/index_lifecycle/queue/publication.rs
@@ -26,8 +26,9 @@
 //!    acknowledgement: vectors through the build planner
 //!    ([`crate::index_lifecycle::vector::publication`]), text as one epoch.
 //!    When planning fails, [`FailureKind`] decides: an entity whose planning
-//!    fails deterministically is held back like one that cannot fit, and the
-//!    rest publish from the next attempt; anything else retries the batch.
+//!    fails deterministically is held back like one that cannot fit, but is
+//!    also retried on a timer, and the rest publish from the next attempt;
+//!    anything else retries the batch.
 //! 7. Stage one acknowledgement naming exactly the published IDs.
 //! 8. Commit through the vector cache's commit fence, release accounting and
 //!    retain the rest of the queue, then retire emptied partition caches and
@@ -83,10 +84,12 @@ const MAX_BACKOFF: Duration = Duration::from_secs(5);
 /// long that work waits once it becomes publishable.
 const MAX_DEFERRED_BACKOFF: Duration = Duration::from_secs(1);
 /// Longest a [`PublicationOutcome::Stalled`] generation waits without new
-/// work. Only a new operation can give it a publishable entity, and one
-/// makes it eligible at once; this bounds how long a retirement, which only
-/// an attempt discovers, leaves its held operations charged before they are
-/// discarded.
+/// work, and how long an entity whose planning failed waits before it is
+/// planned again ([`HeldEntity::Failed`]). A new operation, or a failed
+/// entity's due retry, makes a stalled generation eligible at once; this
+/// bounds how long a retirement, which only an attempt discovers, leaves its
+/// held operations charged before they are discarded, and how soon a failed
+/// entity publishes once what failed is repaired.
 const MAX_STALLED_WAIT: Duration = Duration::from_secs(60);
 /// Most operation IDs one discard transaction acknowledges (about a 1 MiB
 /// map operand); the operand and output bounds may lower it further.
@@ -112,8 +115,10 @@ pub(crate) enum PublicationOutcome {
     Trimmed,
     /// One operation's effect and acknowledgement cannot fit an output budget,
     /// or planning one entity failed deterministically, so its entity is held
-    /// back until a later operation supersedes it; retry immediately with the
-    /// generation's other entities.
+    /// back until a later operation supersedes it (or, after a failure, until
+    /// its retry is due); retry immediately with the generation's other
+    /// entities, or after backoff once two entities in a row failed to plan
+    /// without a publication between them.
     Blocked,
     /// Every queued entity is held back after blocking; nothing was attempted.
     /// The generation waits for a new operation rather than retrying on a
@@ -164,6 +169,12 @@ struct TargetSchedule {
     eligibility: Eligibility,
     /// Consecutive attempts without progress.
     failures: u32,
+    /// Entities held back after failing to plan since the generation last
+    /// published. A failure outside one entity's input, such as a partition's
+    /// missing metadata, fails every entity in turn, so from the second such
+    /// hold on, the next attempt backs off rather than holding back the whole
+    /// queue one immediate attempt at a time.
+    failed_holds: u32,
     /// Entities held back because a lone operation of theirs could not fit a
     /// publication or failed to plan, or draining after their repair. Process
     /// memory only: a restarted publisher rediscovers blocked entities by
@@ -209,6 +220,11 @@ struct TargetSchedule {
 /// last one is acknowledged. Strong searches overlay every queued operation
 /// and never observe it; eventual searches that do not reach the entity
 /// within their budget can.
+///
+/// An entity whose planning failed is also repaired, at full width, once its
+/// retry is due, without a newer operation: what failed may be repaired by
+/// then, for example restored metadata, and a held delete is never written
+/// again.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum HeldEntity {
     /// Skipped until an operation newer than `through` is queued.
@@ -216,6 +232,15 @@ pub(crate) enum HeldEntity {
         /// Newest operation of the entity known not to publish.
         through: QueuedOperationId,
     },
+    /// Planning failed deterministically ([`QueuePublisher::isolate`]):
+    /// skipped until an operation newer than `through` is queued, or until
+    /// `retry`.
+    Failed {
+        /// Newest operation of the entity known not to publish.
+        through: QueuedOperationId,
+        /// When the same operations are planned again.
+        retry: Instant,
+    },
     /// Published its newest selected state, but operations past the last one
     /// acknowledged, `through`, are still queued: repaired again at full
     /// width. Not blocked.
@@ -237,24 +262,26 @@ pub(crate) enum HeldEntity {
 
 impl HeldEntity {
     /// Operations the entity's next repair takes from `queued`, its queued
-    /// operations in order, or `None` while it waits for a newer operation.
-    fn repair_width(self, queued: &[&QueuedOperation]) -> Option<usize> {
-        match self {
-            Self::Waiting { through } | Self::Draining { through } => {
-                let known = queued
-                    .iter()
-                    .position(|operation| operation.id() == through)
-                    .map_or(0, |position| position + 1);
-                (queued.len() > known).then_some(queued.len())
-            }
-            Self::Repairing { width, .. } => Some(width.get().min(queued.len())),
-        }
+    /// operations in order, or `None` while it waits for a newer operation
+    /// or, after failing to plan, for its retry to be due at `now`.
+    fn repair_width(self, queued: &[&QueuedOperation], now: Instant) -> Option<usize> {
+        let (through, due) = match self {
+            Self::Repairing { width, .. } => return Some(width.get().min(queued.len())),
+            Self::Waiting { through } | Self::Draining { through } => (through, false),
+            Self::Failed { through, retry } => (through, retry <= now),
+        };
+        let known = queued
+            .iter()
+            .position(|operation| operation.id() == through)
+            .map_or(0, |position| position + 1);
+        (due || queued.len() > known).then_some(queued.len())
     }
 
-    /// Whether the entity is held back because a state of it could not fit.
+    /// Whether the entity is held back because a state of it could not fit
+    /// or failed to plan.
     const fn is_blocked(self) -> bool {
         match self {
-            Self::Waiting { .. } | Self::Repairing { .. } => true,
+            Self::Waiting { .. } | Self::Failed { .. } | Self::Repairing { .. } => true,
             Self::Draining { .. } => false,
         }
     }
@@ -271,6 +298,7 @@ impl TargetSchedule {
             last_vector_commit: None,
             eligibility: Eligibility::Now,
             failures: 0,
+            failed_holds: 0,
             held: HashMap::new(),
         }
     }
@@ -704,6 +732,7 @@ impl QueuePublisher {
         match outcome {
             PublicationOutcome::Published { .. } => {
                 schedule.failures = 0;
+                schedule.failed_holds = 0;
                 schedule.eligibility = Eligibility::Now;
                 schedule.entity_limit = schedule.entity_limit.saturating_mul(2).min(default_limit);
                 schedule.operation_limit = schedule
@@ -713,6 +742,7 @@ impl QueuePublisher {
             // A retired generation never publishes again.
             PublicationOutcome::Discarded { .. } => {
                 schedule.failures = 0;
+                schedule.failed_holds = 0;
                 schedule.eligibility = Eligibility::Now;
                 schedule.held.clear();
             }
@@ -720,18 +750,34 @@ impl QueuePublisher {
                 schedules.remove(&target);
             }
             // Holding back a blocked entity is progress: the generation's
-            // other entities can publish at once.
+            // other entities can publish at once. A second entity failing to
+            // plan with no publication between suggests the failure is not
+            // the entities' own, so the next one waits (see `failed_holds`).
+            PublicationOutcome::Blocked if schedule.failed_holds > 1 => {
+                let backoff = RETRY_DELAY
+                    .saturating_mul(1_u32 << (schedule.failed_holds - 1).min(9))
+                    .min(MAX_BACKOFF);
+                schedule.eligibility = Eligibility::After(Instant::now() + backoff);
+            }
             PublicationOutcome::Trimmed | PublicationOutcome::Blocked => {
                 schedule.eligibility = Eligibility::Now;
             }
-            // Only a write can give a held-back entity a repair, so polling
-            // would reread and decode the queue for nothing.
+            // Only a write, or a failed entity's due retry, can give a
+            // held-back entity a repair, so polling would reread and decode
+            // the queue for nothing.
             PublicationOutcome::Stalled => {
                 schedule.failures = schedule.failures.saturating_add(1);
-                schedule.eligibility = Eligibility::NewWork {
-                    admitted,
-                    deadline: Instant::now() + MAX_STALLED_WAIT,
-                };
+                let deadline = schedule
+                    .held
+                    .values()
+                    .filter_map(|hold| match hold {
+                        HeldEntity::Failed { retry, .. } => Some(*retry),
+                        HeldEntity::Waiting { .. }
+                        | HeldEntity::Draining { .. }
+                        | HeldEntity::Repairing { .. } => None,
+                    })
+                    .fold(Instant::now() + MAX_STALLED_WAIT, Instant::min);
+                schedule.eligibility = Eligibility::NewWork { admitted, deadline };
             }
             PublicationOutcome::Empty
             | PublicationOutcome::Deferred
@@ -1240,6 +1286,26 @@ impl QueuePublisher {
             .collect()
     }
 
+    /// Makes every entity held back after failing to plan due for its retry,
+    /// as if [`MAX_STALLED_WAIT`] had passed.
+    #[cfg(any(
+        test,
+        all(feature = "production-coverage", feature = "index-lifecycle-testing")
+    ))]
+    pub(crate) fn make_failed_retries_due(&self) {
+        let now = Instant::now();
+        self.schedules
+            .lock()
+            .values_mut()
+            .flat_map(|schedule| schedule.held.values_mut())
+            .for_each(|hold| {
+                let HeldEntity::Failed { retry, .. } = hold else {
+                    return;
+                };
+                *retry = now;
+            });
+    }
+
     /// Shrinks the next selection of `target` after `selection`'s exact
     /// output crossed a budget before anything was published.
     ///
@@ -1278,7 +1344,9 @@ impl QueuePublisher {
             (Some((only, hold)), _) => {
                 let taken = only.taken().collect::<Vec<_>>();
                 let (through, tried) = match hold {
-                    HeldEntity::Waiting { through } => {
+                    // A failed entity that now plans but cannot fit is
+                    // blocked like any other.
+                    HeldEntity::Waiting { through } | HeldEntity::Failed { through, .. } => {
                         (through, taken.last().map_or(through, |last| last.id()))
                     }
                     // A draining entity already serves a state newer than
@@ -1354,16 +1422,20 @@ impl QueuePublisher {
     /// Narrows `target`'s next attempt toward the entity of `failed`, a
     /// selection whose planning failed deterministically with `error`.
     ///
-    /// A failed selection of one entity holds it back as [`HeldEntity`]
-    /// describes: it waits for an operation newer than every state of it
-    /// known not to publish, which a later write supplies and its repair then
-    /// plans, and the rotation moves past it, so the rest of its generation
-    /// keeps publishing. Nothing is acknowledged or dropped, and nothing
-    /// durable records the hold: a restarted publisher plans the entity again
-    /// and holds it back again if it still fails. A text epoch is planned as
-    /// a whole, so its failure does not name an entity; a failed epoch of
-    /// several entities halves the text entity ceiling instead, until a
-    /// failing epoch is one entity.
+    /// A failed selection of one entity holds it back as
+    /// [`HeldEntity::Failed`]: it waits for an operation newer than every
+    /// state of it known not to publish, which a later write supplies and its
+    /// repair then plans, or for its retry [`MAX_STALLED_WAIT`] from now,
+    /// which plans the same operations again in case what failed was
+    /// repaired. The rotation moves past it, so the rest of its generation
+    /// keeps publishing; from the second entity in a row that fails without a
+    /// publication between, the next attempt backs off instead (see
+    /// `TargetSchedule::failed_holds`). Nothing is acknowledged or dropped,
+    /// and nothing durable records the hold: a restarted publisher plans the
+    /// entity again and holds it back again if it still fails. A text epoch
+    /// is planned as a whole, so its failure does not name an entity; a
+    /// failed epoch of several entities halves the text entity ceiling
+    /// instead, until a failing epoch is one entity.
     fn isolate(
         &self,
         target: QueueTarget,
@@ -1384,16 +1456,26 @@ impl QueuePublisher {
         // first attempt already found not to fit.
         let through = match schedule.held.get(&only.entity) {
             Some(HeldEntity::Repairing { tried, .. }) => *tried,
-            Some(HeldEntity::Waiting { .. } | HeldEntity::Draining { .. }) | None => only
+            Some(
+                HeldEntity::Waiting { .. }
+                | HeldEntity::Failed { .. }
+                | HeldEntity::Draining { .. },
+            )
+            | None => only
                 .taken()
                 .last()
                 .expect("a selected entity has operations")
                 .id(),
         };
-        schedule
-            .held
-            .insert(only.entity, HeldEntity::Waiting { through });
+        schedule.held.insert(
+            only.entity,
+            HeldEntity::Failed {
+                through,
+                retry: Instant::now() + MAX_STALLED_WAIT,
+            },
+        );
         schedule.cursor = Some(only.entity);
+        schedule.failed_holds = schedule.failed_holds.saturating_add(1);
         self.metrics
             .blocked_attempts
             .fetch_add(1, Ordering::Relaxed);
@@ -1403,8 +1485,9 @@ impl QueuePublisher {
             entity = ?only.entity,
             index_id = target.index_id.get(),
             generation = target.generation.get(),
+            retry_after_secs = MAX_STALLED_WAIT.as_secs(),
             "planning a queued index operation failed deterministically; its entity is held \
-             back until a later write supersedes it"
+             back until a later write supersedes it or its retry is due"
         );
         PublicationOutcome::Blocked
     }
@@ -1727,9 +1810,10 @@ impl<'a> SelectedEntity<'a> {
 /// acknowledged; the rest are `superseding`. When the rotation reaches it
 /// after other entities, the batch ends there: the next batch starts after
 /// this one's last published entity, so the rotation never carries past a
-/// repair, however the rest of its generation is written. The result is
-/// empty only when every queued entity is held back without a repair to
-/// try.
+/// repair, however the rest of its generation is written. A failed entity's
+/// retry is due once its instant has passed when the selection runs. The
+/// result is empty only when every queued entity is held back without a
+/// repair to try.
 pub(crate) fn select_batch<'a>(
     operations: &'a [QueuedOperation],
     after: Option<IndexEntity>,
@@ -1754,6 +1838,7 @@ pub(crate) fn select_batch<'a>(
         .and_then(|entity| order.iter().position(|candidate| *candidate == entity))
         .map_or(0, |position| position + 1);
     let max_operations = max_operations.min(max_acknowledged).get();
+    let now = Instant::now();
     let mut selected = Vec::new();
     let mut input_bytes = 0_u64;
     let mut selected_operations = 0_usize;
@@ -1762,7 +1847,10 @@ pub(crate) fn select_batch<'a>(
             break;
         }
         let mut queued = grouped.remove(&entity).unwrap_or_default();
-        match held.get(&entity).map(|hold| hold.repair_width(&queued)) {
+        match held
+            .get(&entity)
+            .map(|hold| hold.repair_width(&queued, now))
+        {
             Some(None) => continue,
             Some(Some(_)) if !selected.is_empty() => break,
             Some(Some(width)) => {
diff --git a/crates/db/tests/production_support/queue_publication.rs b/crates/db/tests/production_support/queue_publication.rs
index 77aa46123..560944112 100644
--- a/crates/db/tests/production_support/queue_publication.rs
+++ b/crates/db/tests/production_support/queue_publication.rs
@@ -555,6 +555,31 @@ async fn selection_and_collapse_boundaries() {
         )),
         [(second, 1, 0)]
     );
+    // An entity that failed to plan waits like a blocked one until its retry
+    // is due, then is repaired alone at full width without a newer operation.
+    let failed = |retry| {
+        HashMap::from([(
+            vector[0].entity(),
+            HeldEntity::Failed {
+                through: vector[1].id(),
+                retry,
+            },
+        )])
+    };
+    assert_eq!(
+        shape(&select(
+            2,
+            &failed(Instant::now() + MAX_STALLED_WAIT),
+            all,
+            all,
+            u64::MAX
+        )),
+        [(second, 1, 0)]
+    );
+    assert_eq!(
+        shape(&select(2, &failed(Instant::now()), all, all, u64::MAX)),
+        [(first, 2, 0)]
+    );
 
     assert!(matches!(
         collapse_vector(&SelectedEntity {
@@ -953,8 +978,8 @@ enum Change {
 
 /// Proves publication fails closed when a namespace's metadata disagrees with
 /// its definition or is missing, for both removals and upserts: it writes
-/// nothing and holds back only the entity it planned, which a publisher
-/// restarted after the metadata is restored publishes.
+/// nothing and holds back only the entity it planned, whose retry publishes
+/// it without another write once the metadata is restored.
 async fn inconsistent_namespace_metadata_fails_closed() {
     let db = open_explicit(
         "queue-publication-metadata",
@@ -1054,11 +1079,11 @@ async fn inconsistent_namespace_metadata_fails_closed() {
                 .await
                 .expect("a held entity waits"),
             PublicationOutcome::Stalled,
-            "only a write or a restart plans a held entity again"
+            "a held entity waits for a write or its retry"
         );
-        let restarted = publisher_with_limits(&db, publisher.limits, publisher.text.limits);
+        publisher.make_failed_retries_due();
         assert!(matches!(
-            restarted
+            publisher
                 .publish_once(target)
                 .await
                 .expect("restored metadata publishes"),
diff --git a/docs/database/helix-db/query-guides/troubleshooting.mdx b/docs/database/helix-db/query-guides/troubleshooting.mdx
index 0f6176a6e..ad9a3bbb2 100644
--- a/docs/database/helix-db/query-guides/troubleshooting.mdx
+++ b/docs/database/helix-db/query-guides/troubleshooting.mdx
@@ -102,8 +102,11 @@ fits the lowered limits.
 The worker also holds back a single vector or text entry whose update fails the same
 way every time it is applied, for example on damaged index data, and keeps publishing
 the rest of the index. It counts under `blocked_index_entity_count` too, and the logged
-error names the entry and the failure. Writing the entry again retries it, and so does
-restarting the writer.
+error names the entry and the failure. The worker retries the entry about once a
+minute, so it publishes on its own once the cause is fixed. Until then, every write to
+that entry stays queued and counts toward the index's limit on unpublished work, so
+writing it over and over can fill the limit and return `index_backpressure` for writes
+to the whole index. Stop writing the entry until it publishes.
 
 ## Request returns HTTP 429 (Helix Cloud)
 
diff --git a/docs/llms-full.txt b/docs/llms-full.txt
index 254c7514f..a326f29c6 100644
--- a/docs/llms-full.txt
+++ b/docs/llms-full.txt
@@ -5581,6 +5581,15 @@ error naming each one when it is held back. Raise the limits again. Rewriting or
 deleting a document also clears it once replacing or removing its published version
 fits the lowered limits.
 
+The worker also holds back a single vector or text entry whose update fails the same
+way every time it is applied, for example on damaged index data, and keeps publishing
+the rest of the index. It counts under `blocked_index_entity_count` too, and the logged
+error names the entry and the failure. The worker retries the entry about once a
+minute, so it publishes on its own once the cause is fixed. Until then, every write to
+that entry stays queued and counts toward the index's limit on unpublished work, so
+writing it over and over can fill the limit and return `index_backpressure` for writes
+to the whole index. Stop writing the entry until it publishes.
+
 ## Request returns HTTP 429 (Helix Cloud)
 
 `rate_limited` means the shared request bucket for the Cloud database has no

From 8708fe2ec237f7fbe545fd150b7657983a089224 Mon Sep 17 00:00:00 2001
From: xav-db <xav@helix-db.com>
Date: Sun, 4 Oct 2026 15:08:29 +0100
Subject: [PATCH 7/9] docs: count failed-planning holds in
 blocked_index_entity_count

The health field also counts entities whose planning fails the same way
every time, but its OpenAPI, proto, and server descriptions still named
only operations that cannot fit a publication.
---
 crates/server/proto/helixdb.proto | 3 ++-
 crates/server/src/state.rs        | 7 ++++---
 docs/openapi.json                 | 2 +-
 3 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/crates/server/proto/helixdb.proto b/crates/server/proto/helixdb.proto
index d407cf2d8..84bd165a9 100644
--- a/crates/server/proto/helixdb.proto
+++ b/crates/server/proto/helixdb.proto
@@ -27,7 +27,8 @@ message HealthResponse {
   reserved 4;
   reserved "text_index_runtime";
   // Entities the index worker holds back because a queued operation of
-  // theirs can never fit a publication under the current limits. Only the
+  // theirs can never fit a publication under the current limits, or because
+  // applying their queued change fails the same way every time. Only the
   // count: the writer's logs name each one.
   uint64 blocked_index_entity_count = 5;
 }
diff --git a/crates/server/src/state.rs b/crates/server/src/state.rs
index a46713b07..86be52cc3 100644
--- a/crates/server/src/state.rs
+++ b/crates/server/src/state.rs
@@ -49,9 +49,10 @@ impl ServerState {
 
     /// How many entities the writer's index worker holds back: each has a
     /// queued vector/text operation that can never fit a publication under
-    /// the current limits. Zero on a reader, and rebuilt after a restart as
-    /// publication blocks again. Health responses report only this count, so
-    /// unauthenticated probes never learn tenant or entity IDs.
+    /// the current limits, or whose planning fails deterministically. Zero on
+    /// a reader, and rebuilt after a restart as publication blocks again.
+    /// Health responses report only this count, so unauthenticated probes
+    /// never learn tenant or entity IDs.
     pub fn blocked_index_entity_count(&self) -> u64 {
         self.db.blocked_index_entity_count()
     }
diff --git a/docs/openapi.json b/docs/openapi.json
index 373f17a39..8ab29ef8c 100644
--- a/docs/openapi.json
+++ b/docs/openapi.json
@@ -870,7 +870,7 @@
           "blocked_index_entity_count": {
             "type": "integer",
             "minimum": 0,
-            "description": "Nodes and edges whose queued vector or text index work the index worker holds back because it can never fit a publication under the current limits, typically after the limits were lowered. Only the count: the writer logs an error naming each one when it is held back. Always 0 on a reader. Rebuilt after a restart as publication reaches them again."
+            "description": "Nodes and edges whose queued vector or text index work the index worker holds back: either it can never fit a publication under the current limits, typically after the limits were lowered, or applying it fails the same way every time, for example on damaged index data. Only the count: the writer logs an error naming each one when it is held back. Always 0 on a reader. Rebuilt after a restart as publication reaches them again."
           }
         },
         "additionalProperties": false,

From 61240f7456b08cc1a9f7f91148b26bfb82fd0218 Mon Sep 17 00:00:00 2001
From: xav-db <xav@helix-db.com>
Date: Sun, 4 Oct 2026 22:57:00 +0100
Subject: [PATCH 8/9] fix(db): re-read the queue after a failed-planning hold

A hold after failed planning kept the attempt's queue for the next
attempt, against the rule that a blocked outcome drops it. With enough
entities whose retries keep falling due, no attempt stalled, storage was
never read again, and newer writes, including repairs of held entities,
never published. A hold now drops the queue like a size-blocked one; a
trimmed text epoch still keeps it.
---
 .../index_lifecycle/queue/isolation_tests.rs  | 99 +++++++++++++++++++
 .../src/index_lifecycle/queue/publication.rs  | 39 +++++---
 2 files changed, 122 insertions(+), 16 deletions(-)

diff --git a/crates/db/src/index_lifecycle/queue/isolation_tests.rs b/crates/db/src/index_lifecycle/queue/isolation_tests.rs
index acc077d15..d420ade8e 100644
--- a/crates/db/src/index_lifecycle/queue/isolation_tests.rs
+++ b/crates/db/src/index_lifecycle/queue/isolation_tests.rs
@@ -480,6 +480,105 @@ async fn consecutive_planning_failures_back_off() {
     db.close().await.unwrap();
 }
 
+/// Entities that keep failing to plan never keep publication on a queue read
+/// before newer writes. Past twelve of them, the 5 s backoffs between their
+/// holds outlast the 60 s retry wait, so each retry is due again when the
+/// rotation returns to it and no attempt stalls. Each hold reads the queue
+/// again, so new writes, and one that repairs a held entity, publish within
+/// two rotations, and every other held operation stays queued.
+#[tokio::test]
+async fn writes_publish_while_many_failed_entities_keep_retrying() {
+    let db = open(
+        "isolate-many-retrying",
+        Arc::new(InMemory::new()),
+        queued(tuning()),
+    )
+    .await;
+    install_vector(&db, None).await;
+    let mut failing = Vec::new();
+    for x in 0..16_u8 {
+        failing.push(add_doc(&db, vec![f32::from(x), 0.0], "doc").await.unwrap());
+    }
+    let target = target(&db, QueueFamily::Vector).await;
+    settle(&db, target).await;
+    for id in &failing {
+        set(&db, *id, "embedding", vec![50.0_f32, 50.0].into()).await;
+        let corrupt = newest(&db, QueueFamily::Vector, *id).await;
+        inject(&db, corrupt, Some(InjectedPlanningFailure::Corrupt));
+    }
+    let mut holds = vec![PublicationOutcome::Blocked; failing.len()];
+    holds.push(PublicationOutcome::Stalled);
+    assert_eq!(settle(&db, target).await, holds);
+    // Every retry is due whenever the rotation reaches it, from this attempt
+    // on, whose queue predates the writes below.
+    publisher(&db).make_failed_retries_due();
+    assert_eq!(
+        publisher(&db).publish_once(target).await.unwrap(),
+        PublicationOutcome::Blocked
+    );
+    let repaired = failing[8];
+    set(&db, repaired, "embedding", vec![60.0_f32, 60.0].into()).await;
+    let inserted = [
+        add_doc(&db, vec![100.0, 0.0], "doc").await.unwrap(),
+        add_doc(&db, vec![101.0, 0.0], "doc").await.unwrap(),
+    ];
+    let still_failing = failing
+        .iter()
+        .copied()
+        .filter(|id| *id != repaired)
+        .collect::<Vec<_>>();
+
+    // A publication that drains the rotation's cursor entity restarts the
+    // rotation at the queue's head, so the repair and the inserts' batch
+    // each publish within two rotations of the held entities.
+    let mut outcomes = Vec::new();
+    while queued_ids(&db, QueueFamily::Vector).await != still_failing {
+        assert!(
+            outcomes.len() < 2 * failing.len(),
+            "newer writes did not publish within two rotations: {outcomes:?}"
+        );
+        publisher(&db).make_failed_retries_due();
+        outcomes.push(publisher(&db).publish_once(target).await.unwrap());
+    }
+    assert!(
+        outcomes.iter().all(|outcome| matches!(
+            outcome,
+            PublicationOutcome::Blocked | PublicationOutcome::Published { .. }
+        )),
+        "{outcomes:?}"
+    );
+    for published in [
+        PublicationOutcome::Published {
+            operations: 2,
+            entities: 1,
+        },
+        PublicationOutcome::Published {
+            operations: 2,
+            entities: 2,
+        },
+    ] {
+        assert!(
+            outcomes.contains(&published),
+            "the repair and the inserts publish: {outcomes:?}"
+        );
+    }
+    assert_eq!(
+        held(&db).into_iter().collect::<HashSet<_>>(),
+        still_failing.iter().copied().collect::<HashSet<_>>(),
+        "every other entity stays held"
+    );
+    assert_eq!(published_nearest(&db, [60.0, 60.0]).await, Some(repaired));
+    assert_eq!(
+        published_nearest(&db, [100.0, 0.0]).await,
+        Some(inserted[0])
+    );
+    assert_eq!(
+        published_nearest(&db, [101.0, 0.0]).await,
+        Some(inserted[1])
+    );
+    db.close().await.unwrap();
+}
+
 /// A text epoch fails as a whole, so its entity ceiling halves until the
 /// failing entity publishes alone and is held back; every other entity
 /// publishes, and a newer write to it publishes it.
diff --git a/crates/db/src/index_lifecycle/queue/publication.rs b/crates/db/src/index_lifecycle/queue/publication.rs
index d310284e3..9c6f37541 100644
--- a/crates/db/src/index_lifecycle/queue/publication.rs
+++ b/crates/db/src/index_lifecycle/queue/publication.rs
@@ -1035,10 +1035,9 @@ impl QueuePublisher {
             .collect::<std::result::Result<Vec<_>, _>>()
         {
             Ok(effects) => effects,
+            // A hold drops the queue (see `Self::isolate`).
             Err((failed, error)) => {
-                let outcome = self.isolate(target, std::slice::from_ref(failed), &error);
-                self.store.retained().retain(target, stored, &[]);
-                return Ok(outcome);
+                return Ok(self.isolate(target, std::slice::from_ref(failed), &error));
             }
         };
         #[cfg(test)]
@@ -1099,12 +1098,12 @@ impl QueuePublisher {
             Ok(StagedEffects::Failed { position, error })
                 if FailureKind::of(&error) == FailureKind::Deterministic =>
             {
-                let outcome =
-                    self.isolate(target, std::slice::from_ref(&selection[position]), &error);
-                // Nothing committed: the next selection, without the held
-                // entity, reuses the queue.
-                self.store.retained().retain(target, stored, &[]);
-                return Ok(outcome);
+                // A hold drops the queue (see `Self::isolate`).
+                return Ok(self.isolate(
+                    target,
+                    std::slice::from_ref(&selection[position]),
+                    &error,
+                ));
             }
             // Planning proved the transaction cannot commit.
             Ok(StagedEffects::Failed { error, .. }) | Err(error)
@@ -1430,12 +1429,19 @@ impl QueuePublisher {
     /// repaired. The rotation moves past it, so the rest of its generation
     /// keeps publishing; from the second entity in a row that fails without a
     /// publication between, the next attempt backs off instead (see
-    /// `TargetSchedule::failed_holds`). Nothing is acknowledged or dropped,
+    /// `TargetSchedule::failed_holds`). Nothing is acknowledged or discarded,
     /// and nothing durable records the hold: a restarted publisher plans the
     /// entity again and holds it back again if it still fails. A text epoch
     /// is planned as a whole, so its failure does not name an entity; a
     /// failed epoch of several entities halves the text entity ceiling
     /// instead, until a failing epoch is one entity.
+    ///
+    /// The caller drops its queue after a hold, as after any blocked
+    /// operation ([`super::storage::RetainedQueues`]), so the next attempt
+    /// reads storage and sees every newer write, including one that repairs a
+    /// held entity. Retaining it would let held entities whose retries keep
+    /// falling due keep every attempt on that queue, so no newer write would
+    /// ever publish.
     fn isolate(
         &self,
         target: QueueTarget,
@@ -1593,10 +1599,9 @@ impl QueuePublisher {
             .collect::<std::result::Result<Vec<_>, _>>()
         {
             Ok(effects) => effects,
+            // A hold drops the queue (see `Self::isolate`).
             Err((failed, error)) => {
-                let outcome = self.isolate(target, std::slice::from_ref(failed), &error);
-                self.store.retained().retain(target, stored, &[]);
-                return Ok(outcome);
+                return Ok(self.isolate(target, std::slice::from_ref(failed), &error));
             }
         };
         #[cfg(test)]
@@ -1647,9 +1652,11 @@ impl QueuePublisher {
             }
             Err(error) if FailureKind::of(&error) == FailureKind::Deterministic => {
                 let outcome = self.isolate(target, &selection, &error);
-                // Nothing committed: the next, narrower epoch reuses the
-                // queue.
-                self.store.retained().retain(target, stored, &[]);
+                if outcome == PublicationOutcome::Trimmed {
+                    // Nothing committed: the next, narrower epoch reuses the
+                    // queue. A hold drops it (see `Self::isolate`).
+                    self.store.retained().retain(target, stored, &[]);
+                }
                 return Ok(outcome);
             }
             Err(error) => return Err(error),

From bf8f74959b67a2d293a51048cb83bcbabe194dea Mon Sep 17 00:00:00 2001
From: xav-db <xav@helix-db.com>
Date: Sun, 4 Oct 2026 22:57:00 +0100
Subject: [PATCH 9/9] test(db): check the consecutive-failure backoff without a
 wall-clock window

Judge eligibility as of an instant before the attempt rescheduled and
bound the retry instant by the attempt's start and return, so a slow
host cannot let the 20 ms backoff pass before the check.
---
 .../db/src/index_lifecycle/queue/isolation_tests.rs  | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/crates/db/src/index_lifecycle/queue/isolation_tests.rs b/crates/db/src/index_lifecycle/queue/isolation_tests.rs
index d420ade8e..f00b69d87 100644
--- a/crates/db/src/index_lifecycle/queue/isolation_tests.rs
+++ b/crates/db/src/index_lifecycle/queue/isolation_tests.rs
@@ -447,13 +447,23 @@ async fn consecutive_planning_failures_back_off() {
         NextTarget::Ready(target),
         "one failure holds back only its entity"
     );
+    let attempted = Instant::now();
     assert_eq!(
         scheduler.publish_once(target).await.unwrap(),
         PublicationOutcome::Blocked
     );
-    let NextTarget::Delayed(retry) = scheduler.next_target(&none, Instant::now()) else {
+    let returned = Instant::now();
+    // Judged as of an instant before the attempt rescheduled, so the backoff
+    // has not passed however slowly the attempt ran.
+    let NextTarget::Delayed(retry) = scheduler.next_target(&none, attempted) else {
         panic!("a second failure in a row backs off");
     };
+    // Twice the 10 ms retry delay, from when the attempt rescheduled.
+    let backoff = Duration::from_millis(20);
+    assert!(
+        (attempted + backoff..=returned + backoff).contains(&retry),
+        "the second failure in a row backs off by {backoff:?}"
+    );
     assert_eq!(
         scheduler.next_target(&none, retry),
         NextTarget::Ready(target)