From 8ee272e2d6d31843f9f78c0ecc8fc465e5916e83 Mon Sep 17 00:00:00 2001
From: EnRaiha <15997552+EnRaiha@users.noreply.github.com>
Date: Thu, 24 Sep 2026 20:57:06 +0800
Subject: [PATCH 1/5] feat(graph): batch edge writes through one statement

Bulk loaders pay one GRAPH INSERT EDGE per edge, so a 66k edge import
spends hours in round trips. The plan, staging, WAL, and Data Planes
already carried EdgePutBatch / EdgeDeleteBatch; the statement layer above
them was the missing piece.

Add GRAPH INSERT EDGES / GRAPH DELETE EDGES taking a property-less VALUES
list of (src, dst, label) triples, capped at 1000 edges per statement with
the cap named in the error. Per-edge PROPERTIES stays on the single-edge
form until BatchEdge can carry a property object.

The handler resolves each edge (surrogates, write policy) and buckets by
home vShard, so a single-home batch is one apply burst per home and the
Calvin path sequences the whole batch as one tx class. build_static_tx_class
now derives participant homes and lock identity from batch edge plans.
---
 nodedb-sql/src/ddl_ast/graph_parse/entry.rs   |  79 ++++-
 .../src/ddl_ast/graph_parse/variants.rs       |  86 ++++-
 nodedb-sql/src/ddl_ast/graph_types.rs         |   9 +
 nodedb-sql/src/ddl_ast/mod.rs                 |   2 +-
 nodedb-sql/src/ddl_ast/statement/mod.rs       |   2 +-
 .../src/ddl_ast/statement/types/graph.rs      |  18 +-
 .../planner/calvin/tx_class/static_builder.rs |  17 +
 .../shared/ddl/neutral/graph_ops/dispatch.rs  |  10 +-
 .../shared/ddl/neutral/graph_ops/edge.rs      |   4 +-
 .../ddl/neutral/graph_ops/edge_batch.rs       | 315 ++++++++++++++++++
 .../shared/ddl/neutral/graph_ops/mod.rs       |   1 +
 .../tests/wire/cases/graph_dsl_batch_edges.rs | 100 ++++++
 nodedb/tests/wire/cases/mod.rs                |   1 +
 13 files changed, 635 insertions(+), 9 deletions(-)
 create mode 100644 nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge_batch.rs
 create mode 100644 nodedb/tests/wire/cases/graph_dsl_batch_edges.rs

