feat(pgwire): add nodedb.read_consistency session parameter

Introduces `session/read_consistency.rs` with parsing and validation
for the `nodedb.read_consistency` SET parameter. Accepted values are
`strong`, `bounded_staleness:<secs>`, and `eventual`. Invalid values
are rejected at SET time with SQLSTATE 22023 and a descriptive
error message rather than silently ignored.

Author: farhan-syah

nodedb/src/control/server/pgwire/handler/session_cmds.rs

@@ -67,6 +67,19 @@ impl NodeDbPgHandler {
             }
         }
 
+        if key == super::super::session::read_consistency::PARAM_KEY
+            && super::super::session::read_consistency::parse_value(&value).is_none()
+        {
+            return Err(PgWireError::UserError(Box::new(ErrorInfo::new(
+                "ERROR".to_owned(),
+                "22023".to_owned(),
+                format!(
+                    "invalid value for {}: '{value}'. Valid: strong, bounded_staleness:<secs>, eventual",
+                    super::super::session::read_consistency::PARAM_KEY
+                ),
+            ))));
+        }
+
         if key == "nodedb.tenant_id" && value.parse::<u32>().is_err() {
             return Err(PgWireError::UserError(Box::new(ErrorInfo::new(
                 "ERROR".to_owned(),

nodedb/src/control/server/pgwire/session/mod.rs

@@ -2,6 +2,7 @@ mod cursor;
 pub mod cursor_spill;
 mod live;
 mod params;
+pub mod read_consistency;
 mod state;
 mod store;
 pub mod temp_tables;

nodedb/src/control/server/pgwire/session/read_consistency.rs

@@ -0,0 +1,155 @@
+//! Session-level `ReadConsistency` — wire `SET` / `SHOW` for the
+//! `default_read_consistency` session parameter.
+//!
+//! Accepted values:
+//!
+//! - `'strong'`
+//! - `'bounded_staleness:<secs>'` or `'bounded_staleness:<secs>s'`
+//! - `'eventual'`
+//!
+//! The value is stored as a plain string in the session parameter
+//! map. This module provides the typed parse + accessor.
+
+use std::net::SocketAddr;
+use std::time::Duration;
+
+use crate::types::ReadConsistency;
+
+use super::store::SessionStore;
+
+/// Session parameter key.
+pub const PARAM_KEY: &str = "default_read_consistency";
+
+/// Parse a user-supplied string into a `ReadConsistency`. Returns
+/// `None` on unrecognised input so the caller can return a helpful
+/// error message.
+pub fn parse_value(value: &str) -> Option<ReadConsistency> {
+    let lower = value.trim().to_lowercase();
+    match lower.as_str() {
+        "strong" => Some(ReadConsistency::Strong),
+        "eventual" => Some(ReadConsistency::Eventual),
+        _ => {
+            let stripped = lower.strip_prefix("bounded_staleness:")?;
+            let secs_str = stripped.trim_end_matches('s').trim();
+            let secs: f64 = secs_str.parse().ok()?;
+            if secs <= 0.0 {
+                return None;
+            }
+            Some(ReadConsistency::BoundedStaleness(Duration::from_secs_f64(
+                secs,
+            )))
+        }
+    }
+}
+
+/// Format a `ReadConsistency` back into the canonical string form
+/// so `SHOW default_read_consistency` returns something parseable.
+pub fn format_value(rc: &ReadConsistency) -> String {
+    match rc {
+        ReadConsistency::Strong => "strong".to_string(),
+        ReadConsistency::Eventual => "eventual".to_string(),
+        ReadConsistency::BoundedStaleness(d) => {
+            format!("bounded_staleness:{}s", d.as_secs_f64())
+        }
+    }
+}
+
+impl SessionStore {
+    /// Resolve the effective `ReadConsistency` for a session. Falls
+    /// back to `Strong` if the parameter is unset or unparseable.
+    pub fn read_consistency(&self, addr: &SocketAddr) -> ReadConsistency {
+        self.get_parameter(addr, PARAM_KEY)
+            .and_then(|v| parse_value(&v))
+            .unwrap_or_default()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_strong() {
+        assert_eq!(parse_value("strong"), Some(ReadConsistency::Strong));
+        assert_eq!(parse_value("STRONG"), Some(ReadConsistency::Strong));
+    }
+
+    #[test]
+    fn parse_eventual() {
+        assert_eq!(parse_value("eventual"), Some(ReadConsistency::Eventual));
+    }
+
+    #[test]
+    fn parse_bounded_staleness_seconds() {
+        let rc = parse_value("bounded_staleness:5").unwrap();
+        assert_eq!(
+            rc,
+            ReadConsistency::BoundedStaleness(Duration::from_secs(5))
+        );
+    }
+
+    #[test]
+    fn parse_bounded_staleness_with_s_suffix() {
+        let rc = parse_value("bounded_staleness:5s").unwrap();
+        assert_eq!(
+            rc,
+            ReadConsistency::BoundedStaleness(Duration::from_secs(5))
+        );
+    }
+
+    #[test]
+    fn parse_bounded_staleness_fractional() {
+        let rc = parse_value("bounded_staleness:0.5s").unwrap();
+        assert_eq!(
+            rc,
+            ReadConsistency::BoundedStaleness(Duration::from_millis(500))
+        );
+    }
+
+    #[test]
+    fn parse_rejects_zero_staleness() {
+        assert!(parse_value("bounded_staleness:0").is_none());
+    }
+
+    #[test]
+    fn parse_rejects_garbage() {
+        assert!(parse_value("foobar").is_none());
+        assert!(parse_value("").is_none());
+    }
+
+    #[test]
+    fn format_roundtrip_strong() {
+        let s = format_value(&ReadConsistency::Strong);
+        assert_eq!(parse_value(&s), Some(ReadConsistency::Strong));
+    }
+
+    #[test]
+    fn format_roundtrip_bounded() {
+        let rc = ReadConsistency::BoundedStaleness(Duration::from_secs(10));
+        let s = format_value(&rc);
+        assert_eq!(parse_value(&s), Some(rc));
+    }
+
+    #[test]
+    fn format_roundtrip_eventual() {
+        let s = format_value(&ReadConsistency::Eventual);
+        assert_eq!(parse_value(&s), Some(ReadConsistency::Eventual));
+    }
+
+    #[test]
+    fn session_store_defaults_to_strong() {
+        let store = SessionStore::new();
+        let addr: SocketAddr = "127.0.0.1:5432".parse().unwrap();
+        store.ensure_session(addr);
+        assert_eq!(store.read_consistency(&addr), ReadConsistency::Strong);
+    }
+
+    #[test]
+    fn session_store_reads_set_value() {
+        let store = SessionStore::new();
+        let addr: SocketAddr = "127.0.0.1:5432".parse().unwrap();
+        store.ensure_session(addr);
+        store.set_parameter(&addr, PARAM_KEY.to_string(), "eventual".to_string());
+        assert_eq!(store.read_consistency(&addr), ReadConsistency::Eventual);
+    }
+}