pond-db 0.20.3

// The embed worker and the ingest handler nest Lance futures 130 query-depth
// levels deep when rustc computes their layout, past the default limit of 128.
#![recursion_limit = "256"]

pub mod adapter;
pub mod config;
pub mod embed;
pub mod handlers;
pub mod memory;
// Bench/profiling instrumentation only: the module (and with it the
// `#[global_allocator]` swap) does not exist in a default build.
#[cfg(any(feature = "mem-probe", feature = "dhat-heap"))]
pub mod memprobe;
pub mod render;
pub mod rowmap;
pub mod sessions;
pub mod sql;
pub mod substrate;
pub mod transport;
pub mod wire;

pub const PROTOCOL_VERSION: u16 = 1;

use std::time::Duration;

use chrono::{DateTime, Utc};
use serde_json::Value;

pub trait Clock: Send + Sync {
    fn now(&self) -> DateTime<Utc>;
}

#[derive(Debug, Clone, Copy, Default)]
pub struct SystemClock;

impl Clock for SystemClock {
    fn now(&self) -> DateTime<Utc> {
        Utc::now()
    }
}

#[derive(Debug, Clone, Copy, PartialEq)]
pub struct RetryPolicy {
    pub attempts: u8,
    pub initial_backoff: Duration,
    pub max_backoff: Duration,
    /// Symmetric jitter factor applied to the exponential backoff: the sleep
    /// is multiplied by `1.0 + jitter * uniform(-1.0, 1.0)` before clamping
    /// to `max_backoff`. De-correlates concurrent retriers on a contended
    /// Lance manifest (spec.md#lance-retry-jitter).
    pub jitter: f64,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            attempts: 3,
            initial_backoff: Duration::from_millis(300),
            max_backoff: Duration::from_secs(5),
            jitter: 0.2,
        }
    }
}

pub mod output {
    use std::{
        io::{self, IsTerminal, Write},
        sync::OnceLock,
    };

    use anstyle::{AnsiColor, Style};
    use anyhow::Context;

    /// Whether the user-facing CLI surface should emit ANSI styling. Honors
    /// `NO_COLOR` (no-color.org) and falls back to plain text when stdout is
    /// not a TTY (piped to a file, captured by tests, ...). Cached so per-call
    /// overhead is a single pointer load.
    fn use_color() -> bool {
        static USE: OnceLock<bool> = OnceLock::new();
        *USE.get_or_init(|| std::env::var_os("NO_COLOR").is_none() && io::stdout().is_terminal())
    }

    /// Wrap `text` in `style`'s SGR sequence when color is enabled; return the
    /// raw text otherwise. The caller writes the result to stdout via
    /// [`line`].
    pub fn paint(text: &str, style: Style) -> String {
        if use_color() {
            format!("{}{text}{}", style.render(), style.render_reset())
        } else {
            text.to_owned()
        }
    }

    pub fn bold() -> Style {
        Style::new().bold()
    }
    pub fn dim() -> Style {
        Style::new().dimmed()
    }
    pub fn green() -> Style {
        Style::new().fg_color(Some(AnsiColor::Green.into()))
    }
    pub fn yellow() -> Style {
        Style::new().fg_color(Some(AnsiColor::Yellow.into()))
    }
    pub fn red() -> Style {
        Style::new().fg_color(Some(AnsiColor::Red.into()))
    }
    pub fn cyan() -> Style {
        Style::new().fg_color(Some(AnsiColor::Cyan.into()))
    }

    pub fn line(message: &str) -> anyhow::Result<()> {
        let mut stdout = io::stdout().lock();
        writeln!(stdout, "{message}")
            .inspect_err(exit_if_broken_pipe)
            .context("failed to write command output")
    }

    /// [`line`] for output that is already bytes, flushed so a closed pipe
    /// surfaces here rather than being dropped at exit.
    pub fn raw(bytes: &[u8]) -> anyhow::Result<()> {
        let mut stdout = io::stdout().lock();
        stdout
            .write_all(bytes)
            .and_then(|()| stdout.flush())
            .inspect_err(exit_if_broken_pipe)
            .context("failed to write command output")
    }

