reddb_server/
api.rs

1//! Public API layer for the RedDB crate.
2//!
3//! This module is the first layer to consume from applications:
4//! - stable options and contracts
5//! - capability declarations
6//! - typed errors and lightweight metadata snapshots
7//! - cross-layer traits for catalog/operations observability
8
9use std::collections::{BTreeMap, BTreeSet};
10use std::fmt;
11use std::io;
12use std::path::{Path, PathBuf};
13use std::sync::Arc;
14use std::time::{SystemTime, UNIX_EPOCH};
15
16use crate::auth::AuthConfig;
17use crate::replication::ReplicationConfig;
18
19pub const DEFAULT_SNAPSHOT_RETENTION: usize = 16;
20pub const DEFAULT_EXPORT_RETENTION: usize = 16;
21
22pub const REDDB_PROTOCOL_VERSION: &str = "reddb-v2";
23pub const REDDB_FORMAT_VERSION: u32 = 2;
24/// Default group-commit window.
25///
26/// `0` = "no wait" — the background flusher fsyncs as soon as any
27/// writer's pending commit arrives. Under single-writer workloads a
28/// non-zero window would be pure latency (no one to batch with),
29/// capping individual commit throughput at ~1000/s for window=1.
30/// Concurrent writers still batch naturally via the `Mutex<WalWriter>`
31/// contention path without needing an explicit timer.
32///
33/// Operators with many concurrent clients can raise this (e.g. 1-5ms)
34/// to amortise fsync cost across a bigger batch — at the cost of p99
35/// tail latency going up by the window size.
36pub const DEFAULT_GROUP_COMMIT_WINDOW_MS: u64 = 0;
37pub const DEFAULT_GROUP_COMMIT_MAX_STATEMENTS: usize = 128;
38pub const DEFAULT_GROUP_COMMIT_MAX_WAL_BYTES: u64 = 1024 * 1024;
39pub(crate) const EPHEMERAL_RUNTIME_METADATA_KEY: &str = "__reddb_ephemeral_runtime";
40
41pub type RedDBResult<T> = Result<T, RedDBError>;
42
43#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
44pub enum StorageMode {
45    /// Durable, file-backed database with WAL + checkpointing.
46    #[default]
47    Persistent,
48}
49
50impl StorageMode {
51    pub const fn is_persistent(self) -> bool {
52        matches!(self, Self::Persistent)
53    }
54}
55
56#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
57pub enum DurabilityMode {
58    #[default]
59    Strict,
60    WalDurableGrouped,
61    /// Fire-and-forget. Writers return as soon as the WAL record is
62    /// in the in-memory ring buffer — they do NOT wait for fsync.
63    /// The group-commit background thread still flushes on its
64    /// cadence, so committed-but-unflushed work is bounded by
65    /// `GroupCommitOptions::window_ms`. A crash inside the window
66    /// drops whatever wasn't flushed — matches PG's
67    /// `synchronous_commit=off` contract.
68    Async,
69}
70
71impl DurabilityMode {
72    pub const fn as_str(self) -> &'static str {
73        match self {
74            Self::Strict => "strict",
75            Self::WalDurableGrouped => "wal_durable_grouped",
76            Self::Async => "async",
77        }
78    }
79
80    pub fn from_str(value: &str) -> Option<Self> {
81        let normalized = value.trim().to_ascii_lowercase();
82        match normalized.as_str() {
83            // Legacy / opt-out form. Every commit pays its own fsync.
84            "strict" => Some(Self::Strict),
85            // Group-commit sync path — the perf-parity default. Matches
86            // PostgreSQL's `synchronous_commit=on` behaviour: the
87            // writer waits for durability, but fsyncs are batched
88            // across concurrent writers so a burst of N commits pays
89            // ~O(1) fsyncs instead of O(N).
90            "sync"
91            | "wal_durable_grouped"
92            | "wal-durable-grouped"
93            | "grouped"
94            | "wal_grouped"
95            | "wal-grouped" => Some(Self::WalDurableGrouped),
96            // Fire-and-forget async: writers return as soon as the
97            // WAL buffer accepts the record; background flusher runs
98            // on its configured cadence.
99            "async" | "fire_and_forget" | "fire-and-forget" => Some(Self::Async),
100            _ => None,
101        }
102    }
103}
104
105#[derive(Debug, Clone, Copy, PartialEq, Eq)]
106pub struct GroupCommitOptions {
107    pub window_ms: u64,
108    pub max_statements: usize,
109    pub max_wal_bytes: u64,
110}
111
112impl Default for GroupCommitOptions {
113    fn default() -> Self {
114        Self {
115            window_ms: DEFAULT_GROUP_COMMIT_WINDOW_MS,
116            max_statements: DEFAULT_GROUP_COMMIT_MAX_STATEMENTS,
117            max_wal_bytes: DEFAULT_GROUP_COMMIT_MAX_WAL_BYTES,
118        }
119    }
120}
121
122#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
123pub enum Capability {
124    /// Structured row storage.
125    Table,
126    /// Graph nodes/edges.
127    Graph,
128    /// Vector collections and ANN search.
129    Vector,
130    /// Full-text / lexical search.
131    FullText,
132    /// Text/metadata security and enrichment modules.
133    Security,
134    /// Encryption at rest.
135    Encryption,
136}
137
138impl Capability {
139    pub const fn as_str(self) -> &'static str {
140        match self {
141            Self::Table => "table",
142            Self::Graph => "graph",
143            Self::Vector => "vector",
144            Self::FullText => "fulltext",
145            Self::Security => "security",
146            Self::Encryption => "encryption",
147        }
148    }
149}
150
151#[derive(Debug, Clone, Default)]
152pub struct CapabilitySet {
153    items: BTreeSet<Capability>,
154}
155
156impl CapabilitySet {
157    pub fn new() -> Self {
158        Self::default()
159    }
160
161    pub fn with(mut self, capability: Capability) -> Self {
162        self.items.insert(capability);
163        self
164    }
165
166    pub fn with_all(mut self, capabilities: &[Capability]) -> Self {
167        capabilities.iter().copied().for_each(|capability| {
168            self.items.insert(capability);
169        });
170        self
171    }
172
173    pub fn has(&self, capability: Capability) -> bool {
174        self.items.contains(&capability)
175    }
176
177    pub fn as_slice(&self) -> Vec<Capability> {
178        self.items.iter().copied().collect()
179    }
180}
181
182pub struct RedDBOptions {
183    pub mode: StorageMode,
184    pub data_path: Option<PathBuf>,
185    pub read_only: bool,
186    pub create_if_missing: bool,
187    pub verify_checksums: bool,
188    pub durability_mode: DurabilityMode,
189    pub group_commit: GroupCommitOptions,
190    pub auto_checkpoint_pages: u32,
191    /// Maximum hypertable chunks the checkpoint path may seal into derived
192    /// columnar artifacts per checkpoint. `usize::MAX` means no practical cap.
193    pub checkpoint_columnar_emission_budget_chunks: usize,
194    /// Minimum rows a sealed in-scope chunk must carry before the derived
195    /// columnar projection is materialized.
196    pub columnar_projection_size_floor_rows: usize,
197    pub cache_pages: usize,
198    pub snapshot_retention: usize,
199    pub export_retention: usize,
200    pub feature_gates: CapabilitySet,
201    pub force_create: bool,
202    pub metadata: BTreeMap<String, String>,
203    /// Optional remote storage backend for snapshot transport.
204    pub remote_backend: Option<Arc<dyn crate::storage::backend::RemoteBackend>>,
205    /// Optional CAS-capable handle to the same backend, populated by
206    /// the factory when the configured backend implements
207    /// `AtomicRemoteBackend` (S3/local always; HTTP only when
208    /// `RED_HTTP_CONDITIONAL_WRITES=true`). `None` for backends that
209    /// do not provide compare-and-swap (Turso, D1, plain HTTP).
210    /// `LeaseStore` and any future CAS consumer pull from this field.
211    pub remote_backend_atomic: Option<Arc<dyn crate::storage::backend::AtomicRemoteBackend>>,
212    /// Remote object key used by the remote backend.
213    pub remote_key: Option<String>,
214    /// Exact immutable snapshot selected by the serverless publication head
215    /// for this boot. Internal orchestration sets it before opening the
216    /// runtime; ordinary callers should leave it as `None`.
217    #[doc(hidden)]
218    pub remote_bootstrap_key: Option<String>,
219    /// Replication configuration.
220    pub replication: ReplicationConfig,
221    /// Whether replica runtimes start the background replication loop.
222    pub replica_loop_enabled: bool,
223    /// Authentication & authorization configuration.
224    pub auth: AuthConfig,
225    /// Control Event Ledger configuration (issue #652). Read from
226    /// `REDDB_COMPLIANCE_MODE` at boot.
227    pub control_events: crate::runtime::control_events::ControlEventConfig,
228    /// Scoped data-plane query audit configuration. Disabled by
229    /// default; the regulated preset enables the stream without
230    /// adding catch-all rules.
231    pub query_audit: crate::runtime::query_audit::QueryAuditConfig,
232    /// Auto-create a HASH index on a user `id` column the first time a
233    /// row carrying that column is inserted into a collection. See
234    /// `UnifiedStoreConfig::auto_index_id`. Defaults to `true`; set to
235    /// `false` to opt out per workload (e.g. ingest pipelines that
236    /// don't need point-lookups by `id`).
237    pub auto_index_id: bool,
238    /// Tiered storage layout preset. Drives the defaults of the six
239    /// tier-flag toggles (`.meta.json` sidecar, seq-N catalog journal,
240    /// `-shm` provisioning, audit/slow log destinations,
241    /// `fold_dwb_into_wal`). Explicit per-feature setters still win over
242    /// the tier default. See `crate::storage::layout::StorageLayout`.
243    pub layout: crate::storage::layout::StorageLayout,
244    /// Per-feature overrides applied on top of the resolved layout preset.
245    pub layout_overrides: crate::storage::layout::LayoutOverrides,
246    /// Operator-facing storage/deploy profile contract. This records the
247    /// selected posture before the physical directory layout is expanded.
248    pub storage_profile: crate::storage::profile::StorageProfileSelection,
249    /// True only when the caller explicitly selected a layout via
250    /// `with_layout`. When false, `apply_tier_defaults` short-circuits so
251    /// pre-existing process-global toggles (set by tests / env hatches
252    /// before opening a runtime) are not clobbered by the implicit
253    /// `Standard` default. The new tier-driven behavior is opt-in.
254    pub layout_explicit: bool,
255    /// Explicit operator memory budget in bytes — the top tier of the ADR
256    /// 0073 §1 precedence chain. `None` defers to `REDDB_MEMORY_BUDGET`, the
257    /// deployment-profile default, the cgroup limit, and finally a fraction
258    /// of physical RAM. There is no "unlimited" value: `Some(0)` is rejected
259    /// at boot, exactly like an unparseable env var.
260    pub memory_budget_bytes: Option<u64>,
261    /// Opt into bounded resident segments and generation-bound disk indexes.
262    /// Currently supported only for standalone, single-file databases. Preparing
263    /// the file upgrades its writer format; retain an unconverted downgrade copy.
264    /// This is not a guarantee that every server workspace is memory-bounded.
265    pub bounded_disk_residency: bool,
266}
267
268impl fmt::Debug for RedDBOptions {
269    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
270        let backend_name = self.remote_backend.as_ref().map(|b| b.name().to_string());
271        f.debug_struct("RedDBOptions")
272            .field("mode", &self.mode)
273            .field("data_path", &self.data_path)
274            .field("read_only", &self.read_only)
275            .field("create_if_missing", &self.create_if_missing)
276            .field("verify_checksums", &self.verify_checksums)
277            .field("durability_mode", &self.durability_mode)
278            .field("group_commit", &self.group_commit)
279            .field("auto_checkpoint_pages", &self.auto_checkpoint_pages)
280            .field(
281                "checkpoint_columnar_emission_budget_chunks",
282                &self.checkpoint_columnar_emission_budget_chunks,
283            )
284            .field(
285                "columnar_projection_size_floor_rows",
286                &self.columnar_projection_size_floor_rows,
287            )
288            .field("cache_pages", &self.cache_pages)
289            .field("snapshot_retention", &self.snapshot_retention)
290            .field("export_retention", &self.export_retention)
291            .field("feature_gates", &self.feature_gates)
292            .field("force_create", &self.force_create)
293            .field("metadata", &self.metadata)
294            .field("remote_backend", &backend_name)
295            .field("remote_key", &self.remote_key)
296            .field("remote_bootstrap_key", &self.remote_bootstrap_key)
297            .field("replication", &self.replication)
298            .field("replica_loop_enabled", &self.replica_loop_enabled)
299            .field("auth", &self.auth)
300            .field("control_events", &self.control_events)
301            .field("query_audit", &self.query_audit)
302            .field("layout", &self.layout)
303            .field("layout_overrides", &self.layout_overrides)
304            .field("storage_profile", &self.storage_profile)
305            .field("memory_budget_bytes", &self.memory_budget_bytes)
306            .field("bounded_disk_residency", &self.bounded_disk_residency)
307            .finish()
308    }
309}
310
311impl Clone for RedDBOptions {
312    fn clone(&self) -> Self {
313        Self {
314            mode: self.mode,
315            data_path: self.data_path.clone(),
316            read_only: self.read_only,
317            create_if_missing: self.create_if_missing,
318            verify_checksums: self.verify_checksums,
319            durability_mode: self.durability_mode,
320            group_commit: self.group_commit,
321            auto_checkpoint_pages: self.auto_checkpoint_pages,
322            checkpoint_columnar_emission_budget_chunks: self
323                .checkpoint_columnar_emission_budget_chunks,
324            columnar_projection_size_floor_rows: self.columnar_projection_size_floor_rows,
325            cache_pages: self.cache_pages,
326            snapshot_retention: self.snapshot_retention,
327            export_retention: self.export_retention,
328            feature_gates: self.feature_gates.clone(),
329            force_create: self.force_create,
330            metadata: self.metadata.clone(),
331            remote_backend: self.remote_backend.clone(),
332            remote_backend_atomic: self.remote_backend_atomic.clone(),
333            remote_key: self.remote_key.clone(),
334            remote_bootstrap_key: self.remote_bootstrap_key.clone(),
335            replication: self.replication.clone(),
336            replica_loop_enabled: self.replica_loop_enabled,
337            auth: self.auth.clone(),
338            control_events: self.control_events,
339            query_audit: self.query_audit.clone(),
340            auto_index_id: self.auto_index_id,
341            layout: self.layout,
342            layout_overrides: self.layout_overrides.clone(),
343            storage_profile: self.storage_profile,
344            layout_explicit: self.layout_explicit,
345            memory_budget_bytes: self.memory_budget_bytes,
346            bounded_disk_residency: self.bounded_disk_residency,
347        }
348    }
349}
350
351impl Default for RedDBOptions {
352    fn default() -> Self {
353        Self {
354            mode: StorageMode::Persistent,
355            data_path: None,
356            read_only: false,
357            create_if_missing: true,
358            verify_checksums: true,
359            // Perf-parity default — `WalDurableGrouped` matches
360            // PostgreSQL's `synchronous_commit=on` behaviour while
361            // amortising fsync cost across concurrent writers. The
362            // legacy `Strict` tier (per-commit fsync) stays available
363            // via `durability.mode = "strict"` / `REDDB_DURABILITY=strict`.
364            durability_mode: DurabilityMode::WalDurableGrouped,
365            group_commit: GroupCommitOptions::default(),
366            auto_checkpoint_pages: 1000,
367            checkpoint_columnar_emission_budget_chunks: usize::MAX,
368            columnar_projection_size_floor_rows: 4,
369            cache_pages: 10_000,
370            snapshot_retention: DEFAULT_SNAPSHOT_RETENTION,
371            export_retention: DEFAULT_EXPORT_RETENTION,
372            feature_gates: CapabilitySet::new()
373                .with(Capability::Table)
374                .with(Capability::Graph)
375                .with(Capability::Vector),
376            force_create: true,
377            metadata: BTreeMap::new(),
378            remote_backend: None,
379            remote_backend_atomic: None,
380            remote_key: None,
381            remote_bootstrap_key: None,
382            replication: ReplicationConfig::standalone(),
383            replica_loop_enabled: true,
384            auth: AuthConfig::default(),
385            control_events: crate::runtime::control_events::ControlEventConfig::default(),
386            query_audit: crate::runtime::query_audit::QueryAuditConfig::default(),
387            auto_index_id: true,
388            layout: crate::storage::layout::StorageLayout::default(),
389            layout_overrides: crate::storage::layout::LayoutOverrides::default(),
390            storage_profile: crate::storage::profile::StorageProfileSelection::embedded_single_file(
391            ),
392            layout_explicit: false,
393            memory_budget_bytes: None,
394            bounded_disk_residency: false,
395        }
396    }
397}
398
399impl RedDBOptions {
400    pub fn persistent<P: Into<PathBuf>>(path: P) -> Self {
401        Self {
402            mode: StorageMode::Persistent,
403            data_path: Some(path.into()),
404            ..Default::default()
405        }
406    }
407
408    /// Ephemeral, tempfile-backed database.
409    ///
410    /// The underlying storage is a real persistent file placed under the system
411    /// temp directory with a unique name — there is no longer a true in-memory
412    /// execution mode. Prefer [`RedDBOptions::persistent`] when the data should
413    /// outlive the process.
414    pub fn in_memory() -> Self {
415        static NEXT_EPHEMERAL_ID: std::sync::atomic::AtomicU64 =
416            std::sync::atomic::AtomicU64::new(0);
417
418        let now_nanos = std::time::SystemTime::now()
419            .duration_since(std::time::UNIX_EPOCH)
420            .map(|duration| duration.as_nanos())
421            .unwrap_or(0);
422        let unique = NEXT_EPHEMERAL_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
423        // Each ephemeral runtime gets its OWN directory, not just a unique
424        // filename in the shared temp dir. Sibling artifacts (the audit log,
425        // WAL, snapshots) are derived from this data path's PARENT, so a
426        // shared parent collapses every ephemeral runtime's `.audit.log`
427        // onto one file — which parallel test processes (nextest runs each
428        // test in its own process, frequently under one shared `TMPDIR`)
429        // then truncate/remove out from under one another. A per-instance
430        // directory keeps each runtime's siblings self-contained.
431        let dir = std::env::temp_dir().join(format!(
432            "reddb-ephemeral-{}-{}-{}",
433            std::process::id(),
434            now_nanos,
435            unique
436        ));
437        let _ = std::fs::create_dir_all(&dir);
438        let path = dir.join("db.rdb");
439        let _ = std::fs::remove_file(&path);
440        let mut metadata = BTreeMap::new();
441        metadata.insert(
442            EPHEMERAL_RUNTIME_METADATA_KEY.to_string(),
443            "true".to_string(),
444        );
445        Self {
446            mode: StorageMode::Persistent,
447            data_path: Some(path),
448            auto_checkpoint_pages: 0,
449            checkpoint_columnar_emission_budget_chunks: usize::MAX,
450            columnar_projection_size_floor_rows: 4,
451            cache_pages: 2_000,
452            snapshot_retention: DEFAULT_SNAPSHOT_RETENTION,
453            export_retention: DEFAULT_EXPORT_RETENTION,
454            read_only: false,
455            force_create: true,
456            metadata,
457            ..Default::default()
458        }
459    }
460
461    pub fn with_mode(mut self, mode: StorageMode) -> Self {
462        self.mode = mode;
463        self
464    }
465
466    pub fn with_data_path<P: Into<PathBuf>>(mut self, path: P) -> Self {
467        // `in_memory()` eagerly creates a `reddb-ephemeral-*` dir and points
468        // `data_path` at it. Overriding the path abandons that dir — and the
469        // per-runtime cleanup keys off the FINAL `data_path`, so the orphan
470        // would leak in TMPDIR (caught by scripts/check-temp-residue.sh). Remove
471        // it now if the previous path was such an eagerly-created ephemeral dir.
472        if let Some(old) = self.data_path.take() {
473            let tmp = std::env::temp_dir();
474            if let Some(parent) = old.parent() {
475                let orphaned = parent
476                    .file_name()
477                    .and_then(|name| name.to_str())
478                    .is_some_and(|name| name.starts_with("reddb-ephemeral-"))
479                    && parent.parent() == Some(tmp.as_path());
480                if orphaned {
481                    let _ = std::fs::remove_dir_all(parent);
482                }
483            }
484        }
485        self.data_path = Some(path.into());
486        self
487    }
488
489    pub fn with_read_only(mut self, read_only: bool) -> Self {
490        self.read_only = read_only;
491        self
492    }
493
494    pub fn with_auto_checkpoint(mut self, pages: u32) -> Self {
495        self.auto_checkpoint_pages = pages;
496        self
497    }
498
499    pub fn with_checkpoint_columnar_emission_budget_chunks(mut self, chunks: usize) -> Self {
500        self.checkpoint_columnar_emission_budget_chunks = chunks;
501        self
502    }
503
504    pub fn with_columnar_projection_size_floor_rows(mut self, rows: usize) -> Self {
505        self.columnar_projection_size_floor_rows = rows;
506        self
507    }
508
509    pub fn with_durability_mode(mut self, mode: DurabilityMode) -> Self {
510        self.durability_mode = mode;
511        self
512    }
513
514    pub fn with_group_commit_window_ms(mut self, window_ms: u64) -> Self {
515        // `0` is a legitimate setting — "no wait, fsync on every wakeup".
516        // See `DEFAULT_GROUP_COMMIT_WINDOW_MS` docs for the tradeoff.
517        self.group_commit.window_ms = window_ms;
518        self
519    }
520
521    pub fn with_group_commit_max_statements(mut self, max_statements: usize) -> Self {
522        self.group_commit.max_statements = max_statements.max(1);
523        self
524    }
525
526    pub fn with_group_commit_max_wal_bytes(mut self, max_wal_bytes: u64) -> Self {
527        self.group_commit.max_wal_bytes = max_wal_bytes.max(1);
528        self
529    }
530
531    pub fn with_cache_pages(mut self, pages: usize) -> Self {
532        self.cache_pages = pages.max(2);
533        self
534    }
535
536    pub fn with_snapshot_retention(mut self, limit: usize) -> Self {
537        self.snapshot_retention = limit.max(1);
538        self
539    }
540
541    pub fn with_export_retention(mut self, limit: usize) -> Self {
542        self.export_retention = limit.max(1);
543        self
544    }
545
546    pub fn with_metadata<K: Into<String>, V: Into<String>>(mut self, key: K, value: V) -> Self {
547        self.metadata.insert(key.into(), value.into());
548        self
549    }
550
551    /// Toggle the implicit HASH index on user `id` columns at first
552    /// insert (#112). Defaults to enabled — pass `false` to fall back
553    /// to the legacy "scan unless `CREATE INDEX` is issued" behaviour.
554    pub fn with_auto_index_id(mut self, enabled: bool) -> Self {
555        self.auto_index_id = enabled;
556        self
557    }
558
559    pub fn with_capability(mut self, capability: Capability) -> Self {
560        self.feature_gates = self.feature_gates.with(capability);
561        self
562    }
563
564    /// Attach a remote storage backend for snapshot transport.
565    ///
566    /// On open, the database snapshot is downloaded from the remote `key`
567    /// to the local data path. On flush, the local file is uploaded back
568    /// to the remote backend under the same key.
569    pub fn with_remote_backend(
570        mut self,
571        backend: Arc<dyn crate::storage::backend::RemoteBackend>,
572        key: impl Into<String>,
573    ) -> Self {
574        self.remote_backend = Some(backend);
575        self.remote_key = Some(key.into());
576        self
577    }
578
579    /// Attach a CAS-capable backend handle. Pass the same `Arc` as
580    /// `with_remote_backend` (factories should construct the backend
581    /// once and call both setters); this method exists so the type
582    /// system, not runtime config, decides whether `LeaseStore` is
583    /// reachable.
584    pub fn with_atomic_remote_backend(
585        mut self,
586        backend: Arc<dyn crate::storage::backend::AtomicRemoteBackend>,
587    ) -> Self {
588        self.remote_backend_atomic = Some(backend);
589        self
590    }
591
592    pub fn with_replication(mut self, config: ReplicationConfig) -> Self {
593        self.replication = config;
594        self
595    }
596
597    pub fn with_replica_loop_enabled(mut self, enabled: bool) -> Self {
598        self.replica_loop_enabled = enabled;
599        self
600    }
601
602    pub fn with_auth(mut self, config: AuthConfig) -> Self {
603        self.auth = config;
604        self
605    }
606
607    pub fn resolved_path(&self, fallback: impl AsRef<Path>) -> PathBuf {
608        self.data_path
609            .clone()
610            .unwrap_or_else(|| fallback.as_ref().to_path_buf())
611    }
612
613    pub fn remote_namespace_prefix(&self) -> String {
614        let Some(remote_key) = &self.remote_key else {
615            return String::new();
616        };
617        let normalized = remote_key.trim_matches('/');
618        if normalized.is_empty() {
619            return String::new();
620        }
621        match normalized.rsplit_once('/') {
622            Some((parent, _)) if !parent.is_empty() => format!("{parent}/"),
623            _ => String::new(),
624        }
625    }
626
627    pub fn default_backup_head_key(&self) -> String {
628        if let Some(value) = self.metadata.get("red.config.backup.head_key") {
629            return value.clone();
630        }
631        reddb_file::backup_head_key(&self.remote_namespace_prefix())
632    }
633
634    pub fn default_snapshot_prefix(&self) -> String {
635        if let Some(value) = self.metadata.get("red.config.backup.snapshot_prefix") {
636            return value.clone();
637        }
638        reddb_file::backup_snapshot_prefix(&self.remote_namespace_prefix())
639    }
640
641    pub fn default_wal_archive_prefix(&self) -> String {
642        if let Some(value) = self.metadata.get("red.config.wal.archive.prefix") {
643            return value.clone();
644        }
645        reddb_file::backup_wal_prefix(&self.remote_namespace_prefix())
646    }
647
648    pub fn has_capability(&self, capability: Capability) -> bool {
649        self.feature_gates.has(capability)
650    }
651
652    /// Select the active storage-layout preset. Drives tier defaults for
653    /// the six tier-flag toggles. Per-feature setters still win.
654    pub fn with_layout(mut self, layout: crate::storage::layout::StorageLayout) -> Self {
655        self.layout = layout;
656        self.layout_explicit = true;
657        self
658    }
659
660    /// Override individual layout knobs (dedicated dirs + log routing) on
661    /// top of the active preset.
662    pub fn with_layout_overrides(
663        mut self,
664        overrides: crate::storage::layout::LayoutOverrides,
665    ) -> Self {
666        self.layout_overrides = overrides;
667        self
668    }
669
670    /// Declare the process memory budget explicitly (ADR 0073 §1, tier 1).
671    /// Wins over `REDDB_MEMORY_BUDGET`, the profile default, the cgroup limit
672    /// and physical-RAM detection. A zero budget is rejected when the runtime
673    /// opens, not silently swapped for a default.
674    pub fn with_memory_budget(mut self, bytes: u64) -> Self {
675        self.memory_budget_bytes = Some(bytes);
676        self
677    }
678
679    /// Select disk-backed residency explicitly, using the segment share of the
680    /// resolved memory budget. Opening an unsupported profile fails before I/O.
681    pub fn with_bounded_disk_residency(mut self, enabled: bool) -> Self {
682        self.bounded_disk_residency = enabled;
683        self
684    }
685
686    pub fn with_storage_profile(
687        mut self,
688        selection: crate::storage::profile::StorageProfileSelection,
689    ) -> Result<Self, String> {
690        self.storage_profile = selection.validate()?;
691        Ok(self)
692    }
693
694    /// Resolve `(data_path, TieredLayoutPaths)` for this options bundle.
695    /// Returns `None` when `data_path` is unset (in-memory ephemeral
696    /// instances don't materialise a support tree).
697    pub fn resolve_tiered_layout(
698        &self,
699    ) -> Option<(PathBuf, crate::storage::layout::TieredLayoutPaths)> {
700        let data_path = self.data_path.clone()?;
701        let paths = crate::storage::layout::TieredLayoutPaths::new(
702            &data_path,
703            self.layout,
704            self.layout_overrides.clone(),
705        );
706        Some((data_path, paths))
707    }
708
709    /// Flip the process-global tier-flag toggles to match this options
710    /// bundle's layout, and stash the resolved [`TieredLayoutPaths`] for
711    /// status surfaces. Idempotent. Per-feature env escape hatches
712    /// (`REDDB_META_JSON_SIDECAR=...` and friends) still override what
713    /// this method sets — they are read inside the per-toggle getter, not
714    /// here.
715    ///
716    /// Tier defaults:
717    ///
718    /// | toggle                     | minimal | standard | performance | max |
719    /// |----------------------------|:-------:|:--------:|:-----------:|:---:|
720    /// | `.meta.json` sidecar       |   off   |    off   |     off     |  on |
721    /// | seq-N catalog journal      |   off   |    off   |     off     |  on |
722    /// | `-shm` provisioning        |   off   | **on**   |   **on**    |  on |
723    /// | `fold_dwb_into_wal`        |   off   |    off   |     off     |  on |
724    /// | audit/slow log destination | stderr  |  stderr  |  file       | file|
725    ///
726    /// Retention for the seq-N journal: 32 at `Max`, 4 when the operator
727    /// opts in on a lower tier via `REDDB_SEQN_JOURNAL=1`.
728    pub fn apply_tier_defaults(&self) {
729        use crate::storage::layout::StorageLayout;
730
731        // Opt-in: only flip the process-global toggles when the operator
732        // explicitly chose a layout via `with_layout`. Tests and env
733        // hatches that pre-set toggles before opening a runtime must not
734        // be silently overridden by the implicit `Standard` default.
735        if !self.layout_explicit {
736            if let Some((_, paths)) = self.resolve_tiered_layout() {
737                tier_wiring::stash_layout_paths(paths);
738            }
739            return;
740        }
741
742        let layout = self.layout;
743        // .meta.json sidecar — Max only.
744        crate::physical::set_meta_json_sidecar_enabled(matches!(layout, StorageLayout::Max));
745
746        // Seq-N catalog journal — Max only. Retention = 32 for Max,
747        // OPT_IN baseline (4) for lower tiers (the value is consulted
748        // when an env hatch flips the toggle on outside Max).
749        crate::physical::set_seqn_journal_enabled(matches!(layout, StorageLayout::Max));
750        crate::physical::set_seqn_journal_retention(match layout {
751            StorageLayout::Max => crate::physical::DEFAULT_METADATA_JOURNAL_RETENTION,
752            _ => crate::physical::OPT_IN_METADATA_JOURNAL_RETENTION,
753        });
754
755        // `-shm` provisioning — anything ≥ Standard (gh-475 acceptance).
756        crate::physical::set_shm_provisioning_enabled(matches!(
757            layout,
758            StorageLayout::Standard | StorageLayout::Performance | StorageLayout::Max
759        ));
760
761        // Fold DWB into WAL — Max only (a single recovery substrate is the
762        // Max story). The pager manifest is folded into the `.rdb` on every
763        // tier since ADR 0038 §4 phase 1, so it has no toggle.
764        crate::physical::set_fold_dwb_into_wal_enabled(matches!(layout, StorageLayout::Max));
765
766        // Cache resolved layout paths for `red status` / diagnostics.
767        if let Some((_, paths)) = self.resolve_tiered_layout() {
768            tier_wiring::stash_layout_paths(paths);
769        }
770    }
771}
772
773/// Process-global cache of the most recently applied [`TieredLayoutPaths`].
774/// Read by `red status` to surface resolved log destinations. Written by
775/// `RedDBOptions::apply_tier_defaults` after each open.
776pub mod tier_wiring {
777    use std::sync::Mutex;
778
779    use crate::storage::layout::{LogDestination, TieredLayoutPaths};
780
781    static CURRENT_LAYOUT_PATHS: Mutex<Option<TieredLayoutPaths>> = Mutex::new(None);
782
783    pub fn stash_layout_paths(paths: TieredLayoutPaths) {
784        if let Ok(mut slot) = CURRENT_LAYOUT_PATHS.lock() {
785            *slot = Some(paths);
786        }
787    }
788
789    pub fn current_layout_paths() -> Option<TieredLayoutPaths> {
790        CURRENT_LAYOUT_PATHS
791            .lock()
792            .ok()
793            .and_then(|slot| slot.clone())
794    }
795
796    /// `(audit_log, slow_log)` destinations resolved from the active
797    /// layout. Falls back to `(Stderr, Stderr)` when no layout has been
798    /// applied yet (e.g. ephemeral in-memory paths).
799    pub fn current_log_destinations() -> (LogDestination, LogDestination) {
800        match current_layout_paths() {
801            Some(p) => (p.audit_log_destination, p.slow_log_destination),
802            None => (LogDestination::Stderr, LogDestination::Stderr),
803        }
804    }
805}
806
807#[derive(Debug, Clone, Default)]
808pub struct CollectionStats {
809    pub entities: usize,
810    pub cross_refs: usize,
811    pub segments: usize,
812}
813
814#[derive(Debug, Clone)]
815pub struct CatalogSnapshot {
816    pub name: String,
817    pub total_entities: usize,
818    pub total_collections: usize,
819    pub stats_by_collection: BTreeMap<String, CollectionStats>,
820    pub updated_at: SystemTime,
821}
822
823impl Default for CatalogSnapshot {
824    fn default() -> Self {
825        Self {
826            name: String::new(),
827            total_entities: 0,
828            total_collections: 0,
829            stats_by_collection: BTreeMap::new(),
830            updated_at: UNIX_EPOCH,
831        }
832    }
833}
834
835#[derive(Debug, Clone)]
836pub struct SchemaManifest {
837    pub format_version: u32,
838    pub created_at_unix_ms: u128,
839    pub updated_at_unix_ms: u128,
840    pub options: RedDBOptions,
841    pub collection_count: usize,
842}
843
844impl SchemaManifest {
845    pub fn now(options: RedDBOptions, collection_count: usize) -> Self {
846        let now = SystemTime::now()
847            .duration_since(UNIX_EPOCH)
848            .unwrap_or_default()
849            .as_millis();
850        Self {
851            format_version: REDDB_FORMAT_VERSION,
852            created_at_unix_ms: now,
853            updated_at_unix_ms: now,
854            options,
855            collection_count,
856        }
857    }
858}
859
860pub use reddb_engine::storage::error::StorageIntegrityError;
861
862#[derive(Debug)]
863pub enum RedDBError {
864    InvalidConfig(String),
865    SchemaVersionMismatch {
866        expected: u32,
867        found: u32,
868    },
869    FeatureNotEnabled(String),
870    NotFound(String),
871    ReadOnly(String),
872    /// The local commit is durable, but the requested replica durability was
873    /// not confirmed. Retrying the mutation may execute it a second time.
874    CommitUncertain {
875        lsn: u64,
876        detail: String,
877    },
878    InvalidOperation(String),
879    Engine(String),
880    Catalog(String),
881    StorageIntegrity(StorageIntegrityError),
882    Query(String),
883    Validation {
884        message: String,
885        validation: crate::json::Value,
886    },
887    Io(io::Error),
888    VersionUnavailable,
889    /// Operator-pinned cap exceeded (PLAN.md Phase 4.1). The string
890    /// payload should follow the `quota_exceeded:<limit_name>:<current>:<max>`
891    /// shape so HTTP / wire surfaces can map to the right status
892    /// (507 for storage, 429 for rate, 504 for duration, 413 for
893    /// payload).
894    QuotaExceeded(String),
895    /// Issue #769 (PRD #759 / S10) — an aggregating executor
896    /// (`aggregation`, `sort`, `window`) exceeded
897    /// `stream.executor.max_materialized_rows` in its in-memory state.
898    /// Carries the executor that fired, the configured ceiling and the
899    /// live row count at the breach so the client can decide whether to
900    /// redesign the query, raise the limit, or pre-aggregate. Surfaces
901    /// as HTTP 507 with a structured `materialization_limit_exceeded`
902    /// envelope; NDJSON streams emit the same code mid-stream.
903    MaterializationLimitExceeded {
904        executor: &'static str,
905        limit: usize,
906        current: usize,
907    },
908    Internal(String),
909}
910
911/// The text a wire client may see for `err`.
912///
913/// `Io` and `Internal` carry OS messages, filesystem paths and storage
914/// invariants meant for the operator's log, not for a peer; every other
915/// variant is a statement-level message the caller needs verbatim.
916pub fn client_facing_message(err: &RedDBError) -> String {
917    match err {
918        RedDBError::Io(inner) => format!("io error: {:?}", inner.kind()),
919        RedDBError::Internal(_) => "internal error; see the server log".to_string(),
920        other => other.to_string(),
921    }
922}
923
924impl fmt::Display for RedDBError {
925    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
926        match self {
927            Self::InvalidConfig(msg) => write!(f, "invalid config: {msg}"),
928            Self::SchemaVersionMismatch { expected, found } => {
929                write!(
930                    f,
931                    "schema version mismatch: expected {expected}, found {found}"
932                )
933            }
934            Self::FeatureNotEnabled(msg) => write!(f, "feature disabled: {msg}"),
935            Self::NotFound(msg) => write!(f, "not found: {msg}"),
936            Self::ReadOnly(msg) => write!(f, "read-only violation: {msg}"),
937            Self::CommitUncertain { lsn, detail } => write!(f,
938                "commit_uncertain: commit policy timed out at lsn {lsn}: {detail}; local commit is durable; replica durability is unconfirmed"),
939            Self::InvalidOperation(msg) => write!(f, "INVALID_OPERATION: {msg}"),
940            Self::Engine(msg) => write!(f, "engine error: {msg}"),
941            Self::Catalog(msg) => write!(f, "catalog error: {msg}"),
942            Self::StorageIntegrity(err) => write!(f, "{err}"),
943            Self::Query(msg) => write!(f, "query error: {msg}"),
944            Self::Validation { message, .. } => write!(f, "validation error: {message}"),
945            Self::Io(err) => write!(f, "io error: {err}"),
946            Self::VersionUnavailable => write!(f, "version information unavailable"),
947            Self::QuotaExceeded(msg) => write!(f, "quota exceeded: {msg}"),
948            Self::MaterializationLimitExceeded {
949                executor,
950                limit,
951                current,
952            } => write!(
953                f,
954                "materialization limit exceeded: executor={executor} current={current} limit={limit}"
955            ),
956            Self::Internal(msg) => write!(f, "internal error: {msg}"),
957        }
958    }
959}
960
961impl std::error::Error for RedDBError {}
962
963impl From<io::Error> for RedDBError {
964    fn from(err: io::Error) -> Self {
965        Self::Io(err)
966    }
967}
968
969impl From<crate::storage::engine::DatabaseError> for RedDBError {
970    fn from(err: crate::storage::engine::DatabaseError) -> Self {
971        Self::Engine(err.to_string())
972    }
973}
974
975impl From<crate::storage::wal::TxError> for RedDBError {
976    fn from(err: crate::storage::wal::TxError) -> Self {
977        Self::Engine(err.to_string())
978    }
979}
980
981impl From<reddb_engine::catalog::enforcement::ContractError> for RedDBError {
982    fn from(error: reddb_engine::catalog::enforcement::ContractError) -> Self {
983        match error {
984            reddb_engine::catalog::enforcement::ContractError::Query(message) => {
985                Self::Query(message)
986            }
987            reddb_engine::catalog::enforcement::ContractError::InvalidOperation(message) => {
988                Self::InvalidOperation(message)
989            }
990        }
991    }
992}
993
994impl From<reddb_engine::catalog::schema::ContractExpressionError> for RedDBError {
995    fn from(error: reddb_engine::catalog::schema::ContractExpressionError) -> Self {
996        Self::Query(error.message)
997    }
998}
999
1000impl From<crate::storage::StoreError> for RedDBError {
1001    fn from(err: crate::storage::StoreError) -> Self {
1002        match err {
1003            crate::storage::StoreError::StorageIntegrity(err) => Self::StorageIntegrity(err),
1004            other => Self::Catalog(other.to_string()),
1005        }
1006    }
1007}
1008
1009impl From<crate::storage::unified::devx::DevXError> for RedDBError {
1010    fn from(err: crate::storage::unified::devx::DevXError) -> Self {
1011        match err {
1012            crate::storage::unified::devx::DevXError::Validation(msg) => Self::InvalidConfig(msg),
1013            crate::storage::unified::devx::DevXError::Storage(msg) => Self::Engine(msg),
1014            crate::storage::unified::devx::DevXError::NotFound(msg) => Self::NotFound(msg),
1015        }
1016    }
1017}
1018
1019pub trait CatalogService {
1020    fn list_collections(&self) -> Vec<String>;
1021    fn collection_stats(&self, collection: &str) -> Option<CollectionStats>;
1022    fn catalog_snapshot(&self) -> CatalogSnapshot;
1023}
1024
1025pub trait QueryPlanner {
1026    fn plan_cost(&self, query: &str) -> Option<f64>;
1027}
1028
1029pub trait DataOps {
1030    fn execute_query(&self, query: &str) -> RedDBResult<()>;
1031}
1032
1033pub mod prelude {
1034    pub use super::{
1035        Capability, CapabilitySet, CatalogService, CatalogSnapshot, CollectionStats, DataOps,
1036        QueryPlanner, RedDBError, RedDBOptions, RedDBResult, SchemaManifest, StorageMode,
1037        REDDB_FORMAT_VERSION, REDDB_PROTOCOL_VERSION,
1038    };
1039}
1040
1041#[cfg(test)]
1042mod client_facing_message_tests {
1043    use super::{client_facing_message, RedDBError};
1044
1045    #[test]
1046    fn client_facing_message_hides_io_and_internal_detail() {
1047        let io = RedDBError::Io(std::io::Error::new(
1048            std::io::ErrorKind::NotFound,
1049            "/var/lib/reddb/data.rdb: No such file",
1050        ));
1051        assert_eq!(client_facing_message(&io), "io error: NotFound");
1052        let internal = RedDBError::Internal("pager invariant broke at /srv/db.wal".to_string());
1053        assert!(!client_facing_message(&internal).contains("/srv"));
1054        let query = RedDBError::Query("unknown column `x`".to_string());
1055        assert_eq!(client_facing_message(&query), query.to_string());
1056    }
1057}
reddb_server/api.rs

reddb_server/
api.rs
