feat(sql): return the read pin from SHOW SNAPSHOT

A reader that pages with a fresh snapshot per page can land between two
producer transactions and return a half-applied batch, with no token to tell a
complete page from an incomplete one.

SHOW SNAPSHOT returns the monotonic WAL sequence at the moment of the call,
with the node id and the build version. Capture the pin once, and a producer
that records the same pin in its commit marker lets the reader decide whether
the batch it paged is complete.

Not tenant-admin gated: it exposes no tenant data, and a reader needs the pin
before it has paged anything.

Author: EnRaiha

nodedb/src/control/server/shared/ddl/neutral/observability.rs

@@ -158,6 +158,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.

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));
     }

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}"
+    );
+}