refactor(lint): fix root causes instead of suppressing clippy lints

Replace needless range-loops with iterator adapters in the OPQ/RaBitQ
quantizers and Vamana graph build, add OpLog::is_empty instead of
allowing len_without_is_empty, drop a stale doc_markdown allow, and
move the Send/Sync safety justification for the mmap-backed vector
segment onto MmapVectorSegment itself so PlainMmapBacking no longer
needs an arc_with_non_send_sync allow.

Also split the over-long transaction batch commit and sub-plan
dispatch functions into focused per-stage/per-engine helpers
(sub_plan_write.rs), removing the too_many_lines allows.

Author: farhan-syah

nodedb-array/src/sync/op_log.rs

@@ -22,7 +22,6 @@ pub type OpIter<'a> = Box<dyn Iterator<Item = ArrayResult<ArrayOp>> + 'a>;
 /// Implementations are expected to be `Send + Sync` and durable. An
 /// in-memory implementation suitable for tests is provided by
 /// [`InMemoryOpLog`] below.
-#[allow(clippy::len_without_is_empty)]
 pub trait OpLog: Send + Sync {
     /// Append an operation to the log.
     ///
@@ -39,6 +38,11 @@ pub trait OpLog: Send + Sync {
     /// Return the total number of ops in the log (across all arrays).
     fn len(&self) -> ArrayResult<u64>;
 
+    /// Return `true` if the log contains no ops in any array.
+    fn is_empty(&self) -> ArrayResult<bool> {
+        Ok(self.len()? == 0)
+    }
+
     /// Drop all ops whose `hlc < hlc` and return the count dropped.
     ///
     /// Used by GC after snapshotting ops below the min-ack frontier.

nodedb-codec/src/vector_quant/opq.rs

@@ -209,11 +209,10 @@ fn matvec(r: &[f32], v: &[f32], dim: usize) -> Vec<f32> {
 
 fn pq_encode(v: &[f32], codebooks: &[Vec<Vec<f32>>], m: usize, sub_dim: usize) -> Vec<u8> {
     let mut codes = Vec::with_capacity(m);
-    #[allow(clippy::needless_range_loop)]
-    for s in 0..m {
+    for (s, codebook) in codebooks.iter().enumerate().take(m) {
         let offset = s * sub_dim;
         let sub = &v[offset..offset + sub_dim];
-        let best = codebooks[s]
+        let best = codebook
             .iter()
             .enumerate()
             .min_by(|(_, a), (_, b)| {

nodedb-codec/src/vector_quant/rabitq.rs

@@ -245,14 +245,13 @@ impl RaBitQCodec {
         wht_inplace(&mut sign_buf);
         // Re-apply signed diagonal (D is its own inverse since flips are ±1)
         let mut seed = self.rotation_seed;
-        #[allow(clippy::needless_range_loop)]
-        for i in 0..dim {
+        for x in sign_buf.iter_mut().take(dim) {
             let flip = if xorshift64(&mut seed) & 1 == 0 {
                 1.0f32
             } else {
                 -1.0f32
             };
-            sign_buf[i] *= flip;
+            *x *= flip;
         }
         let dot_raw: f32 = residual
             .iter()

nodedb-physical/src/physical_plan/document/op.rs

@@ -70,7 +70,6 @@ pub enum DocumentOp {
         /// If non-empty, the Data Plane evaluates these after fetching
         /// the document. Returns `NOT_FOUND` on denial (no info leak).
         /// Injected by the Control Plane planner from RLS policies.
-        #[allow(clippy::doc_markdown)]
         rls_filters: Vec<u8>,
         /// System-time selection. `Current` = current state. Honored only by
         /// bitemporal collections; the planner rejects temporal point-gets on

nodedb-vector/src/mmap_segment/reader.rs

@@ -21,7 +21,8 @@ use crate::error::VectorError;
 /// Exposes a `&[f32]` view of the vector data block and a `&[u64]` view of
 /// the surrogate ID block — both zero-copy slices into the mmap region.
 ///
-/// Not `Send` or `Sync` — owned by a single Data Plane core.
+/// Typically owned by a single Data Plane core, but safe to share across
+/// threads behind an [`Arc`]: see the `Send`/`Sync` safety note below.
 #[derive(Debug)]
 pub struct MmapVectorSegment {
     path: PathBuf,
@@ -41,6 +42,13 @@ pub struct MmapVectorSegment {
     _budget_guard: Option<BudgetGuard>,
 }
 
+// SAFETY: `MmapVectorSegment` holds a `*const u8` (`base`) into a read-only
+// `MAP_PRIVATE` mmap region. The region is immutable after construction,
+// process-global, and valid for the lifetime of the value. There is no
+// interior mutability, so concurrent reads from multiple threads are sound.
+unsafe impl Send for MmapVectorSegment {}
+unsafe impl Sync for MmapVectorSegment {}
+
 impl MmapVectorSegment {
     // ── Constructors ──────────────────────────────────────────────────────────
 

nodedb-vector/src/segment_backing/plain.rs

@@ -36,28 +36,15 @@ use super::VectorSegmentBacking;
 /// The `Arc<MmapVectorSegment>` wrapper ensures the segment (and therefore
 /// the mmap region) outlives any `&[f32]` slice handed out through this type.
 ///
-/// SAFETY: given the above invariants, treating `PlainMmapBacking` as
-/// `Send + Sync` is correct.
+/// `PlainMmapBacking` is `Send + Sync` automatically: [`MmapVectorSegment`]
+/// carries the `unsafe impl Send + Sync` (justified by the invariants above),
+/// so `Arc<MmapVectorSegment>` — and therefore this wrapper — is too.
 pub struct PlainMmapBacking {
     inner: Arc<MmapVectorSegment>,
 }
 
-// SAFETY: see struct-level doc comment.  `MmapVectorSegment` holds a
-// `*const u8` (`base`) into a read-only MAP_PRIVATE mmap region.  The region
-// is immutable after construction, process-global, and valid for the lifetime
-// of the Arc.  No interior mutability exists; concurrent reads are safe.
-unsafe impl Send for PlainMmapBacking {}
-unsafe impl Sync for PlainMmapBacking {}
-
 impl PlainMmapBacking {
     /// Wrap a [`MmapVectorSegment`] that is not yet reference-counted.
-    ///
-    /// `MmapVectorSegment` holds a `*const u8` raw pointer which makes it
-    /// `!Send + !Sync` by default. The `unsafe impl Send + Sync` on
-    /// `PlainMmapBacking` (see struct-level doc comment) establishes the safety
-    /// invariants; clippy cannot see through the `Arc` to the impl, so we
-    /// suppress the lint here.
-    #[allow(clippy::arc_with_non_send_sync)]
     pub fn new(seg: MmapVectorSegment) -> Self {
         Self {
             inner: Arc::new(seg),

nodedb-vector/src/vamana/build.rs

@@ -93,9 +93,8 @@ pub fn build_vamana<C: VectorCodec>(
     }
 
     // --- Pass 1 refinement: beam-search to replace random neighbors ---
-    #[allow(clippy::needless_range_loop)]
-    for i in 0..n {
-        let query = codec.prepare_query(&vectors[i]);
+    for (i, vector) in vectors.iter().enumerate().take(n) {
+        let query = codec.prepare_query(vector);
         let candidates = greedy_beam_search_quantized(&graph, &query, codec, quantized, i, l);
         // Take the R closest from the candidates, excluding self.
         let mut nbrs: Vec<(u32, f32)> = candidates

nodedb/src/data/executor/handlers/transaction/batch.rs

@@ -18,6 +18,11 @@ use nodedb_physical::physical_plan::PhysicalPlan;
 
 use super::undo::UndoEntry;
 
+/// A CRDT delta buffered during a transaction batch: `(delta_bytes, id,
+/// collection)`. Deltas accumulate in a scratch buffer and are applied to
+/// the LoroDoc only on full-batch success.
+type CrdtDelta = (Vec<u8>, u64, String);
+
 impl CoreLoop {
     /// Execute a transaction batch atomically.
     ///
@@ -27,7 +32,6 @@ impl CoreLoop {
     ///
     /// The Control Plane has already written a single `RecordType::Transaction`
     /// WAL record covering all operations before dispatching this batch.
-    #[allow(clippy::too_many_lines)]
     pub(in crate::data::executor) fn execute_transaction_batch(
         &mut self,
         task: &ExecutionTask,
@@ -40,16 +44,68 @@ impl CoreLoop {
             "transaction batch begin"
         );
 
-        let mut undo_log: Vec<UndoEntry> = Vec::with_capacity(plans.len());
-        let mut crdt_deltas: Vec<(Vec<u8>, u64, String)> = Vec::new();
+        let undo_log: Vec<UndoEntry> = Vec::with_capacity(plans.len());
+        let crdt_deltas: Vec<CrdtDelta> = Vec::new();
+
+        let (last_response, undo_log, crdt_deltas) =
+            match self.run_sub_plans(task, tid, plans, undo_log, crdt_deltas) {
+                Ok(v) => v,
+                Err(resp) => return resp,
+            };
+
+        let undo_log = match self.apply_balanced_constraint_check(task, tid, undo_log) {
+            Ok(u) => u,
+            Err(resp) => return resp,
+        };
+
+        if let Some(resp) = self.apply_crdt_deltas(task, tid, crdt_deltas) {
+            return resp;
+        }
+
+        debug!(
+            core = self.core_id,
+            committed = plans.len(),
+            "transaction batch committed"
+        );
+
+        self.emit_deferred_writes(task, undo_log);
+
+        // Return the last sub-plan payload, but keyed to the outer transaction request.
+        Response {
+            request_id: task.request_id(),
+            status: Status::Ok,
+            attempt: 1,
+            partial: false,
+            payload: last_response.payload,
+            watermark_lsn: self.watermark,
+            error_code: None,
+        }
+    }
+
+    /// Run every sub-plan in order, tracking undo entries and buffered CRDT
+    /// deltas as it goes.
+    ///
+    /// On success, returns the last sub-plan's response plus the accumulated
+    /// undo log and CRDT delta buffer (for the caller's subsequent commit
+    /// steps). On failure, rolls back all writes performed so far and
+    /// returns the terminal error `Response` directly — the caller must
+    /// return it unchanged.
+    ///
+    /// A panic (real or test-injected) during a sub-apply is caught so it
+    /// routes through the same typed-rollback path instead of unwinding past
+    /// `undo_log`, which would drop the log without running rollback and
+    /// leave the shard half-committed.
+    fn run_sub_plans(
+        &mut self,
+        task: &ExecutionTask,
+        tid: u64,
+        plans: &[PhysicalPlan],
+        mut undo_log: Vec<UndoEntry>,
+        mut crdt_deltas: Vec<CrdtDelta>,
+    ) -> Result<(Response, Vec<UndoEntry>, Vec<CrdtDelta>), Response> {
         let mut last_response = self.response_ok(task);
 
         for (i, plan) in plans.iter().enumerate() {
-            // Wrap the sub-apply + post-apply fail-point in catch_unwind so a
-            // panic (real or test-injected) routes through the typed-rollback
-            // arm below instead of unwinding past `undo_log`. Without this,
-            // a panic between sub-applies would drop the undo log without
-            // running rollback and leave the shard half-committed.
             let user_roles = &task.request.user_roles;
             let outcome = catch_unwind(AssertUnwindSafe(|| {
                 let r = self.execute_tx_sub_plan(
@@ -118,20 +174,32 @@ impl CoreLoop {
                     // Discard CRDT scratch buffer (never applied).
                     drop(crdt_deltas);
 
-                    return Response {
+                    return Err(Response {
                         request_id: task.request_id(),
                         status: Status::Error,
                         attempt: 1,
                         partial: false,
                         payload: crate::bridge::envelope::Payload::empty(),
                         watermark_lsn: self.watermark,
                         error_code: Some(rollback_error_code),
-                    };
+                    });
                 }
             }
         }
 
-        // Pre-commit: BALANCED constraint check across all inserts in this transaction.
+        Ok((last_response, undo_log, crdt_deltas))
+    }
+
+    /// Pre-commit: check the `BALANCED` constraint across all inserts in
+    /// this transaction. On violation, rolls back and returns the terminal
+    /// error `Response`; on success, hands the undo log back for the
+    /// deferred-trigger step.
+    fn apply_balanced_constraint_check(
+        &mut self,
+        task: &ExecutionTask,
+        tid: u64,
+        undo_log: Vec<UndoEntry>,
+    ) -> Result<Vec<UndoEntry>, Response> {
         if let Err(error_code) =
             self.check_balanced_constraints(task.request.database_id.as_u64(), tid, &undo_log)
         {
@@ -160,24 +228,35 @@ impl CoreLoop {
                     }
                 }
             };
-            return Response {
+            return Err(Response {
                 request_id: task.request_id(),
                 status: Status::Error,
                 attempt: 1,
                 partial: false,
                 payload: crate::bridge::envelope::Payload::empty(),
                 watermark_lsn: self.watermark,
                 error_code: Some(rollback_error_code),
-            };
+            });
         }
+        Ok(undo_log)
+    }
 
-        // All sub-plans succeeded. Apply buffered CRDT deltas.
-        // Failure here means the CRDT state is inconsistent with the already-committed
-        // forward writes — return RollbackFailed so the client knows the shard needs
-        // a restart to restore consistency via WAL replay. Never warn-and-continue.
+    /// Apply all buffered CRDT deltas now that every sub-plan and the
+    /// `BALANCED` constraint check have succeeded.
+    ///
+    /// Failure here means the CRDT state is inconsistent with the
+    /// already-committed forward writes — returns a `RollbackFailed`
+    /// `Response` so the client knows the shard needs a restart to restore
+    /// consistency via WAL replay. Never warn-and-continue.
+    fn apply_crdt_deltas(
+        &mut self,
+        task: &ExecutionTask,
+        tid: u64,
+        crdt_deltas: Vec<CrdtDelta>,
+    ) -> Option<Response> {
         for (crdt_idx, (delta, _peer_id, collection)) in crdt_deltas.into_iter().enumerate() {
-            // Second crash-injection point — between forward-write commit and
-            // CRDT apply. WAL replay must roll the CRDT side forward (or roll
+            // Crash-injection point — between forward-write commit and CRDT
+            // apply. WAL replay must roll the CRDT side forward (or roll
             // forward writes back) to restore consistency.
             crate::fail_point!("transaction_batch::between_crdt_delta");
 
@@ -197,7 +276,7 @@ impl CoreLoop {
                             "CRDT delta apply failed after forward writes committed; \
                              shard state unknown — restart required for WAL replay"
                         );
-                        return Response {
+                        return Some(Response {
                             request_id: task.request_id(),
                             status: Status::Error,
                             attempt: 1,
@@ -208,7 +287,7 @@ impl CoreLoop {
                                 entry_index: crdt_idx,
                                 detail: format!("CRDT delta apply failed: {e}"),
                             }),
-                        };
+                        });
                     }
                 }
                 Err(e) => {
@@ -219,7 +298,7 @@ impl CoreLoop {
                         "CRDT engine not found after forward writes committed; \
                          shard state unknown — restart required for WAL replay"
                     );
-                    return Response {
+                    return Some(Response {
                         request_id: task.request_id(),
                         status: Status::Error,
                         attempt: 1,
@@ -230,18 +309,16 @@ impl CoreLoop {
                             entry_index: crdt_idx,
                             detail: format!("CRDT engine not available: {e}"),
                         }),
-                    };
+                    });
                 }
             }
         }
+        None
+    }
 
-        debug!(
-            core = self.core_id,
-            committed = plans.len(),
-            "transaction batch committed"
-        );
-
-        // Emit deferred trigger events for all writes in the committed transaction.
+    /// Emit deferred trigger events for every write recorded in the
+    /// committed transaction's undo log.
+    fn emit_deferred_writes(&mut self, task: &ExecutionTask, undo_log: Vec<UndoEntry>) {
         use crate::data::executor::core_loop::deferred::DeferredWrite;
         let deferred_writes: Vec<DeferredWrite> = undo_log
             .into_iter()
@@ -286,17 +363,6 @@ impl CoreLoop {
                 task.request.vshard_id,
             );
         }
-
-        // Return the last sub-plan payload, but keyed to the outer transaction request.
-        Response {
-            request_id: task.request_id(),
-            status: Status::Ok,
-            attempt: 1,
-            partial: false,
-            payload: last_response.payload,
-            watermark_lsn: self.watermark,
-            error_code: None,
-        }
     }
 }
 

nodedb/src/data/executor/handlers/transaction/mod.rs

@@ -7,4 +7,5 @@ mod sub_plan;
 mod sub_plan_doc;
 mod sub_plan_kv;
 mod sub_plan_kv_ops;
+mod sub_plan_write;
 pub(super) mod undo;