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

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