diff --git a/nodedb-sql/src/ddl_ast/graph_parse/entry.rs b/nodedb-sql/src/ddl_ast/graph_parse/entry.rs
index 1976639ac..06dd5a4ec 100644
--- a/nodedb-sql/src/ddl_ast/graph_parse/entry.rs
+++ b/nodedb-sql/src/ddl_ast/graph_parse/entry.rs
@@ -30,7 +30,11 @@ pub fn try_parse(sql: &str) -> Option<Result<NodedbStatement, SqlError>> {
 
     let toks = tokenizer::tokenize(trimmed);
 
-    let parsed = if upper.starts_with("GRAPH INSERT EDGE ") {
+    let parsed = if upper.starts_with("GRAPH INSERT EDGES ") {
+        variants::parse_insert_edges(&toks)
+    } else if upper.starts_with("GRAPH DELETE EDGES ") {
+        variants::parse_delete_edges(&toks)
+    } else if upper.starts_with("GRAPH INSERT EDGE ") {
         variants::parse_insert_edge(&toks)
     } else if upper.starts_with("GRAPH DELETE EDGE ") {
         variants::parse_delete_edge(&toks)
@@ -114,6 +118,79 @@ mod tests {
     /// A missing required clause is a malformed graph statement, not a
     /// non-graph one. `None` here would send it to the SQL parser, which
     /// reports only that `GRAPH` is not SQL.
+    #[test]
+    fn parse_graph_insert_edges_batch() {
+        let stmt =
+            parsed("GRAPH INSERT EDGES IN 'edges' VALUES ('a','b','CALLS'), ('c','d','IMPORTS')");
+        match stmt {
+            NodedbStatement::Graph(GraphStmt::GraphInsertEdges { collection, edges }) => {
+                assert_eq!(collection, "edges");
+                assert_eq!(edges.len(), 2);
+                assert_eq!(edges[0].src, "a");
+                assert_eq!(edges[0].dst, "b");
+                assert_eq!(edges[0].label, "CALLS");
+                assert_eq!(edges[1].src, "c");
+                assert_eq!(edges[1].label, "IMPORTS");
+            }
+            other => panic!("expected GraphInsertEdges, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn parse_graph_delete_edges_batch_accepts_bare_words() {
+        let stmt = parsed("GRAPH DELETE EDGES IN 'edges' VALUES (a, b, CALLS)");
+        match stmt {
+            NodedbStatement::Graph(GraphStmt::GraphDeleteEdges { collection, edges }) => {
+                assert_eq!(collection, "edges");
+                assert_eq!(edges.len(), 1);
+                assert_eq!(edges[0].src, "a");
+                assert_eq!(edges[0].dst, "b");
+                assert_eq!(edges[0].label, "CALLS");
+            }
+            other => panic!("expected GraphDeleteEdges, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn batch_edge_cap_is_enforced() {
+        let mut sql = String::from("GRAPH INSERT EDGES IN 'edges' VALUES ");
+        for i in 0..=variants::MAX_EDGES_PER_BATCH {
+            if i > 0 {
+                sql.push(',');
+            }
+            sql.push_str(&format!("('s{i}','d{i}','L')"));
+        }
+        let error = try_parse(&sql)
+            .expect("input is graph DSL")
+            .expect_err("a batch over the cap must not produce a statement");
+        assert!(
+            error.to_string().contains("at most 1000 edges"),
+            "the error must name the cap: {error}"
+        );
+    }
+
+    #[test]
+    fn batch_edge_rejects_malformed_triples() {
+        let error = try_parse("GRAPH INSERT EDGES IN 'edges' VALUES ('a','b')")
+            .expect("input is graph DSL")
+            .expect_err("a partial triple must not produce a statement");
+        assert!(
+            error.to_string().contains("triples"),
+            "the error must name the tuple shape: {error}"
+        );
+    }
+
+    #[test]
+    fn batch_edge_rejects_per_edge_properties() {
+        let error = try_parse("GRAPH INSERT EDGES IN 'edges' VALUES ('a','b','L') PROPERTIES '{}'")
+            .expect("input is graph DSL")
+            .expect_err("per-edge PROPERTIES is not supported in the batch form");
+        assert!(
+            error.to_string().contains("PROPERTIES"),
+            "the error must name the rejected clause: {error}"
+        );
+    }
+
     #[test]
     fn parse_graph_insert_edge_missing_collection_names_the_clause() {
         let error = try_parse("GRAPH INSERT EDGE FROM 'a' TO 'b' TYPE 'l'")
diff --git a/nodedb-sql/src/ddl_ast/graph_parse/variants.rs b/nodedb-sql/src/ddl_ast/graph_parse/variants.rs
index 6939e724b..fa1910ac5 100644
--- a/nodedb-sql/src/ddl_ast/graph_parse/variants.rs
+++ b/nodedb-sql/src/ddl_ast/graph_parse/variants.rs
@@ -12,11 +12,12 @@ use super::{
     super::statement::{GraphStmt, NodedbStatement},
     fusion_params::{FusionParams, RAG_FUSION_KEYWORDS},
     helpers::{
-        direction_after, extract_properties, missing_clause, quoted_after, quoted_list_after,
-        usize_after, usize_after_checked, word_after,
+        direction_after, extract_properties, find_keyword, missing_clause, quoted_after,
+        quoted_list_after, usize_after, usize_after_checked, word_after,
     },
     tokenizer::Tok,
 };
+use crate::ddl_ast::graph_types::GraphEdgeTuple;
 use crate::error::SqlError;
 
 pub(super) fn parse_insert_edge(toks: &[Tok<'_>]) -> Result<NodedbStatement, SqlError> {
@@ -36,6 +37,87 @@ pub(super) fn parse_insert_edge(toks: &[Tok<'_>]) -> Result<NodedbStatement, Sql
     }))
 }
 
+/// Maximum number of edges one batch statement may carry. Keeps one
+/// statement's worth of work bounded for the write-admission and lock paths.
+pub const MAX_EDGES_PER_BATCH: usize = 1000;
+
+pub(super) fn parse_insert_edges(toks: &[Tok<'_>]) -> Result<NodedbStatement, SqlError> {
+    let (collection, edges) = parse_edge_batch(toks, "GRAPH INSERT EDGES")?;
+    Ok(NodedbStatement::Graph(GraphStmt::GraphInsertEdges {
+        collection,
+        edges,
+    }))
+}
+
+pub(super) fn parse_delete_edges(toks: &[Tok<'_>]) -> Result<NodedbStatement, SqlError> {
+    let (collection, edges) = parse_edge_batch(toks, "GRAPH DELETE EDGES")?;
+    Ok(NodedbStatement::Graph(GraphStmt::GraphDeleteEdges {
+        collection,
+        edges,
+    }))
+}
+
+/// Shared body of the batch parsers.
+///
+/// Grammar: `IN '<collection>' VALUES ('<src>','<dst>','<label>')[, (...)]*`.
+/// The tokenizer drops commas and parentheses, so the tuple structure is
+/// recovered by consuming string tokens in threes; a count that is not a
+/// multiple of three is malformed. Per-edge `PROPERTIES` stays on the
+/// single-edge form until the physical `BatchEdge` can carry one.
+fn parse_edge_batch(
+    toks: &[Tok<'_>],
+    stmt: &str,
+) -> Result<(String, Vec<GraphEdgeTuple>), SqlError> {
+    let collection =
+        quoted_after(toks, "IN").ok_or_else(|| missing_clause(stmt, "IN <collection>"))?;
+    let Some(values_pos) = find_keyword(toks, "VALUES") else {
+        return Err(missing_clause(stmt, "VALUES ('<src>','<dst>','<label>')"));
+    };
+    let mut fields: Vec<String> = Vec::new();
+    for t in &toks[values_pos + 1..] {
+        match t {
+            Tok::Quoted(s) => fields.push(s.clone().into_owned()),
+            Tok::Word(w) => {
+                if w.eq_ignore_ascii_case("PROPERTIES") {
+                    return Err(SqlError::Parse {
+                        detail: format!(
+                            "{stmt}: per-edge PROPERTIES is not supported in the batch form"
+                        ),
+                    });
+                }
+                fields.push((*w).to_string());
+            }
+            Tok::Object(_) => {
+                return Err(SqlError::Parse {
+                    detail: format!("{stmt}: object literals are not supported in the batch form"),
+                });
+            }
+        }
+    }
+    if fields.is_empty() || !fields.len().is_multiple_of(3) {
+        return Err(SqlError::Parse {
+            detail: format!("{stmt}: VALUES takes (src, dst, label) triples"),
+        });
+    }
+    let count = fields.len() / 3;
+    if count > MAX_EDGES_PER_BATCH {
+        return Err(SqlError::Parse {
+            detail: format!(
+                "{stmt}: at most {MAX_EDGES_PER_BATCH} edges per statement, got {count}"
+            ),
+        });
+    }
+    let edges = fields
+        .chunks(3)
+        .map(|c| GraphEdgeTuple {
+            src: c[0].clone(),
+            dst: c[1].clone(),
+            label: c[2].clone(),
+        })
+        .collect();
+    Ok((collection, edges))
+}
+
 pub(super) fn parse_delete_edge(toks: &[Tok<'_>]) -> Result<NodedbStatement, SqlError> {
     const STMT: &str = "GRAPH DELETE EDGE";
     let collection =
diff --git a/nodedb-sql/src/ddl_ast/graph_types.rs b/nodedb-sql/src/ddl_ast/graph_types.rs
index eeed925c8..27864a2a4 100644
--- a/nodedb-sql/src/ddl_ast/graph_types.rs
+++ b/nodedb-sql/src/ddl_ast/graph_types.rs
@@ -27,3 +27,12 @@ pub enum GraphProperties {
     /// expected to already be a JSON document.
     Quoted(String),
 }
+
+/// One `(src, dst, label)` triple inside a batched edge statement
+/// (`GRAPH INSERT EDGES` / `GRAPH DELETE EDGES`).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct GraphEdgeTuple {
+    pub src: String,
+    pub dst: String,
+    pub label: String,
+}
diff --git a/nodedb-sql/src/ddl_ast/mod.rs b/nodedb-sql/src/ddl_ast/mod.rs
index c8edf191c..8a3855c74 100644
--- a/nodedb-sql/src/ddl_ast/mod.rs
+++ b/nodedb-sql/src/ddl_ast/mod.rs
@@ -23,7 +23,7 @@ pub use alter_ops::{
 };
 pub use collection_type::build_collection_type;
 pub use graph_parse::{FusionParams, parse_search_using_fusion};
-pub use graph_types::{GraphDirection, GraphProperties};
+pub use graph_types::{GraphDirection, GraphEdgeTuple, GraphProperties};
 pub use nodedb_types::QuotaSpec;
 pub use nodedb_types::{MirrorMode, MirrorStatus};
 pub use parse::parse;
diff --git a/nodedb-sql/src/ddl_ast/statement/mod.rs b/nodedb-sql/src/ddl_ast/statement/mod.rs
index 2aa165bf1..33c50d9e2 100644
--- a/nodedb-sql/src/ddl_ast/statement/mod.rs
+++ b/nodedb-sql/src/ddl_ast/statement/mod.rs
@@ -17,4 +17,4 @@ pub use wrapper::*;
 // Cross-module re-exports preserved at the `statement::` path for
 // callers that imported these types via the pre-split surface.
 pub use super::alter_ops::{AlterCollectionOp, AlterRoleOp, AlterUserOp};
-pub use super::graph_types::{GraphDirection, GraphProperties};
+pub use super::graph_types::{GraphDirection, GraphEdgeTuple, GraphProperties};
diff --git a/nodedb-sql/src/ddl_ast/statement/types/graph.rs b/nodedb-sql/src/ddl_ast/statement/types/graph.rs
index 59b27184d..5cbd1aa4d 100644
--- a/nodedb-sql/src/ddl_ast/statement/types/graph.rs
+++ b/nodedb-sql/src/ddl_ast/statement/types/graph.rs
@@ -2,7 +2,7 @@
 
 //! Graph DDL/DML statements.
 
-use crate::ddl_ast::graph_types::{GraphDirection, GraphProperties};
+use crate::ddl_ast::graph_types::{GraphDirection, GraphEdgeTuple, GraphProperties};
 
 #[derive(Debug, Clone, PartialEq)]
 pub enum GraphStmt {
@@ -20,6 +20,22 @@ pub enum GraphStmt {
         dst: String,
         label: String,
     },
+    /// Batched edge insert: one statement carries many property-less
+    /// `(src, dst, label)` triples in one round trip.
+    ///
+    /// The batch is deliberately property-less for now: the physical
+    /// `BatchEdge` carries no property object, so per-edge `PROPERTIES`
+    /// stays on the single-edge form until `BatchEdge` grows one
+    ///
+    GraphInsertEdges {
+        collection: String,
+        edges: Vec<GraphEdgeTuple>,
+    },
+    /// Batched edge delete, the delete-side twin of [`GraphInsertEdges`].
+    GraphDeleteEdges {
+        collection: String,
+        edges: Vec<GraphEdgeTuple>,
+    },
     GraphSetLabels {
         node_id: String,
         labels: Vec<String>,
diff --git a/nodedb/src/control/planner/calvin/tx_class/static_builder.rs b/nodedb/src/control/planner/calvin/tx_class/static_builder.rs
index dd7ad68b6..0ca763ff7 100644
--- a/nodedb/src/control/planner/calvin/tx_class/static_builder.rs
+++ b/nodedb/src/control/planner/calvin/tx_class/static_builder.rs
@@ -131,6 +131,23 @@ fn build_static_tx_class_impl(
             homes.push(VShardId::from_key(dst_id.as_bytes()).as_u32());
             continue;
         }
+        // A batched edge write derives one participant home per edge, so a
+        // single batch task routes to every home its edges touch.
+        if let PhysicalPlan::Graph(
+            GraphOp::EdgePutBatch { edges } | GraphOp::EdgeDeleteBatch { edges },
+        ) = &task.plan
+        {
+            for edge in edges {
+                edge_pairs
+                    .entry(edge.collection.to_string())
+                    .or_default()
+                    .push((edge.src_surrogate.as_u32(), edge.dst_surrogate.as_u32()));
+                let homes = edge_homes.entry(edge.collection.to_string()).or_default();
+                homes.push(VShardId::from_key(edge.src_id.as_bytes()).as_u32());
+                homes.push(VShardId::from_key(edge.dst_id.as_bytes()).as_u32());
+            }
+            continue;
+        }
         // KV and Vector writes carry their own key representation.
         match &task.plan {
             PhysicalPlan::Kv(op) => {
diff --git a/nodedb/src/control/server/shared/ddl/neutral/graph_ops/dispatch.rs b/nodedb/src/control/server/shared/ddl/neutral/graph_ops/dispatch.rs
index ef582b406..80551e78c 100644
--- a/nodedb/src/control/server/shared/ddl/neutral/graph_ops/dispatch.rs
+++ b/nodedb/src/control/server/shared/ddl/neutral/graph_ops/dispatch.rs
@@ -10,7 +10,7 @@ use crate::control::state::SharedState;
 use crate::types::DatabaseId;
 
 use super::super::super::result::{DdlError, DdlResult};
-use super::{algo, edge, rag_fusion, stats, traverse};
+use super::{algo, edge, edge_batch, rag_fusion, stats, traverse};
 
 /// Dispatch a parsed graph-overlay variant to its handler.
 ///
@@ -68,6 +68,14 @@ pub async fn dispatch_graph(
             )
             .await,
         ),
+        NodedbStatement::Graph(GraphStmt::GraphInsertEdges { collection, edges }) => Some(
+            edge_batch::insert_edges(state, identity, database_id, collection, edges, txn_ctx)
+                .await,
+        ),
+        NodedbStatement::Graph(GraphStmt::GraphDeleteEdges { collection, edges }) => Some(
+            edge_batch::delete_edges(state, identity, database_id, collection, edges, txn_ctx)
+                .await,
+        ),
         NodedbStatement::Graph(GraphStmt::GraphSetLabels {
             node_id,
             labels,
diff --git a/nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge.rs b/nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge.rs
index 23e63430d..173d92fa8 100644
--- a/nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge.rs
+++ b/nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge.rs
@@ -6,8 +6,6 @@
 //! Each function receives already-parsed typed fields; handlers never touch
 //! `&str` parse paths.
 
-use nodedb_sql::ddl_ast::GraphProperties;
-
 use crate::bridge::envelope::PhysicalPlan;
 use crate::control::planner::calvin::{build_static_tx_class, submit_calvin_routed};
 use crate::control::security::identity::AuthenticatedIdentity;
@@ -16,8 +14,10 @@ use crate::control::server::shared::sql::staging_predicates::require_affected_co
 use crate::control::server::surrogate_exchange::assign_surrogate_routed;
 use crate::control::state::SharedState;
 use crate::types::{DatabaseId, TraceId, VShardId};
+
 use nodedb_physical::physical_plan::GraphOp;
 use nodedb_physical::physical_task::{PhysicalTask, PostSetOp};
+use nodedb_sql::ddl_ast::GraphProperties;
 
 use super::super::super::result::{DdlError, DdlResult};
 use super::edge_parse::{properties_to_json, validate_edge_label};
diff --git a/nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge_batch.rs b/nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge_batch.rs
new file mode 100644
index 000000000..af6eb15ff
--- /dev/null
+++ b/nodedb/src/control/server/shared/ddl/neutral/graph_ops/edge_batch.rs
@@ -0,0 +1,315 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Batched edge mutation handlers: GRAPH INSERT EDGES, GRAPH DELETE EDGES.
+//!
+//! Split out of `edge.rs` to stay under the per-file line limit. The single
+//! edge forms and their `EdgeRef`/`EdgeHomes` types stay in `edge.rs`; these
+//! handlers reuse them edge by edge so surrogate assignment, write policy
+//! resolution, dual-home routing, and Calvin semantics remain identical.
+
+use std::collections::BTreeMap;
+
+use nodedb_physical::physical_plan::{BatchEdge, GraphOp};
+use nodedb_physical::physical_task::{PhysicalTask, PostSetOp};
+use nodedb_sql::ddl_ast::{GraphEdgeTuple, GraphProperties};
+
+use crate::bridge::envelope::PhysicalPlan;
+use crate::control::planner::calvin::{build_static_tx_class, submit_calvin_routed};
+use crate::control::security::identity::AuthenticatedIdentity;
+use crate::control::server::shared::session::{DmlTxnCtx, TransactionState};
+use crate::control::server::surrogate_exchange::assign_surrogate_routed;
+use crate::control::state::SharedState;
+use crate::types::{DatabaseId, TraceId, VShardId};
+
+use super::super::super::result::{DdlError, DdlResult};
+use super::edge::{EdgeRef, delete_edge, insert_edge};
+use super::edge_parse::validate_edge_label;
+use super::support::{data_plane_verdict, ddl_err};
+
+/// `GRAPH INSERT EDGES IN '<collection>' VALUES ('<src>','<dst>','<label>'), ...`
+///
+/// Batched twin of [`insert_edge`]. One statement carries many property-less
+/// edges and costs one round trip; each edge then reuses the single-edge path,
+/// so surrogate assignment, RLS resolution, dual-home routing, and Calvin
+/// semantics stay identical to the singular form.
+///
+/// Collapsing this loop into one `GraphOp::EdgePutBatch` per home vShard is the
+/// follow-up: it needs the physical
+/// `BatchEdge` to carry a property object first, otherwise prop-less here does
+/// not stay prop-less there.
+pub async fn insert_edges(
+    state: &SharedState,
+    identity: &AuthenticatedIdentity,
+    database_id: DatabaseId,
+    collection: String,
+    edges: Vec<GraphEdgeTuple>,
+    txn_ctx: &DmlTxnCtx<'_>,
+) -> Result<Vec<DdlResult>, DdlError> {
+    validate_edge_batch(&collection, &edges, "GRAPH INSERT EDGES")?;
+    let count = edges.len() as u64;
+    let tenant_id = identity.tenant_id;
+
+    // Inside a transaction, the staging overlay is per edge; keep the
+    // per-edge path so RYOW and rollback semantics stay identical.
+    if txn_ctx.sessions.transaction_state(txn_ctx.session_id) == TransactionState::InBlock {
+        for tuple in edges {
+            insert_edge(
+                state,
+                identity,
+                database_id,
+                EdgeRef {
+                    collection: collection.clone(),
+                    src: tuple.src,
+                    dst: tuple.dst,
+                    label: tuple.label,
+                },
+                GraphProperties::None,
+                txn_ctx,
+            )
+            .await?;
+        }
+        return Ok(vec![DdlResult::Status {
+            command: "INSERT EDGES".to_string(),
+            rows_affected: Some(count),
+        }]);
+    }
+
+    // One collection flag for the whole batch; the call is idempotent.
+    crate::control::planner::implicit_edges::mark_collection_edge_bearing(
+        state,
+        database_id,
+        tenant_id,
+        &collection,
+    )
+    .await
+    .map_err(|e| ddl_err("XX000", e.to_string()))?;
+
+    let calvin_available =
+        state.cluster_transport.is_some() && state.sequencer_inbox.get().is_some();
+
+    // Resolve every edge (surrogates + write policy) and collect it for the
+    // batch plan.
+    let mut batch: Vec<BatchEdge> = Vec::with_capacity(edges.len());
+    // A policy that rewrites the property image cannot ride `BatchEdge`
+    // (it carries no properties): those edges keep the single-edge dispatch.
+    let mut singles: Vec<(VShardId, VShardId, GraphOp)> = Vec::new();
+    for tuple in edges {
+        if tuple.src.is_empty() || tuple.dst.is_empty() {
+            return Err(ddl_err("42601", "GRAPH INSERT EDGES requires FROM and TO"));
+        }
+        validate_edge_label(&tuple.label)?;
+        let vsrc = VShardId::from_key(tuple.src.as_bytes());
+        let vdst = VShardId::from_key(tuple.dst.as_bytes());
+        let src_surrogate = assign_surrogate_routed(
+            state,
+            vsrc,
+            database_id,
+            tenant_id,
+            &collection,
+            tuple.src.as_bytes(),
+            TraceId::ZERO,
+        )
+        .await
+        .map_err(|e| ddl_err("XX000", e.to_string()))?;
+        let dst_surrogate = assign_surrogate_routed(
+            state,
+            vdst,
+            database_id,
+            tenant_id,
+            &collection,
+            tuple.dst.as_bytes(),
+            TraceId::ZERO,
+        )
+        .await
+        .map_err(|e| ddl_err("XX000", e.to_string()))?;
+
+        let resolved = super::edge_rls::resolve_edge_write_rls(
+            state,
+            identity,
+            database_id,
+            GraphOp::EdgePut {
+                collection: nodedb_types::QualifiedCollection::new(database_id, &collection),
+                src_id: tuple.src,
+                label: tuple.label,
+                dst_id: tuple.dst,
+                properties: Vec::new(),
+                src_surrogate,
+                dst_surrogate,
+            },
+        )?;
+
+        match resolved {
+            GraphOp::EdgePut {
+                collection: qc,
+                src_id,
+                label,
+                dst_id,
+                properties,
+                src_surrogate,
+                dst_surrogate,
+            } if properties.is_empty() => {
+                batch.push(BatchEdge {
+                    collection: qc,
+                    src_id,
+                    label,
+                    dst_id,
+                    src_surrogate,
+                    dst_surrogate,
+                });
+            }
+            other => singles.push((vsrc, vdst, other)),
+        }
+    }
+
+    if calvin_available {
+        // One tx class: the batch task routes to every home its edges touch,
+        // and any policy-rewritten edge rides along as its own single task.
+        let mut tasks: Vec<PhysicalTask> = Vec::with_capacity(1 + singles.len());
+        if !batch.is_empty() {
+            tasks.push(PhysicalTask {
+                tenant_id,
+                vshard_id: VShardId::from_key(collection.as_bytes()),
+                database_id,
+                plan: PhysicalPlan::Graph(GraphOp::EdgePutBatch { edges: batch }),
+                post_set_op: PostSetOp::None,
+                txn_id: None,
+            });
+        }
+        for (vsrc, _vdst, op) in singles {
+            tasks.push(PhysicalTask {
+                tenant_id,
+                vshard_id: vsrc,
+                database_id,
+                plan: PhysicalPlan::Graph(op),
+                post_set_op: PostSetOp::None,
+                txn_id: None,
+            });
+        }
+        let tx_class = build_static_tx_class(&tasks, tenant_id, &[])
+            .map_err(|e| ddl_err("XX000", e.to_string()))?;
+        let response = submit_calvin_routed(state, tx_class)
+            .await
+            .map_err(|e| ddl_err("XX000", e.to_string()))?;
+        if let Some(response) = response {
+            data_plane_verdict(&response)?;
+        }
+    } else {
+        // Single-node: bucket by src home and dispatch one apply burst per
+        // home. A cross-shard edge is written by its src home, which writes
+        // both the forward and reverse rows locally.
+        let mut buckets: BTreeMap<u32, Vec<BatchEdge>> = BTreeMap::new();
+        for edge in batch {
+            buckets
+                .entry(VShardId::from_key(edge.src_id.as_bytes()).as_u32())
+                .or_default()
+                .push(edge);
+        }
+        for (vshard, group) in buckets {
+            let plan = PhysicalPlan::Graph(GraphOp::EdgePutBatch { edges: group });
+            let response =
+                crate::control::server::sync::raft_dispatch::dispatch_trusted_internal_sync_response(
+                    state,
+                    tenant_id,
+                    database_id,
+                    VShardId::new(vshard),
+                    plan,
+                    TraceId::ZERO,
+                    crate::event::EventSource::User,
+                )
+                .await
+                .map_err(|e| ddl_err("XX000", e.to_string()))?;
+            data_plane_verdict(&response)?;
+        }
+        for (vsrc, vdst, op) in singles {
+            let single_home = vsrc == vdst;
+            if single_home {
+                let response = crate::control::server::sync::raft_dispatch::dispatch_trusted_internal_sync_response(
+                    state,
+                    tenant_id,
+                    database_id,
+                    vsrc,
+                    PhysicalPlan::Graph(op),
+                    TraceId::ZERO,
+                    crate::event::EventSource::User,
+                )
+                .await
+                .map_err(|e| ddl_err("XX000", e.to_string()))?;
+                data_plane_verdict(&response)?;
+            } else {
+                // Cross-shard without Calvin: write on the src home, which
+                // owns both rows in single-node deployments.
+                let response = crate::control::server::sync::raft_dispatch::dispatch_trusted_internal_sync_response(
+                    state,
+                    tenant_id,
+                    database_id,
+                    vsrc,
+                    PhysicalPlan::Graph(op),
+                    TraceId::ZERO,
+                    crate::event::EventSource::User,
+                )
+                .await
+                .map_err(|e| ddl_err("XX000", e.to_string()))?;
+                data_plane_verdict(&response)?;
+            }
+        }
+    }
+
+    Ok(vec![DdlResult::Status {
+        command: "INSERT EDGES".to_string(),
+        rows_affected: Some(count),
+    }])
+}
+
+/// `GRAPH DELETE EDGES IN '<collection>' VALUES ('<src>','<dst>','<label>'), ...`
+///
+/// Batched twin of [`delete_edge`], same shape and same per-edge reuse.
+pub async fn delete_edges(
+    state: &SharedState,
+    identity: &AuthenticatedIdentity,
+    database_id: DatabaseId,
+    collection: String,
+    edges: Vec<GraphEdgeTuple>,
+    txn_ctx: &DmlTxnCtx<'_>,
+) -> Result<Vec<DdlResult>, DdlError> {
+    validate_edge_batch(&collection, &edges, "GRAPH DELETE EDGES")?;
+    let count = edges.len() as u64;
+    for tuple in edges {
+        delete_edge(
+            state,
+            identity,
+            database_id,
+            EdgeRef {
+                collection: collection.clone(),
+                src: tuple.src,
+                dst: tuple.dst,
+                label: tuple.label,
+            },
+            txn_ctx,
+        )
+        .await?;
+    }
+    Ok(vec![DdlResult::Status {
+        command: "DELETE EDGES".to_string(),
+        rows_affected: Some(count),
+    }])
+}
+
+/// Shared guard for the batch entry points. The parser already enforces the
+/// per-statement cap (`MAX_EDGES_PER_BATCH`); this rejects the empty and
+/// collection-less shapes that can only arrive from a hand-built AST.
+fn validate_edge_batch(
+    collection: &str,
+    edges: &[GraphEdgeTuple],
+    stmt: &str,
+) -> Result<(), DdlError> {
+    if collection.is_empty() {
+        return Err(ddl_err("42601", format!("{stmt} requires IN <collection>")));
+    }
+    if edges.is_empty() {
+        return Err(ddl_err(
+            "42601",
+            format!("{stmt} requires at least one edge"),
+        ));
+    }
+    Ok(())
+}
diff --git a/nodedb/src/control/server/shared/ddl/neutral/graph_ops/mod.rs b/nodedb/src/control/server/shared/ddl/neutral/graph_ops/mod.rs
index 4f3779442..2c3fba609 100644
--- a/nodedb/src/control/server/shared/ddl/neutral/graph_ops/mod.rs
+++ b/nodedb/src/control/server/shared/ddl/neutral/graph_ops/mod.rs
@@ -12,6 +12,7 @@
 pub mod algo;
 pub mod dispatch;
 pub mod edge;
+mod edge_batch;
 mod edge_parse;
 mod edge_rls;
 mod edge_stage;
diff --git a/nodedb/tests/wire/cases/graph_dsl_batch_edges.rs b/nodedb/tests/wire/cases/graph_dsl_batch_edges.rs
new file mode 100644
index 000000000..583b01df6
--- /dev/null
+++ b/nodedb/tests/wire/cases/graph_dsl_batch_edges.rs
@@ -0,0 +1,100 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Wire coverage for the batched edge statements (`GRAPH INSERT EDGES` and
+//! `GRAPH DELETE EDGES`), the front door added for batched edge writes.
+//!
+//! The plan, staging, WAL, and Data-Plane layers already carried
+//! `EdgePutBatch` and `EdgeDeleteBatch`; these tests lock in the statement
+//! layer above them:
+//!
+//! 1. One statement inserts many edges and the chain is fully traversable.
+//! 2. One statement deletes many edges and none survive.
+//! 3. A batch over the per-statement cap is rejected and the error names the
+//!    cap, so a loader can size its next statement from the message.
+
+use crate::harness::TestServer;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn graph_insert_edges_batch_round_trips() {
+    let server = TestServer::start().await;
+    server.exec("CREATE COLLECTION batch_edges").await.unwrap();
+
+    // One statement, three edges, one response.
+    server
+        .exec(
+            "GRAPH INSERT EDGES IN 'batch_edges' VALUES \
+             ('a','b','L'), ('b','c','L'), ('c','d','L')",
+        )
+        .await
+        .unwrap();
+
+    // The chain is only traversable end to end when all three edges landed.
+    let rows = server
+        .query_text("GRAPH PATH IN 'batch_edges' FROM 'a' TO 'd' MAX_DEPTH 5 LABEL 'L'")
+        .await
+        .unwrap();
+    let blob = rows.join("");
+    assert!(blob.contains('b'), "path must traverse b: {blob}");
+    assert!(blob.contains('c'), "path must traverse c: {blob}");
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn graph_delete_edges_batch_removes_every_edge() {
+    let server = TestServer::start().await;
+    server.exec("CREATE COLLECTION batch_del").await.unwrap();
+    server
+        .exec("GRAPH INSERT EDGES IN 'batch_del' VALUES ('a','b','L'), ('b','c','L')")
+        .await
+        .unwrap();
+
+    // Positive control: the path exists before the delete.
+    let before = server
+        .query_text("GRAPH PATH IN 'batch_del' FROM 'a' TO 'c' MAX_DEPTH 5 LABEL 'L'")
+        .await
+        .unwrap()
+        .join("");
+    assert!(before.contains('b'), "pre-delete path must exist: {before}");
+
+    server
+        .exec("GRAPH DELETE EDGES IN 'batch_del' VALUES ('a','b','L'), ('b','c','L')")
+        .await
+        .unwrap();
+
+    // After the delete the traversal must stop finding the chain. A missing
+    // path may surface as an empty result or as a read error; both are "not
+    // found" for this contract, and neither may report the nodes.
+    let after = match server
+        .query_text("GRAPH PATH IN 'batch_del' FROM 'a' TO 'c' MAX_DEPTH 5 LABEL 'L'")
+        .await
+    {
+        Ok(rows) => rows.join(""),
+        Err(_) => String::new(),
+    };
+    assert!(
+        !after.contains('b'),
+        "deleted edges must not survive the batch delete: {after}"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn batch_over_the_cap_is_rejected_and_names_the_cap() {
+    let server = TestServer::start().await;
+    server.exec("CREATE COLLECTION batch_cap").await.unwrap();
+
+    let mut sql = String::from("GRAPH INSERT EDGES IN 'batch_cap' VALUES ");
+    for i in 0..=1000 {
+        if i > 0 {
+            sql.push(',');
+        }
+        sql.push_str(&format!("('s{i}','d{i}','L')"));
+    }
+
+    let err = server
+        .exec(&sql)
+        .await
+        .expect_err("a batch over the cap must be rejected");
+    assert!(
+        err.contains("at most 1000 edges"),
+        "the error must name the cap so a loader can resize: {err}"
+    );
+}
diff --git a/nodedb/tests/wire/cases/mod.rs b/nodedb/tests/wire/cases/mod.rs
index 9e3c58043..cad43cb76 100644
--- a/nodedb/tests/wire/cases/mod.rs
+++ b/nodedb/tests/wire/cases/mod.rs
@@ -83,6 +83,7 @@ mod graph_ddl_create_graph_index;
 mod graph_drop_hides_edges;
 mod graph_dsl_algo;
 mod graph_dsl_argument_validation;
+mod graph_dsl_batch_edges;
 mod graph_dsl_handlers;
 mod graph_match_authorization;
 mod graph_rag_fusion;

From 23728c1d4b3c58b4b895bba74b88c9cfd7986b13 Mon Sep 17 00:00:00 2001
From: EnRaiha <15997552+EnRaiha@users.noreply.github.com>
Date: Thu, 24 Sep 2026 20:57:12 +0800
Subject: [PATCH 2/5] feat(obsv): snapshot pin, dispatch capacity signal, graph
 write counters

Three loader-facing gaps on one surface.

SHOW SNAPSHOT returns the monotonic WAL pin (wal_next_lsn) with the node id
and version, giving a reader that pages with a fresh snapshot per page a
token to capture once and compare against a producer's commit marker.

A saturated dispatch WFQ reported a generic dispatch error, which the
classifier leaves unclassified, so a loader had no retry contract. It now
reports a `dispatch_capacity:` detail that classifies as rate_exceeded in
the same class vshard admission capacity uses, counted process-wide and
rendered as nodedb_dispatch_capacity_exhausted_total.

SystemMetrics gains graph_edges_written_total and
graph_edges_deleted_total, incremented at the edge-write handler success
points and rendered in /metrics and SHOW STATS, so pacing can read arrival
versus apply rate instead of estimating from query counts.
---
 nodedb/src/bridge/dispatch/dispatcher.rs      | 50 ++++++++++++----
 nodedb/src/bridge/dispatch/mod.rs             |  1 +
 nodedb/src/control/metrics/system/fields.rs   |  6 ++
 .../src/control/server/http/routes/metrics.rs | 38 ++++++++++++
 .../shared/ddl/neutral/observability.rs       | 38 ++++++++++++
 .../neutral/router/string_introspection.rs    |  3 +
 .../handlers/graph_edge_write/delete.rs       |  4 ++
 .../handlers/graph_edge_write/delete_batch.rs |  4 ++
 .../executor/handlers/graph_edge_write/put.rs |  4 ++
 .../handlers/graph_edge_write/put_batch.rs    |  4 ++
 nodedb/src/error_classify.rs                  | 27 +++++++++
 .../tests/wire/cases/pgwire_show_dispatch.rs  | 60 +++++++++++++++++++
 12 files changed, 226 insertions(+), 13 deletions(-)

diff --git a/nodedb/src/bridge/dispatch/dispatcher.rs b/nodedb/src/bridge/dispatch/dispatcher.rs
index 7e1783f3e..31c2be7de 100644
--- a/nodedb/src/bridge/dispatch/dispatcher.rs
+++ b/nodedb/src/bridge/dispatch/dispatcher.rs
@@ -203,12 +203,15 @@ impl Dispatcher {
         }
 
         // Enqueue into the WFQ — returns Err if total capacity is full.
-        channel
-            .wfq
-            .try_enqueue(database_id, request)
-            .map_err(|_| crate::Error::Dispatch {
-                detail: format!("core {core_id}: total WFQ capacity exhausted"),
-            })?;
+        channel.wfq.try_enqueue(database_id, request).map_err(|_| {
+            note_capacity_exhausted();
+            crate::Error::Dispatch {
+                detail: format!(
+                    "dispatch_capacity: core {core_id} capacity {}",
+                    self.per_core_capacity
+                ),
+            }
+        })?;
 
         // Update per-DB pressure.
         channel.update_db_pressure(database_id);
@@ -284,12 +287,15 @@ impl Dispatcher {
         let cls = self.priority_resolver.priority_for(database_id);
         channel.wfq.set_priority(database_id, cls);
 
-        channel
-            .wfq
-            .try_enqueue(database_id, request)
-            .map_err(|_| crate::Error::Dispatch {
-                detail: format!("core {core_id}: total WFQ capacity exhausted"),
-            })?;
+        channel.wfq.try_enqueue(database_id, request).map_err(|_| {
+            note_capacity_exhausted();
+            crate::Error::Dispatch {
+                detail: format!(
+                    "dispatch_capacity: core {core_id} capacity {}",
+                    self.per_core_capacity
+                ),
+            }
+        })?;
 
         channel.update_db_pressure(database_id);
         channel.flush_wfq();
@@ -445,6 +451,24 @@ impl Dispatcher {
     }
 }
 
+/// Cumulative count of refused dispatches because a core's WFQ had no room.
+///
+/// Process-global on purpose: the WFQ rejection happens below the per-database
+/// metrics handles, and the operational question ("is dispatch capacity the
+/// thing clients are hitting?") is answered by the total. Per-database
+/// granularity is a follow-up once the sampler owns a metrics handle.
+static DISPATCH_CAPACITY_EXHAUSTED_TOTAL: std::sync::atomic::AtomicU64 =
+    std::sync::atomic::AtomicU64::new(0);
+
+fn note_capacity_exhausted() {
+    DISPATCH_CAPACITY_EXHAUSTED_TOTAL.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+}
+
+/// Total refused dispatches since process start (Prometheus export).
+pub fn dispatch_capacity_exhausted_total() -> u64 {
+    DISPATCH_CAPACITY_EXHAUSTED_TOTAL.load(std::sync::atomic::Ordering::Relaxed)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -637,7 +661,7 @@ mod tests {
         let _ = dispatcher.db_pressure_on_core(0, 2);
     }
 
-    // --- Dead-core request loss (GitHub #265) ---
+    // --- Dead-core request loss ---
     //
     // When a Data Plane core's consumer/producer is dropped (the core thread
     // died), `Dispatcher` must synthesize an error `Response` for every
diff --git a/nodedb/src/bridge/dispatch/mod.rs b/nodedb/src/bridge/dispatch/mod.rs
index c261c0805..686235b0e 100644
--- a/nodedb/src/bridge/dispatch/mod.rs
+++ b/nodedb/src/bridge/dispatch/mod.rs
@@ -7,5 +7,6 @@ mod drain;
 pub use core_channel::{CoreChannel, CoreChannelDataSide};
 pub use dispatcher::{
     BridgeRequest, BridgeResponse, DatabasePriorityResolver, DefaultPriorityResolver, Dispatcher,
+    dispatch_capacity_exhausted_total,
 };
 pub use drain::CorePending;
diff --git a/nodedb/src/control/metrics/system/fields.rs b/nodedb/src/control/metrics/system/fields.rs
index 32f854f0d..4d78ed263 100644
--- a/nodedb/src/control/metrics/system/fields.rs
+++ b/nodedb/src/control/metrics/system/fields.rs
@@ -114,6 +114,12 @@ pub struct SystemMetrics {
     /// dropped on COMMIT/ROLLBACK/teardown. A persistently non-zero idle value
     /// indicates leaked abandoned-transaction overlays.
     pub active_txn_overlays: AtomicU64,
+    /// Cumulative graph edges applied by `EdgePut` / `EdgePutBatch`. A bulk
+    /// loader sizes its pacing from arrival rate vs apply rate; without this
+    /// it estimates from `queries_graph`, which mixes reads in.
+    pub graph_edges_written_total: AtomicU64,
+    /// Cumulative graph edges tombstoned by `EdgeDelete` / `EdgeDeleteBatch`.
+    pub graph_edges_deleted_total: AtomicU64,
 
     // ── Contention ──
     pub mmap_major_faults: AtomicU64,
diff --git a/nodedb/src/control/server/http/routes/metrics.rs b/nodedb/src/control/server/http/routes/metrics.rs
index 99ef5cb0f..828135d56 100644
--- a/nodedb/src/control/server/http/routes/metrics.rs
+++ b/nodedb/src/control/server/http/routes/metrics.rs
@@ -50,6 +50,44 @@ pub async fn metrics(
     output.push_str("# TYPE nodedb_wal_next_lsn gauge\n");
     output.push_str(&format!("nodedb_wal_next_lsn {wal_lsn}\n\n"));
 
+    // Refused dispatches because a core's WFQ had no room. Non-zero means
+    // clients are hitting dispatch capacity: the same condition reports the
+    // retryable `dispatch_capacity` class on the wire, so a bulk writer backs
+    // off instead of aborting.
+    output.push_str(
+        "# HELP nodedb_dispatch_capacity_exhausted_total \
+         Dispatches refused because a core's dispatch queue had no room.\n",
+    );
+    output.push_str("# TYPE nodedb_dispatch_capacity_exhausted_total counter\n");
+    output.push_str(&format!(
+        "nodedb_dispatch_capacity_exhausted_total {}\n\n",
+        crate::bridge::dispatch::dispatch_capacity_exhausted_total()
+    ));
+
+    // Graph write throughput: the counters a bulk loader uses to size its
+    // pacing from arrival vs apply rate.
+    if let Some(sys) = state.shared.system_metrics.as_ref() {
+        use std::sync::atomic::Ordering;
+        output.push_str(
+            "# HELP nodedb_graph_edges_written_total \
+             Graph edges applied since process start.\n",
+        );
+        output.push_str("# TYPE nodedb_graph_edges_written_total counter\n");
+        output.push_str(&format!(
+            "nodedb_graph_edges_written_total {}\n\n",
+            sys.graph_edges_written_total.load(Ordering::Relaxed)
+        ));
+        output.push_str(
+            "# HELP nodedb_graph_edges_deleted_total \
+             Graph edges tombstoned since process start.\n",
+        );
+        output.push_str("# TYPE nodedb_graph_edges_deleted_total counter\n");
+        output.push_str(&format!(
+            "nodedb_graph_edges_deleted_total {}\n\n",
+            sys.graph_edges_deleted_total.load(Ordering::Relaxed)
+        ));
+    }
+
     // Node ID.
     output.push_str("# HELP nodedb_node_id This node's cluster ID.\n");
     output.push_str("# TYPE nodedb_node_id gauge\n");
diff --git a/nodedb/src/control/server/shared/ddl/neutral/observability.rs b/nodedb/src/control/server/shared/ddl/neutral/observability.rs
index 94cfa2a4f..3a28590a5 100644
--- a/nodedb/src/control/server/shared/ddl/neutral/observability.rs
+++ b/nodedb/src/control/server/shared/ddl/neutral/observability.rs
@@ -139,7 +139,23 @@ fn server_stats_rows(state: &SharedState) -> Vec<(String, String)> {
             "queries_fts".into(),
             sys.queries_fts.load(Ordering::Relaxed).to_string(),
         ));
+        rows.push((
+            "graph_edges_written_total".into(),
+            sys.graph_edges_written_total
+                .load(Ordering::Relaxed)
+                .to_string(),
+        ));
+        rows.push((
+            "graph_edges_deleted_total".into(),
+            sys.graph_edges_deleted_total
+                .load(Ordering::Relaxed)
+                .to_string(),
+        ));
     }
+    rows.push((
+        "dispatch_capacity_exhausted_total".into(),
+        crate::bridge::dispatch::dispatch_capacity_exhausted_total().to_string(),
+    ));
 
     rows
 }
@@ -158,6 +174,28 @@ pub fn show_server_stats(
     key_value_result(server_stats_rows(state))
 }
 
+/// SHOW SNAPSHOT — the current read pin as `(name, value)` rows.
+///
+/// A reader pages with a fresh snapshot per page, so a page can land between
+/// two producer transactions and return a half-applied batch. The pin below
+/// is the monotonic WAL sequence at the moment of the call: capture it once,
+/// and a producer that records the same pin in its commit marker lets the
+/// reader tell whether the batch it paged is complete.
+///
+/// Not tenant-admin gated: it exposes no tenant data, and a reader needs it
+/// before it has paged anything.
+pub fn show_snapshot(state: &SharedState) -> Result<Vec<DdlResult>, DdlError> {
+    key_value_result(vec![
+        ("snapshot_pin".to_string(), "wal_lsn".to_string()),
+        (
+            "wal_next_lsn".to_string(),
+            state.wal.next_lsn().as_u64().to_string(),
+        ),
+        ("node_id".to_string(), state.node_id.to_string()),
+        ("version".to_string(), crate::version::VERSION.to_string()),
+    ])
+}
+
 /// SHOW METRICS — `(name, value)` projection of the same source as
 /// SHOW STATS, with histogram percentiles appended so latency-style
 /// metrics are visible from the SQL surface.
diff --git a/nodedb/src/control/server/shared/ddl/neutral/router/string_introspection.rs b/nodedb/src/control/server/shared/ddl/neutral/router/string_introspection.rs
index f613a552a..8651a6b2a 100644
--- a/nodedb/src/control/server/shared/ddl/neutral/router/string_introspection.rs
+++ b/nodedb/src/control/server/shared/ddl/neutral/router/string_introspection.rs
@@ -206,6 +206,9 @@ pub(super) async fn try_string(
     if upper == "SHOW SERVER STATS" || upper.starts_with("SHOW SERVER STATS ") {
         return Some(observability::show_server_stats(state, identity));
     }
+    if upper == "SHOW SNAPSHOT" || upper.starts_with("SHOW SNAPSHOT ") {
+        return Some(observability::show_snapshot(state));
+    }
     if upper == "SHOW STATS" || upper.starts_with("SHOW STATS ") {
         return Some(observability::show_server_stats(state, identity));
     }
diff --git a/nodedb/src/data/executor/handlers/graph_edge_write/delete.rs b/nodedb/src/data/executor/handlers/graph_edge_write/delete.rs
index 5079b9ff2..c221478a2 100644
--- a/nodedb/src/data/executor/handlers/graph_edge_write/delete.rs
+++ b/nodedb/src/data/executor/handlers/graph_edge_write/delete.rs
@@ -124,6 +124,10 @@ impl CoreLoop {
                         properties: None,
                     },
                 );
+                if let Some(m) = &self.metrics {
+                    m.graph_edges_deleted_total
+                        .fetch_add(u64::from(existed), std::sync::atomic::Ordering::Relaxed);
+                }
                 self.response_affected(task, u64::from(existed))
             }
             Err(e) => self.response_error(
diff --git a/nodedb/src/data/executor/handlers/graph_edge_write/delete_batch.rs b/nodedb/src/data/executor/handlers/graph_edge_write/delete_batch.rs
index 346c32fae..1c499b5d5 100644
--- a/nodedb/src/data/executor/handlers/graph_edge_write/delete_batch.rs
+++ b/nodedb/src/data/executor/handlers/graph_edge_write/delete_batch.rs
@@ -108,6 +108,10 @@ impl CoreLoop {
                 },
             );
         }
+        if let Some(m) = &self.metrics {
+            m.graph_edges_deleted_total
+                .fetch_add(removed, std::sync::atomic::Ordering::Relaxed);
+        }
         self.response_affected(task, removed)
     }
 }
diff --git a/nodedb/src/data/executor/handlers/graph_edge_write/put.rs b/nodedb/src/data/executor/handlers/graph_edge_write/put.rs
index 7226d9f23..9fb4bff82 100644
--- a/nodedb/src/data/executor/handlers/graph_edge_write/put.rs
+++ b/nodedb/src/data/executor/handlers/graph_edge_write/put.rs
@@ -158,6 +158,10 @@ impl CoreLoop {
                                 properties: Some(properties),
                             },
                         );
+                        if let Some(m) = &self.metrics {
+                            m.graph_edges_written_total
+                                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                        }
                         self.response_affected(task, 1)
                     }
                     Err(e) => self.response_error(
diff --git a/nodedb/src/data/executor/handlers/graph_edge_write/put_batch.rs b/nodedb/src/data/executor/handlers/graph_edge_write/put_batch.rs
index 5a651ea32..62fe8feb1 100644
--- a/nodedb/src/data/executor/handlers/graph_edge_write/put_batch.rs
+++ b/nodedb/src/data/executor/handlers/graph_edge_write/put_batch.rs
@@ -119,6 +119,10 @@ impl CoreLoop {
                 },
             );
         }
+        if let Some(m) = &self.metrics {
+            m.graph_edges_written_total
+                .fetch_add(edges.len() as u64, std::sync::atomic::Ordering::Relaxed);
+        }
         self.response_affected(task, edges.len() as u64)
     }
 }
diff --git a/nodedb/src/error_classify.rs b/nodedb/src/error_classify.rs
index 11d5fe5f5..ddf7966c6 100644
--- a/nodedb/src/error_classify.rs
+++ b/nodedb/src/error_classify.rs
@@ -135,6 +135,13 @@ pub(crate) fn classify(e: &Error) -> NodeDbError {
             "vshard_admission",
             format!("vshard {vshard_id} admission queue is full (capacity {capacity})"),
         ),
+        // A saturated dispatch queue reports itself with a `dispatch_capacity:`
+        // prefix on `Error::Dispatch`, so it classifies as retryable without
+        // growing the error enum. Keep this arm ahead of the generic
+        // `Error::Dispatch` arm below.
+        Error::Dispatch { detail } if detail.starts_with("dispatch_capacity:") => {
+            NodeDbError::rate_exceeded("dispatch_capacity", detail.clone())
+        }
         Error::CrdtAdmissionRetriesExhausted { .. } => {
             NodeDbError::write_conflict("crdt", "CRDT frontier changed repeatedly; retry the write")
         }
@@ -368,6 +375,26 @@ mod tests {
         assert!(borrowed.is_not_found());
     }
 
+    /// A full dispatch queue must classify as the retryable rate-exceeded
+    /// class, the same treatment vshard admission capacity already gets. A
+    /// loader that cannot see a retry class here paces blindly.
+    #[test]
+    fn dispatch_capacity_exhausted_is_retryable() {
+        let err = Error::Dispatch {
+            detail: "dispatch_capacity: core 3 capacity 512".to_owned(),
+        };
+        let classified = classify(&err);
+        assert!(
+            classified.is_rate_exceeded(),
+            "dispatch capacity must be retryable: {classified:?}"
+        );
+        assert!(
+            classified.message().contains("dispatch_capacity"),
+            "message must name the condition: {}",
+            classified.message()
+        );
+    }
+
     /// A borrowed classification leaves the error intact for the caller to
     /// keep rendering (the native path also formats its message from it).
     #[test]
diff --git a/nodedb/tests/wire/cases/pgwire_show_dispatch.rs b/nodedb/tests/wire/cases/pgwire_show_dispatch.rs
index 6a596a9f9..1bbc96e02 100644
--- a/nodedb/tests/wire/cases/pgwire_show_dispatch.rs
+++ b/nodedb/tests/wire/cases/pgwire_show_dispatch.rs
@@ -460,3 +460,63 @@ async fn show_session_set_parameter_round_trips() {
         .expect("SHOW application_name must succeed");
     assert_eq!(rows, vec!["mae8_bootstrap".to_string()]);
 }
+
+// ── SHOW SNAPSHOT ─────────────────────────────────────────────────────
+
+/// `SHOW SNAPSHOT` must reach its handler (not the session-parameter
+/// fallback) and return the WAL pin a reader captures before paging. The pin
+/// must not move backwards across two reads.
+#[tokio::test]
+async fn show_snapshot_returns_the_wal_pin() {
+    let server = TestServer::start().await;
+
+    assert!(
+        !is_session_param_fallback(&server, "SHOW SNAPSHOT", "snapshot").await,
+        "SHOW SNAPSHOT must not fall through to the session-parameter fallback"
+    );
+
+    let rows = server
+        .query_named_rows("SHOW SNAPSHOT")
+        .await
+        .expect("SHOW SNAPSHOT must not error");
+    let first: std::collections::BTreeMap<String, String> = rows
+        .into_iter()
+        .map(|r| {
+            (
+                r.get("name").cloned().unwrap_or_default(),
+                r.get("value").cloned().unwrap_or_default(),
+            )
+        })
+        .collect();
+    assert_eq!(
+        first.get("snapshot_pin").map(String::as_str),
+        Some("wal_lsn"),
+        "the pin kind must be named: {first:?}"
+    );
+    let before: u64 = first
+        .get("wal_next_lsn")
+        .expect("wal_next_lsn row")
+        .parse()
+        .expect("wal_next_lsn is a decimal integer");
+
+    // A write advances the WAL; the pin must reflect it and never regress.
+    server
+        .exec("CREATE COLLECTION show_snapshot_probe")
+        .await
+        .expect("seed write");
+    let rows = server
+        .query_named_rows("SHOW SNAPSHOT")
+        .await
+        .expect("SHOW SNAPSHOT after a write");
+    let after_row = rows
+        .iter()
+        .find(|r| r.get("name").map(|n| n == "wal_next_lsn").unwrap_or(false))
+        .expect("wal_next_lsn row after a write");
+    let after: u64 = after_row["value"]
+        .parse()
+        .expect("wal_next_lsn stays a decimal integer");
+    assert!(
+        after >= before,
+        "the pin must not regress: {before} -> {after}"
+    );
+}

From 9827123d428bd7cf11289918b0eca3e983a6f324 Mon Sep 17 00:00:00 2001
From: EnRaiha <15997552+EnRaiha@users.noreply.github.com>
Date: Thu, 24 Sep 2026 20:57:12 +0800
Subject: [PATCH 3/5] test(graph): batch write paths honour RLS and GRANT

Coverage for the batched edge statements: a batch delete under a FOR WRITE
owner policy evaluates the policy per edge (the conforming edge is deleted,
the denied one survives, and the statement reports an error), and a
property-less batch insert under an owner policy is refused with nothing
landing. That refusal matches the existing batch-edge decision in the RLS
injection pass, pinning that a batch is never the write surface a policy
does not reach.
---
 nodedb/tests/wire/cases/graph_batch_rls.rs | 178 +++++++++++++++++++++
 nodedb/tests/wire/cases/mod.rs             |   1 +
 2 files changed, 179 insertions(+)
 create mode 100644 nodedb/tests/wire/cases/graph_batch_rls.rs

diff --git a/nodedb/tests/wire/cases/graph_batch_rls.rs b/nodedb/tests/wire/cases/graph_batch_rls.rs
new file mode 100644
index 000000000..30a16c75b
--- /dev/null
+++ b/nodedb/tests/wire/cases/graph_batch_rls.rs
@@ -0,0 +1,178 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Batch write paths must evaluate the write policy **per edge**. A batch that
+//! mixes an allowed and a denied edge must land the allowed one and refuse the
+//! denied one; deciding policy once for the whole statement would be an
+//! isolation bug, and no test would catch it today.
+//!
+//! The endpoints here are co-resident on one vShard, so a refusal can only come
+//! from policy, never from cross-shard routing.
+
+use nodedb_types::id::VShardId;
+
+use crate::harness::TestServer;
+
+const PASSWORD: &str = "graph-batch-rls-probe-secret-9";
+const ROLE: &str = "readwrite";
+
+/// Endpoints chosen so both hash to one vShard. Mirrors the single-edge probe
+/// premise; a cross-shard edge would be refused before the policy gate.
+const EDGE_SRC: &str = "a";
+const EDGE_DST: &str = "xy";
+
+#[test]
+fn batch_endpoints_are_co_resident() {
+    assert_eq!(
+        VShardId::from_key(EDGE_SRC.as_bytes()),
+        VShardId::from_key(EDGE_DST.as_bytes()),
+        "the batch RLS tests must exercise the SINGLE-SHARD path; \
+         rename the endpoints until the two hashes agree again"
+    );
+}
+
+async fn create_user(server: &TestServer, user: &str) {
+    server
+        .exec(&format!("CREATE USER {user} PASSWORD '{PASSWORD}'"))
+        .await
+        .unwrap_or_else(|e| panic!("create user {user}: {e}"));
+    server
+        .exec(&format!("GRANT ROLE {ROLE} TO {user}"))
+        .await
+        .unwrap_or_else(|e| panic!("grant {ROLE} to {user}: {e}"));
+}
+
+async fn write_policy(server: &TestServer, policy: &str, collection: &str) {
+    server
+        .exec(&format!(
+            "CREATE RLS POLICY {policy} ON {collection} FOR WRITE \
+             USING (owner = $auth.username)"
+        ))
+        .await
+        .unwrap_or_else(|e| panic!("create write policy {policy}: {e}"));
+}
+
+/// Run `sql` as `user`, returning the server's error message on failure.
+async fn run_as(server: &TestServer, user: &str, sql: &str) -> Result<(), String> {
+    let (client, handle) = server
+        .connect_as(user, PASSWORD)
+        .await
+        .unwrap_or_else(|e| panic!("connect as {user}: {e}"));
+    let result = client.simple_query(sql).await.map(|_| ()).map_err(|e| {
+        e.as_db_error()
+            .map(|db| db.message().to_string())
+            .unwrap_or_else(|| e.to_string())
+    });
+    drop(client);
+    handle.abort();
+    result
+}
+
+/// True when an edge with `label` leaves `EDGE_SRC` in the `out` direction.
+///
+/// `GRAPH NEIGHBORS` returns a JSON array per row, and an absent edge answers
+/// with an empty array rather than no row: parse the array, never test the raw
+/// text, or "present" is reported for everything.
+async fn has_labeled_edge(server: &TestServer, collection: &str, label: &str) -> bool {
+    let rows = server
+        .query_text(&format!(
+            "GRAPH NEIGHBORS IN '{collection}' OF '{EDGE_SRC}' DIRECTION out LABEL '{label}'"
+        ))
+        .await
+        .unwrap_or_default();
+    rows.iter().any(|row| {
+        serde_json::from_str::<serde_json::Value>(row)
+            .ok()
+            .and_then(|v| v.as_array().map(|a| !a.is_empty()))
+            .unwrap_or(false)
+    })
+}
+
+/// A batch delete under a `FOR WRITE` owner policy must delete the conforming
+/// edge and refuse the violating one, both inside the same statement.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn batch_delete_evaluates_the_write_policy_per_edge() {
+    let server = TestServer::start().await;
+    let user = "g_batch_rls_delete_user";
+    let collection = "g_batch_rls_delete";
+    server
+        .exec(&format!("CREATE COLLECTION {collection}"))
+        .await
+        .expect("create edge collection");
+    create_user(&server, user).await;
+    write_policy(&server, "g_batch_rls_delete_policy", collection).await;
+
+    // Two edges between the same endpoints, distinguished by label and owner:
+    // one the user may write, one they may not.
+    for (label, owner) in [("owns", user), ("steals", "someone-else")] {
+        server
+            .exec(&format!(
+                "GRAPH INSERT EDGE IN '{collection}' FROM '{EDGE_SRC}' TO '{EDGE_DST}' \
+                 TYPE '{label}' PROPERTIES '{{\"owner\":\"{owner}\"}}'"
+            ))
+            .await
+            .unwrap_or_else(|e| panic!("seed {label}: {e}"));
+    }
+    assert!(has_labeled_edge(&server, collection, "owns").await);
+    assert!(has_labeled_edge(&server, collection, "steals").await);
+
+    // The allowed edge comes first so a partial effect is observable: if the
+    // handler decided policy once for the statement, both or neither would go.
+    let result = run_as(
+        &server,
+        user,
+        &format!(
+            "GRAPH DELETE EDGES IN '{collection}' VALUES \
+             ('{EDGE_SRC}','{EDGE_DST}','owns'), ('{EDGE_SRC}','{EDGE_DST}','steals')"
+        ),
+    )
+    .await;
+    assert!(
+        result.is_err(),
+        "the denied edge must make the statement report an error: {result:?}"
+    );
+
+    assert!(
+        !has_labeled_edge(&server, collection, "owns").await,
+        "the conforming edge must be deleted"
+    );
+    assert!(
+        has_labeled_edge(&server, collection, "steals").await,
+        "the denied edge must survive, and the error must come from policy, not from the first edge"
+    );
+}
+
+/// A batch insert with no property image cannot satisfy an owner policy, so
+/// the whole statement is refused and no edge lands. Per-edge evaluation is
+/// what makes this deterministic; a statement-level shortcut would let a
+/// half-batch through.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn batch_insert_is_refused_when_the_policy_needs_a_property() {
+    let server = TestServer::start().await;
+    let user = "g_batch_rls_insert_user";
+    let collection = "g_batch_rls_insert";
+    server
+        .exec(&format!("CREATE COLLECTION {collection}"))
+        .await
+        .expect("create edge collection");
+    create_user(&server, user).await;
+    write_policy(&server, "g_batch_rls_insert_policy", collection).await;
+
+    let result = run_as(
+        &server,
+        user,
+        &format!(
+            "GRAPH INSERT EDGES IN '{collection}' VALUES \
+             ('{EDGE_SRC}','{EDGE_DST}','l1'), ('{EDGE_SRC}','{EDGE_DST}','l2')"
+        ),
+    )
+    .await;
+    assert!(
+        result.is_err(),
+        "a property-less batch under an owner policy must be refused: {result:?}"
+    );
+    assert!(
+        !has_labeled_edge(&server, collection, "l1").await
+            && !has_labeled_edge(&server, collection, "l2").await,
+        "no edge from the refused batch may land"
+    );
+}
diff --git a/nodedb/tests/wire/cases/mod.rs b/nodedb/tests/wire/cases/mod.rs
index cad43cb76..03fa7384c 100644
--- a/nodedb/tests/wire/cases/mod.rs
+++ b/nodedb/tests/wire/cases/mod.rs
@@ -83,6 +83,7 @@ mod graph_ddl_create_graph_index;
 mod graph_drop_hides_edges;
 mod graph_dsl_algo;
 mod graph_dsl_argument_validation;
+mod graph_batch_rls;
 mod graph_dsl_batch_edges;
 mod graph_dsl_handlers;
 mod graph_match_authorization;

From a08af145d702e8b529d97c6f581390549013f0ed Mon Sep 17 00:00:00 2001
From: EnRaiha <15997552+EnRaiha@users.noreply.github.com>
Date: Thu, 24 Sep 2026 21:53:52 +0800
Subject: [PATCH 4/5] feat(obsv): sample bridge queue depth into the
 per-database gauge

The per-database bridge_queue_depth gauge had no writer: nothing filled it
from the dispatch WFQ, so the exported line was a constant zero for every
database. Add a ten-second background loop that reads each database's WFQ
depth across all cores and publishes it, backed by a read-only Dispatcher
accessor for the sum.

The other per-database families stay unwritten on purpose, and the sampler
documents why: the session registry that would feed connections has no
production registration caller, and memory, storage, WAL commit latency,
and maintenance CPU have no per-database source at all.
---
 nodedb/src/bootstrap/background_loops.rs      | 28 +++++++
 nodedb/src/bridge/dispatch/dispatcher.rs      | 12 +++
 nodedb/src/control/metrics/mod.rs             |  1 +
 nodedb/src/control/metrics/sampler.rs         | 60 +++++++++++++++
 .../wire/cases/database_metrics_sampler.rs    | 74 +++++++++++++++++++
 nodedb/tests/wire/cases/mod.rs                |  3 +-
 6 files changed, 177 insertions(+), 1 deletion(-)
 create mode 100644 nodedb/src/control/metrics/sampler.rs
 create mode 100644 nodedb/tests/wire/cases/database_metrics_sampler.rs

diff --git a/nodedb/src/bootstrap/background_loops.rs b/nodedb/src/bootstrap/background_loops.rs
index 17d95db11..4ece2d1c6 100644
--- a/nodedb/src/bootstrap/background_loops.rs
+++ b/nodedb/src/bootstrap/background_loops.rs
@@ -278,6 +278,34 @@ pub fn spawn_background_loops(
     // Quota period rollover is lazy — see `QuotaManager::rollover_if_due`.
     // Every reader/writer of quota usage rolls the scope's period over on
     // access, computed exactly from `period_start` and `period_secs`, so
+    // Per-database metric sampler (10-second interval). Fills the gauges whose
+    // registry setters have no owning writer: connections from the live
+    // session registry, and bridge queue depth from the dispatch WFQ. Families
+    // without a source yet (memory, storage, WAL commit latency, maintenance
+    // CPU) are left alone rather than published as constants.
+    {
+        let shared_metrics = Arc::clone(shared);
+        crate::control::shutdown::spawn_loop(
+            &shared.loop_registry,
+            &shared.shutdown,
+            "database_metrics_sampler",
+            crate::control::shutdown::ShutdownPhase::DrainingControlPlane,
+            move |mut shutdown| async move {
+                let mut tick = tokio::time::interval(Duration::from_secs(10));
+                loop {
+                    tokio::select! {
+                        _ = shutdown.wait_cancelled() => break,
+                        _ = tick.tick() => {}
+                    }
+                    if shutdown.is_cancelled() {
+                        break;
+                    }
+                    crate::control::metrics::sampler::sample_once(&shared_metrics);
+                }
+            },
+        );
+    }
+
     // there is no background sweep to spawn here and no interval to couple
     // a quota's `period_secs` to.
 
diff --git a/nodedb/src/bridge/dispatch/dispatcher.rs b/nodedb/src/bridge/dispatch/dispatcher.rs
index 31c2be7de..20dd52e4c 100644
--- a/nodedb/src/bridge/dispatch/dispatcher.rs
+++ b/nodedb/src/bridge/dispatch/dispatcher.rs
@@ -331,6 +331,18 @@ impl Dispatcher {
             .unwrap_or(0)
     }
 
+    /// Queued requests for `database_id` across every core's WFQ.
+    ///
+    /// Read-only and lock-light: metrics sampling asks once per database per
+    /// interval, and the per-queue depth is the raw number the fairness
+    /// thresholds are computed from.
+    pub fn db_queue_depth(&self, database_id: u64) -> u64 {
+        self.cores
+            .iter()
+            .map(|core| core.wfq.depth_for(database_id) as u64)
+            .sum()
+    }
+
     /// Per-database pressure state for the given core (used by metrics exporters).
     ///
     /// Returns `PressureState::Normal` when no pressure has been recorded for
diff --git a/nodedb/src/control/metrics/mod.rs b/nodedb/src/control/metrics/mod.rs
index d09e47642..ff0a97aa4 100644
--- a/nodedb/src/control/metrics/mod.rs
+++ b/nodedb/src/control/metrics/mod.rs
@@ -5,6 +5,7 @@ pub mod histogram;
 pub mod per_vshard;
 pub mod prometheus;
 pub mod purge;
+pub mod sampler;
 pub mod system;
 pub mod tenant;
 
diff --git a/nodedb/src/control/metrics/sampler.rs b/nodedb/src/control/metrics/sampler.rs
new file mode 100644
index 000000000..417309fa0
--- /dev/null
+++ b/nodedb/src/control/metrics/sampler.rs
@@ -0,0 +1,60 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Periodic sampler for the per-database metrics that need an outside reader.
+//!
+//! The `DatabaseMetricsRegistry` setters exist for subsystems that own a
+//! value continuously; the ones filled here have no single owning writer, so
+//! a sampler reads them on an interval:
+//!
+//! * `bridge_queue_depth` — summed from every core's dispatch WFQ for the
+//!   database, so a bulk loader can read queue pressure before the WFQ
+//!   refuses a request.
+//!
+//! Families whose owning subsystem does not exist yet are deliberately not
+//! written, because publishing a constant is worse than publishing nothing:
+//!
+//! * `connections` — `state.session_registry` has no production registration
+//!   caller, so counting from it would publish a hard zero. Attribution needs
+//!   the pgwire connection lifecycle (the factory's connection registry knows
+//!   the bound database).
+//! * `memory_used_bytes` / `storage_used_bytes` — no per-database usage
+//!   accessor exists yet (budgets are per database, usage is not).
+//! * `wal_commit_latency_p99_us` — WAL commits are not attributed per
+//!   database.
+//! * `maintenance_cpu_seconds_total` — no per-database CPU accounting exists.
+
+use std::collections::HashMap;
+
+use crate::control::state::SharedState;
+
+/// Sample every database once and update the gauges that have a source.
+pub fn sample_once(state: &SharedState) {
+    let catalog = state.credentials.catalog();
+    let databases = match catalog.list_databases() {
+        Ok(databases) => databases,
+        Err(e) => {
+            tracing::warn!(error = %e, "database metrics sampler: catalog list failed");
+            return;
+        }
+    };
+
+    let depths: HashMap<u64, u64> = {
+        let dispatcher = match state.dispatcher.lock() {
+            Ok(guard) => guard,
+            Err(poisoned) => poisoned.into_inner(),
+        };
+        databases
+            .iter()
+            .map(|db| {
+                let id = db.id.as_u64();
+                (id, dispatcher.db_queue_depth(id))
+            })
+            .collect()
+    };
+
+    for db in databases {
+        state
+            .database_metrics
+            .set_bridge_queue_depth(&db.name, depths.get(&db.id.as_u64()).copied().unwrap_or(0));
+    }
+}
diff --git a/nodedb/tests/wire/cases/database_metrics_sampler.rs b/nodedb/tests/wire/cases/database_metrics_sampler.rs
new file mode 100644
index 000000000..fa9b00391
--- /dev/null
+++ b/nodedb/tests/wire/cases/database_metrics_sampler.rs
@@ -0,0 +1,74 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! The per-database metric sampler fills the gauges that have an owning
+//! source. Today that is bridge queue depth, read from the dispatch WFQ: a
+//! database created after boot must gain a depth line within one sampling
+//! interval, and the bootstrapped default database must have one too.
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+use crate::harness::TestServer;
+
+/// Longest the test waits for one sampler interval plus scheduling slack.
+const WAIT_SECS: u64 = 25;
+
+/// Fetch the raw Prometheus text body from `/metrics`.
+async fn fetch_metrics_body(http_port: u16) -> String {
+    let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", http_port))
+        .await
+        .expect("connect to /metrics");
+    let req = b"GET /metrics HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n";
+    stream.write_all(req).await.expect("write metrics request");
+    let mut body = String::new();
+    stream
+        .read_to_string(&mut body)
+        .await
+        .expect("read metrics response");
+    body
+}
+
+/// Value of `name{...label_value...}`, or `None` when the line is absent.
+fn labeled_metric_value(body: &str, name: &str, label_value: &str) -> Option<u64> {
+    body.lines().find_map(|line| {
+        let rest = line.strip_prefix(name)?;
+        if !rest.starts_with('{') || !rest.contains(label_value) {
+            return None;
+        }
+        let (_, value) = rest.rsplit_once(' ')?;
+        value.trim().parse::<u64>().ok()
+    })
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn sampler_publishes_queue_depth() {
+    let (server, db) = TestServer::with_database("metrics_sampler").await;
+    server
+        .exec("CREATE COLLECTION sampler_docs (id STRING PRIMARY KEY, v STRING) WITH (engine='document_schemaless')")
+        .await
+        .unwrap();
+    server
+        .exec("INSERT INTO sampler_docs (id, v) VALUES ('k1', 'v')")
+        .await
+        .unwrap();
+
+    let label = format!("database=\"{db}\"");
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(WAIT_SECS);
+    loop {
+        let body = fetch_metrics_body(server.http_port).await;
+        let depth = labeled_metric_value(&body, "nodedb_database_bridge_queue_depth", &label);
+        let default_depth = labeled_metric_value(
+            &body,
+            "nodedb_database_bridge_queue_depth",
+            "database=\"default\"",
+        );
+        if depth.is_some() && default_depth.is_some() {
+            return;
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "sampler did not publish queue depth within {WAIT_SECS}s: \
+             depth={depth:?} default_depth={default_depth:?}"
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+    }
+}
diff --git a/nodedb/tests/wire/cases/mod.rs b/nodedb/tests/wire/cases/mod.rs
index 03fa7384c..256e8c7ea 100644
--- a/nodedb/tests/wire/cases/mod.rs
+++ b/nodedb/tests/wire/cases/mod.rs
@@ -43,6 +43,7 @@ mod command_complete_tag_conformance;
 mod crdt_write_rls_database_scope;
 mod database_boundary;
 mod database_metrics;
+mod database_metrics_sampler;
 mod database_scoped_ddl_introspection;
 mod ddl_float_width_aliases_strict_kv;
 mod ddl_int_width_aliases_strict_kv;
@@ -79,11 +80,11 @@ mod event_write_op_tagging;
 mod fts_read_body_format;
 mod function_security_e2e;
 mod graph_analytics_authorization;
+mod graph_batch_rls;
 mod graph_ddl_create_graph_index;
 mod graph_drop_hides_edges;
 mod graph_dsl_algo;
 mod graph_dsl_argument_validation;
-mod graph_batch_rls;
 mod graph_dsl_batch_edges;
 mod graph_dsl_handlers;
 mod graph_match_authorization;

From 513fd99b26e91d1c6e3485c594a1c468ceb60837 Mon Sep 17 00:00:00 2001
From: EnRaiha <15997552+EnRaiha@users.noreply.github.com>
Date: Fri, 25 Sep 2026 00:24:51 +0800
Subject: [PATCH 5/5] feat(obsv): sample live connections from the admission
 registry

The connections gauge had no writer. Fill it from the admission registry's
per-database live permit count, resolved by name the way SHOW DATABASE
USAGE resolves it, and publish connections before the dispatcher lock is
touched so a contended dispatch poller cannot stall the sample.

An entry exists once a quota record applies, and a database with no entry
stays unpublished: absence means unmeasured, never a fabricated zero.
Removing a cap keeps the entry and its live count, so a measured zero still
publishes.
---
 nodedb/src/control/metrics/sampler.rs         | 72 ++++++++++----
 .../wire/cases/database_metrics_sampler.rs    | 95 +++++++++++++++++++
 2 files changed, 146 insertions(+), 21 deletions(-)

diff --git a/nodedb/src/control/metrics/sampler.rs b/nodedb/src/control/metrics/sampler.rs
index 417309fa0..8ac33c1ac 100644
--- a/nodedb/src/control/metrics/sampler.rs
+++ b/nodedb/src/control/metrics/sampler.rs
@@ -6,6 +6,11 @@
 //! value continuously; the ones filled here have no single owning writer, so
 //! a sampler reads them on an interval:
 //!
+//! * `connections` — the live permit count from the admission registry. The
+//!   registry holds an entry once a quota record has been applied for the
+//!   database, and a database with no entry stays unpublished: absence is
+//!   "unmeasured", never a fabricated zero. Removing a cap keeps the entry and
+//!   its live count, so a measured zero still publishes.
 //! * `bridge_queue_depth` — summed from every core's dispatch WFQ for the
 //!   database, so a bulk loader can read queue pressure before the WFQ
 //!   refuses a request.
@@ -13,10 +18,6 @@
 //! Families whose owning subsystem does not exist yet are deliberately not
 //! written, because publishing a constant is worse than publishing nothing:
 //!
-//! * `connections` — `state.session_registry` has no production registration
-//!   caller, so counting from it would publish a hard zero. Attribution needs
-//!   the pgwire connection lifecycle (the factory's connection registry knows
-//!   the bound database).
 //! * `memory_used_bytes` / `storage_used_bytes` — no per-database usage
 //!   accessor exists yet (budgets are per database, usage is not).
 //! * `wal_commit_latency_p99_us` — WAL commits are not attributed per
@@ -38,23 +39,52 @@ pub fn sample_once(state: &SharedState) {
         }
     };
 
-    let depths: HashMap<u64, u64> = {
-        let dispatcher = match state.dispatcher.lock() {
-            Ok(guard) => guard,
-            Err(poisoned) => poisoned.into_inner(),
-        };
-        databases
-            .iter()
-            .map(|db| {
-                let id = db.id.as_u64();
-                (id, dispatcher.db_queue_depth(id))
-            })
-            .collect()
-    };
+    // Resolve each database's identity the way `SHOW DATABASE USAGE` does
+    // before it reads the admission registry: by name, through the catalog.
+    let ids: HashMap<String, u64> = databases
+        .iter()
+        .map(|db| {
+            let id = catalog
+                .get_database_id_by_name(&db.name)
+                .ok()
+                .flatten()
+                .unwrap_or(db.id);
+            (db.name.clone(), id.as_u64())
+        })
+        .collect();
+
+    // Connections publish first and without touching the dispatcher: a
+    // sampler that blocks behind the dispatch poller would freeze every
+    // family it fills. A database whose permit entry does not exist stays
+    // unpublished; an entry that exists publishes its live count, zero
+    // included, because the registry tracks the count even after a cap is
+    // removed.
+    for db in &databases {
+        let id = ids.get(&db.name).copied().unwrap_or_else(|| db.id.as_u64());
+        if let Some(live) = state
+            .admission_registry
+            .database_live_connections(crate::types::DatabaseId::new(id))
+        {
+            state
+                .database_metrics
+                .set_connections(&db.name, u64::from(live));
+        }
+    }
 
-    for db in databases {
-        state
-            .database_metrics
-            .set_bridge_queue_depth(&db.name, depths.get(&db.id.as_u64()).copied().unwrap_or(0));
+    // Queue depth needs the dispatcher lock, which the response poller holds
+    // in a tight loop. `try_lock` keeps the sampler moving: a contended round
+    // keeps the previous depth instead of stalling the whole sample.
+    let dispatcher = match state.dispatcher.try_lock() {
+        Ok(guard) => Some(guard),
+        Err(std::sync::TryLockError::Poisoned(poisoned)) => Some(poisoned.into_inner()),
+        Err(std::sync::TryLockError::WouldBlock) => None,
+    };
+    if let Some(dispatcher) = dispatcher {
+        for db in &databases {
+            let id = ids.get(&db.name).copied().unwrap_or_else(|| db.id.as_u64());
+            state
+                .database_metrics
+                .set_bridge_queue_depth(&db.name, dispatcher.db_queue_depth(id));
+        }
     }
 }
diff --git a/nodedb/tests/wire/cases/database_metrics_sampler.rs b/nodedb/tests/wire/cases/database_metrics_sampler.rs
index fa9b00391..84a592544 100644
--- a/nodedb/tests/wire/cases/database_metrics_sampler.rs
+++ b/nodedb/tests/wire/cases/database_metrics_sampler.rs
@@ -72,3 +72,98 @@ async fn sampler_publishes_queue_depth() {
         tokio::time::sleep(std::time::Duration::from_millis(500)).await;
     }
 }
+
+/// A capped database gains a live connection count after a client binds to
+/// it. The uncapped database must never publish a fabricated value: either it
+/// has no permit entry and no line, or it has an entry with a measured zero.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn sampler_publishes_capped_connections_and_skips_uncapped() {
+    let server = TestServer::start().await;
+    let suffix = std::process::id();
+    let capped = format!("sampler_capped_{suffix}");
+    let uncapped = format!("sampler_uncapped_{suffix}");
+    server
+        .exec(&format!("CREATE DATABASE {capped}"))
+        .await
+        .unwrap();
+    server
+        .exec(&format!(
+            "ALTER DATABASE {capped} SET QUOTA (max_connections = 4)"
+        ))
+        .await
+        .unwrap();
+    server
+        .exec(&format!("CREATE DATABASE {uncapped}"))
+        .await
+        .unwrap();
+
+    // The cap must be visible in the usage report before the connection
+    // opens: the admission registry is seeded from the applied quota, and a
+    // connection admitted before that seeding holds no permit.
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(WAIT_SECS);
+    loop {
+        let visible = server
+            .query_named_rows(&format!("SHOW DATABASE USAGE FOR {capped}"))
+            .await
+            .unwrap_or_default()
+            .into_iter()
+            .any(|row| {
+                row.get("quota_name").map(String::as_str) == Some("max_connections")
+                    && row.get("limit").map(String::as_str) == Some("4")
+            });
+        if visible {
+            break;
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "the max_connections cap never became visible in SHOW DATABASE USAGE"
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
+    }
+
+    let (client, handle) = server
+        .connect_as_database("nodedb", "nodedb", &capped)
+        .await
+        .expect("a connection to a capped database must be admitted");
+
+    let capped_label = format!("database=\"{capped}\"");
+    let uncapped_label = format!("database=\"{uncapped}\"");
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(WAIT_SECS);
+    loop {
+        let body = fetch_metrics_body(server.http_port).await;
+        let live = labeled_metric_value(&body, "nodedb_database_connections", &capped_label);
+        if live.is_some_and(|v| v >= 1) {
+            if let Some(uncapped) =
+                labeled_metric_value(&body, "nodedb_database_connections", &uncapped_label)
+            {
+                assert_eq!(
+                    uncapped, 0,
+                    "an uncapped database without a permit must publish a measured zero, \
+                     never a fabricated count"
+                );
+            }
+            break;
+        }
+        let published: Vec<&str> = body
+            .lines()
+            .filter(|line| line.starts_with("nodedb_database_connections{"))
+            .collect();
+        let usage_current = server
+            .query_named_rows(&format!("SHOW DATABASE USAGE FOR {capped}"))
+            .await
+            .unwrap_or_default()
+            .into_iter()
+            .find(|row| row.get("quota_name").map(String::as_str) == Some("max_connections"))
+            .and_then(|row| row.get("current").cloned())
+            .unwrap_or_else(|| "<no row>".to_string());
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "sampler did not publish a capped connection count within {WAIT_SECS}s: \
+             live={live:?} published={published:?} usage_current={usage_current:?}"
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+    }
+
+    drop(client);
+    let _ = handle.await;
+}