    /// Downstream (`pond ... | head`) closed the pipe, so stop quietly the way
    /// `cat` does. This replaces the Unix SIGPIPE default, which would also kill
    /// `pond serve` or a sync whenever a socket peer hangs up mid-write.
    pub fn exit_if_broken_pipe(error: &io::Error) {
        if error.kind() == io::ErrorKind::BrokenPipe {
            std::process::exit(0);
        }
    }

    /// Stderr counterpart to [`line`]. Per rust-cli/book stdout-vs-stderr
    /// discipline, anything that is not "the result" (progress, prompts,
    /// disclaimers, hints, interim status) belongs here so piping stdout to a
    /// file or another command yields the machine-readable view alone.
    pub fn line_err(message: &str) -> anyhow::Result<()> {
        let mut stderr = io::stderr().lock();
        writeln!(stderr, "{message}")
            .inspect_err(exit_if_broken_pipe)
            .context("failed to write command meta")
    }
}

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("validation failed: {message}")]
    Validation {
        message: String,
        field: Option<String>,
        value: Option<Value>,
        expected: Option<String>,
    },
    #[error("not found: {message}")]
    NotFound {
        message: String,
        kind: String,
        pk: Value,
    },
    #[error("namespace unknown: {namespace}")]
    NamespaceUnknown { namespace: String },
    #[error("commit conflict after {attempts} attempt(s)")]
    Conflict { attempts: u8 },
    #[error("storage unavailable: {0}")]
    Storage(#[from] anyhow::Error),
    #[error("internal error: {0}")]
    Internal(String),
}

impl Error {
    pub fn validation(message: impl Into<String>) -> Self {
        Self::Validation {
            message: message.into(),
            field: None,
            value: None,
            expected: None,
        }
    }

    pub fn validation_field(
        message: impl Into<String>,
        field: impl Into<String>,
        value: Option<Value>,
        expected: Option<String>,
    ) -> Self {
        Self::Validation {
            message: message.into(),
            field: Some(field.into()),
            value,
            expected,
        }
    }

    pub fn not_found(kind: impl Into<String>, pk: Value, message: impl Into<String>) -> Self {
        Self::NotFound {
            message: message.into(),
            kind: kind.into(),
            pk,
        }
    }

    pub fn namespace_unknown(namespace: impl Into<String>) -> Self {
        Self::NamespaceUnknown {
            namespace: namespace.into(),
        }
    }

    pub fn conflict(attempts: u8) -> Self {
        Self::Conflict { attempts }
    }

    pub fn internal(message: impl Into<String>) -> Self {
        Self::Internal(message.into())
    }
}

#[cfg(test)]
mod tests {
    #![expect(clippy::expect_used, reason = "tests fail by panicking")]

    use crate::adapter::test_support::{manifest_dir, rust_files};

    /// A compile-time read of a checkout path bakes it into the binary, so the
    /// compiler cache keys that binary per worktree and every new worktree
    /// recompiles it. Clippy's `disallowed_macros` cannot see an `env!` nested
    /// in `concat!`, so this scans the source text instead.
    #[test]
    fn no_compile_time_checkout_path_reads() {
        const PATH_VARS: [&str; 4] = [
            "CARGO_MANIFEST_DIR",
            "CARGO_MANIFEST_PATH",
            "CARGO_BIN_EXE_",
            "CARGO_TARGET_TMPDIR",
        ];
        let root = manifest_dir();
        let mut files = Vec::new();
        for dir in ["src", "tests", "benches"] {
            rust_files(&root.join(dir), &mut files);
        }
        let mut violations = Vec::new();
        for path in &files {
            let text = std::fs::read_to_string(path).expect("source is readable");
            for (at, needle) in text.match_indices("env!(") {
                let arg = text[at + needle.len()..].trim_start();
                let reads_path = arg
                    .strip_prefix('"')
                    .is_some_and(|name| PATH_VARS.iter().any(|var| name.starts_with(var)));
                if reads_path {
                    let line = text[..at].matches('\n').count() + 1;
                    let file = path.strip_prefix(&root).unwrap_or(path).display();
                    violations.push(format!("{file}:{line}"));
                }
            }
        }
        assert!(
            violations.is_empty(),
            "read these at runtime (`std::env::var_os`, a cwd-relative path, `pond_bin()`), \
             never via `env!`/`option_env!`: {violations:#?}"
        );
    }
}