nodedb 0.3.0-beta.1

// SPDX-License-Identifier: BUSL-1.1

//! Session-level `CrossShardTxnMode` — wire `SET` / `SHOW` for the
//! `cross_shard_txn` session parameter.
//!
//! Accepted values:
//!
//! - `'strict'`
//! - `'best_effort_non_atomic'`
//!
//! The longer name for the non-atomic mode is intentional — it acts as a
//! foot-gun guard. `'best_effort'` alone is explicitly rejected.

use std::net::SocketAddr;

use super::store::SessionStore;

/// Session parameter key.
pub const PARAM_KEY: &str = "cross_shard_txn";

/// The cross-shard transaction mode for this session.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum CrossShardTxnMode {
    /// Full Calvin atomicity — cross-shard writes go through the sequencer.
    /// This is the default.
    #[default]
    Strict,
    /// Multi-vshard writes dispatched to each vshard independently —
    /// **NOT atomic.** Suitable for bulk loads only.
    BestEffortNonAtomic,
}

/// Parse a user-supplied string into a `CrossShardTxnMode`.
///
/// Returns `None` on unrecognised input (including `"best_effort"` without the
/// `_non_atomic` suffix) so the caller can return a helpful 22023 error.
pub fn parse_value(value: &str) -> Option<CrossShardTxnMode> {
    match value.trim().to_lowercase().as_str() {
        "strict" => Some(CrossShardTxnMode::Strict),
        "best_effort_non_atomic" => Some(CrossShardTxnMode::BestEffortNonAtomic),
        _ => None,
    }
}

/// Format a `CrossShardTxnMode` back into its canonical string form so that
/// `SHOW cross_shard_txn` returns something parseable.
pub fn format_value(mode: CrossShardTxnMode) -> &'static str {
    match mode {
        CrossShardTxnMode::Strict => "strict",
        CrossShardTxnMode::BestEffortNonAtomic => "best_effort_non_atomic",
    }
}

impl SessionStore {
    /// Resolve the effective `CrossShardTxnMode` for a session. Falls back to
    /// `Strict` if the parameter is unset or unparseable.
    pub fn cross_shard_txn_mode(&self, addr: &SocketAddr) -> CrossShardTxnMode {
        self.get_parameter(addr, PARAM_KEY)
            .and_then(|v| parse_value(&v))
            .unwrap_or_default()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parse_value_strict() {
        assert_eq!(parse_value("strict"), Some(CrossShardTxnMode::Strict));
        assert_eq!(parse_value("STRICT"), Some(CrossShardTxnMode::Strict));
        assert_eq!(parse_value("  strict  "), Some(CrossShardTxnMode::Strict));
    }

    #[test]
    fn parse_value_best_effort_non_atomic() {
        assert_eq!(
            parse_value("best_effort_non_atomic"),
            Some(CrossShardTxnMode::BestEffortNonAtomic)
        );
        assert_eq!(
            parse_value("BEST_EFFORT_NON_ATOMIC"),
            Some(CrossShardTxnMode::BestEffortNonAtomic)
        );
    }

    #[test]
    fn parse_value_rejects_bare_best_effort() {
        // The foot-gun guard: bare "best_effort" must be rejected.
        assert_eq!(parse_value("best_effort"), None);
    }

    #[test]
    fn parse_value_rejects_invalid() {
        assert_eq!(parse_value(""), None);
        assert_eq!(parse_value("foobar"), None);
        assert_eq!(parse_value("non_atomic"), None);
    }

    #[test]
    fn default_is_strict() {
        assert_eq!(CrossShardTxnMode::default(), CrossShardTxnMode::Strict);
    }

    #[test]
    fn session_store_defaults_to_strict() {
        let store = SessionStore::new();
        let addr: SocketAddr = "127.0.0.1:5432".parse().unwrap();
        store.ensure_session(addr);
        assert_eq!(store.cross_shard_txn_mode(&addr), CrossShardTxnMode::Strict);
    }

    #[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(),
            "best_effort_non_atomic".to_string(),
        );
        assert_eq!(
            store.cross_shard_txn_mode(&addr),
            CrossShardTxnMode::BestEffortNonAtomic
        );
    }
}