Skip to main content

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    pub cache_pages: usize,
192    pub snapshot_retention: usize,
193    pub export_retention: usize,
194    pub feature_gates: CapabilitySet,
195    pub force_create: bool,
196    pub metadata: BTreeMap<String, String>,
197    /// Optional remote storage backend for snapshot transport.
198    pub remote_backend: Option<Arc<dyn crate::storage::backend::RemoteBackend>>,
199    /// Optional CAS-capable handle to the same backend, populated by
200    /// the factory when the configured backend implements
201    /// `AtomicRemoteBackend` (S3/local always; HTTP only when
202    /// `RED_HTTP_CONDITIONAL_WRITES=true`). `None` for backends that
203    /// do not provide compare-and-swap (Turso, D1, plain HTTP).
204    /// `LeaseStore` and any future CAS consumer pull from this field.
205    pub remote_backend_atomic: Option<Arc<dyn crate::storage::backend::AtomicRemoteBackend>>,
206    /// Remote object key used by the remote backend.
207    pub remote_key: Option<String>,
208    /// Replication configuration.
209    pub replication: ReplicationConfig,
210    /// Authentication & authorization configuration.
211    pub auth: AuthConfig,
212    /// Control Event Ledger configuration (issue #652). Read from
213    /// `REDDB_COMPLIANCE_MODE` at boot.
214    pub control_events: crate::runtime::control_events::ControlEventConfig,
215    /// Scoped data-plane query audit configuration. Disabled by
216    /// default; the regulated preset enables the stream without
217    /// adding catch-all rules.
218    pub query_audit: crate::runtime::query_audit::QueryAuditConfig,
219    /// Auto-create a HASH index on a user `id` column the first time a
220    /// row carrying that column is inserted into a collection. See
221    /// `UnifiedStoreConfig::auto_index_id`. Defaults to `true`; set to
222    /// `false` to opt out per workload (e.g. ingest pipelines that
223    /// don't need point-lookups by `id`).
224    pub auto_index_id: bool,
225    /// Tiered storage layout preset. Drives the defaults of the six
226    /// tier-flag toggles (`.meta.json` sidecar, seq-N catalog journal,
227    /// `-shm` provisioning, audit/slow log destinations, `fold_pager_meta`,
228    /// `fold_dwb_into_wal`). Explicit per-feature setters still win over
229    /// the tier default. See `crate::storage::layout::StorageLayout`.
230    pub layout: crate::storage::layout::StorageLayout,
231    /// Per-feature overrides applied on top of the resolved layout preset.
232    pub layout_overrides: crate::storage::layout::LayoutOverrides,
233    /// Operator-facing storage/deploy profile contract. This records the
234    /// selected posture before the physical directory layout is expanded.
235    pub storage_profile: crate::storage::profile::StorageProfileSelection,
236    /// True only when the caller explicitly selected a layout via
237    /// `with_layout`. When false, `apply_tier_defaults` short-circuits so
238    /// pre-existing process-global toggles (set by tests / env hatches
239    /// before opening a runtime) are not clobbered by the implicit
240    /// `Standard` default. The new tier-driven behavior is opt-in.
241    pub layout_explicit: bool,
242}
243
244impl fmt::Debug for RedDBOptions {
245    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
246        let backend_name = self.remote_backend.as_ref().map(|b| b.name().to_string());
247        f.debug_struct("RedDBOptions")
248            .field("mode", &self.mode)
249            .field("data_path", &self.data_path)
250            .field("read_only", &self.read_only)
251            .field("create_if_missing", &self.create_if_missing)
252            .field("verify_checksums", &self.verify_checksums)
253            .field("durability_mode", &self.durability_mode)
254            .field("group_commit", &self.group_commit)
255            .field("auto_checkpoint_pages", &self.auto_checkpoint_pages)
256            .field("cache_pages", &self.cache_pages)
257            .field("snapshot_retention", &self.snapshot_retention)
258            .field("export_retention", &self.export_retention)
259            .field("feature_gates", &self.feature_gates)
260            .field("force_create", &self.force_create)
261            .field("metadata", &self.metadata)
262            .field("remote_backend", &backend_name)
263            .field("remote_key", &self.remote_key)
264            .field("replication", &self.replication)
265            .field("auth", &self.auth)
266            .field("control_events", &self.control_events)
267            .field("query_audit", &self.query_audit)
268            .field("layout", &self.layout)
269            .field("layout_overrides", &self.layout_overrides)
270            .field("storage_profile", &self.storage_profile)
271            .finish()
272    }
273}
274
275impl Clone for RedDBOptions {
276    fn clone(&self) -> Self {
277        Self {
278            mode: self.mode,
279            data_path: self.data_path.clone(),
280            read_only: self.read_only,
281            create_if_missing: self.create_if_missing,
282            verify_checksums: self.verify_checksums,
283            durability_mode: self.durability_mode,
284            group_commit: self.group_commit,
285            auto_checkpoint_pages: self.auto_checkpoint_pages,
286            cache_pages: self.cache_pages,
287            snapshot_retention: self.snapshot_retention,
288            export_retention: self.export_retention,
289            feature_gates: self.feature_gates.clone(),
290            force_create: self.force_create,
291            metadata: self.metadata.clone(),
292            remote_backend: self.remote_backend.clone(),
293            remote_backend_atomic: self.remote_backend_atomic.clone(),
294            remote_key: self.remote_key.clone(),
295            replication: self.replication.clone(),
296            auth: self.auth.clone(),
297            control_events: self.control_events,
298            query_audit: self.query_audit.clone(),
299            auto_index_id: self.auto_index_id,
300            layout: self.layout,
301            layout_overrides: self.layout_overrides.clone(),
302            storage_profile: self.storage_profile,
303            layout_explicit: self.layout_explicit,
304        }
305    }
306}
307
308impl Default for RedDBOptions {
309    fn default() -> Self {
310        Self {
311            mode: StorageMode::Persistent,
312            data_path: None,
313            read_only: false,
314            create_if_missing: true,
315            verify_checksums: true,
316            // Perf-parity default — `WalDurableGrouped` matches
317            // PostgreSQL's `synchronous_commit=on` behaviour while
318            // amortising fsync cost across concurrent writers. The
319            // legacy `Strict` tier (per-commit fsync) stays available
320            // via `durability.mode = "strict"` / `REDDB_DURABILITY=strict`.
321            durability_mode: DurabilityMode::WalDurableGrouped,
322            group_commit: GroupCommitOptions::default(),
323            auto_checkpoint_pages: 1000,
324            cache_pages: 10_000,
325            snapshot_retention: DEFAULT_SNAPSHOT_RETENTION,
326            export_retention: DEFAULT_EXPORT_RETENTION,
327            feature_gates: CapabilitySet::new()
328                .with(Capability::Table)
329                .with(Capability::Graph)
330                .with(Capability::Vector),
331            force_create: true,
332            metadata: BTreeMap::new(),
333            remote_backend: None,
334            remote_backend_atomic: None,
335            remote_key: None,
336            replication: ReplicationConfig::standalone(),
337            auth: AuthConfig::default(),
338            control_events: crate::runtime::control_events::ControlEventConfig::default(),
339            query_audit: crate::runtime::query_audit::QueryAuditConfig::default(),
340            auto_index_id: true,
341            layout: crate::storage::layout::StorageLayout::default(),
342            layout_overrides: crate::storage::layout::LayoutOverrides::default(),
343            storage_profile: crate::storage::profile::StorageProfileSelection::embedded_single_file(
344            ),
345            layout_explicit: false,
346        }
347    }
348}
349
350impl RedDBOptions {
351    pub fn persistent<P: Into<PathBuf>>(path: P) -> Self {
352        Self {
353            mode: StorageMode::Persistent,
354            data_path: Some(path.into()),
355            ..Default::default()
356        }
357    }
358
359    /// Ephemeral, tempfile-backed database.
360    ///
361    /// The underlying storage is a real persistent file placed under the system
362    /// temp directory with a unique name — there is no longer a true in-memory
363    /// execution mode. Prefer [`RedDBOptions::persistent`] when the data should
364    /// outlive the process.
365    pub fn in_memory() -> Self {
366        static NEXT_EPHEMERAL_ID: std::sync::atomic::AtomicU64 =
367            std::sync::atomic::AtomicU64::new(0);
368
369        let now_nanos = std::time::SystemTime::now()
370            .duration_since(std::time::UNIX_EPOCH)
371            .map(|duration| duration.as_nanos())
372            .unwrap_or(0);
373        let unique = NEXT_EPHEMERAL_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
374        let path = std::env::temp_dir().join(format!(
375            "reddb-ephemeral-{}-{}-{}.rdb",
376            std::process::id(),
377            now_nanos,
378            unique
379        ));
380        let _ = std::fs::remove_file(&path);
381        let mut metadata = BTreeMap::new();
382        metadata.insert(
383            EPHEMERAL_RUNTIME_METADATA_KEY.to_string(),
384            "true".to_string(),
385        );
386        Self {
387            mode: StorageMode::Persistent,
388            data_path: Some(path),
389            auto_checkpoint_pages: 0,
390            cache_pages: 2_000,
391            snapshot_retention: DEFAULT_SNAPSHOT_RETENTION,
392            export_retention: DEFAULT_EXPORT_RETENTION,
393            read_only: false,
394            force_create: true,
395            metadata,
396            ..Default::default()
397        }
398    }
399
400    pub fn with_mode(mut self, mode: StorageMode) -> Self {
401        self.mode = mode;
402        self
403    }
404
405    pub fn with_data_path<P: Into<PathBuf>>(mut self, path: P) -> Self {
406        self.data_path = Some(path.into());
407        self
408    }
409
410    pub fn with_read_only(mut self, read_only: bool) -> Self {
411        self.read_only = read_only;
412        self
413    }
414
415    pub fn with_auto_checkpoint(mut self, pages: u32) -> Self {
416        self.auto_checkpoint_pages = pages;
417        self
418    }
419
420    pub fn with_durability_mode(mut self, mode: DurabilityMode) -> Self {
421        self.durability_mode = mode;
422        self
423    }
424
425    pub fn with_group_commit_window_ms(mut self, window_ms: u64) -> Self {
426        // `0` is a legitimate setting — "no wait, fsync on every wakeup".
427        // See `DEFAULT_GROUP_COMMIT_WINDOW_MS` docs for the tradeoff.
428        self.group_commit.window_ms = window_ms;
429        self
430    }
431
432    pub fn with_group_commit_max_statements(mut self, max_statements: usize) -> Self {
433        self.group_commit.max_statements = max_statements.max(1);
434        self
435    }
436
437    pub fn with_group_commit_max_wal_bytes(mut self, max_wal_bytes: u64) -> Self {
438        self.group_commit.max_wal_bytes = max_wal_bytes.max(1);
439        self
440    }
441
442    pub fn with_cache_pages(mut self, pages: usize) -> Self {
443        self.cache_pages = pages.max(2);
444        self
445    }
446
447    pub fn with_snapshot_retention(mut self, limit: usize) -> Self {
448        self.snapshot_retention = limit.max(1);
449        self
450    }
451
452    pub fn with_export_retention(mut self, limit: usize) -> Self {
453        self.export_retention = limit.max(1);
454        self
455    }
456
457    pub fn with_metadata<K: Into<String>, V: Into<String>>(mut self, key: K, value: V) -> Self {
458        self.metadata.insert(key.into(), value.into());
459        self
460    }
461
462    /// Toggle the implicit HASH index on user `id` columns at first
463    /// insert (#112). Defaults to enabled — pass `false` to fall back
464    /// to the legacy "scan unless `CREATE INDEX` is issued" behaviour.
465    pub fn with_auto_index_id(mut self, enabled: bool) -> Self {
466        self.auto_index_id = enabled;
467        self
468    }
469
470    pub fn with_capability(mut self, capability: Capability) -> Self {
471        self.feature_gates = self.feature_gates.with(capability);
472        self
473    }
474
475    /// Attach a remote storage backend for snapshot transport.
476    ///
477    /// On open, the database snapshot is downloaded from the remote `key`
478    /// to the local data path. On flush, the local file is uploaded back
479    /// to the remote backend under the same key.
480    pub fn with_remote_backend(
481        mut self,
482        backend: Arc<dyn crate::storage::backend::RemoteBackend>,
483        key: impl Into<String>,
484    ) -> Self {
485        self.remote_backend = Some(backend);
486        self.remote_key = Some(key.into());
487        self
488    }
489
490    /// Attach a CAS-capable backend handle. Pass the same `Arc` as
491    /// `with_remote_backend` (factories should construct the backend
492    /// once and call both setters); this method exists so the type
493    /// system, not runtime config, decides whether `LeaseStore` is
494    /// reachable.
495    pub fn with_atomic_remote_backend(
496        mut self,
497        backend: Arc<dyn crate::storage::backend::AtomicRemoteBackend>,
498    ) -> Self {
499        self.remote_backend_atomic = Some(backend);
500        self
501    }
502
503    pub fn with_replication(mut self, config: ReplicationConfig) -> Self {
504        self.replication = config;
505        self
506    }
507
508    pub fn with_auth(mut self, config: AuthConfig) -> Self {
509        self.auth = config;
510        self
511    }
512
513    pub fn resolved_path(&self, fallback: impl AsRef<Path>) -> PathBuf {
514        self.data_path
515            .clone()
516            .unwrap_or_else(|| fallback.as_ref().to_path_buf())
517    }
518
519    pub fn remote_namespace_prefix(&self) -> String {
520        let Some(remote_key) = &self.remote_key else {
521            return String::new();
522        };
523        let normalized = remote_key.trim_matches('/');
524        if normalized.is_empty() {
525            return String::new();
526        }
527        match normalized.rsplit_once('/') {
528            Some((parent, _)) if !parent.is_empty() => format!("{parent}/"),
529            _ => String::new(),
530        }
531    }
532
533    pub fn default_backup_head_key(&self) -> String {
534        if let Some(value) = self.metadata.get("red.config.backup.head_key") {
535            return value.clone();
536        }
537        reddb_file::backup_head_key(&self.remote_namespace_prefix())
538    }
539
540    pub fn default_snapshot_prefix(&self) -> String {
541        if let Some(value) = self.metadata.get("red.config.backup.snapshot_prefix") {
542            return value.clone();
543        }
544        reddb_file::backup_snapshot_prefix(&self.remote_namespace_prefix())
545    }
546
547    pub fn default_wal_archive_prefix(&self) -> String {
548        if let Some(value) = self.metadata.get("red.config.wal.archive.prefix") {
549            return value.clone();
550        }
551        reddb_file::backup_wal_prefix(&self.remote_namespace_prefix())
552    }
553
554    pub fn has_capability(&self, capability: Capability) -> bool {
555        self.feature_gates.has(capability)
556    }
557
558    /// Select the active storage-layout preset. Drives tier defaults for
559    /// the six tier-flag toggles. Per-feature setters still win.
560    pub fn with_layout(mut self, layout: crate::storage::layout::StorageLayout) -> Self {
561        self.layout = layout;
562        self.layout_explicit = true;
563        self
564    }
565
566    /// Override individual layout knobs (dedicated dirs + log routing) on
567    /// top of the active preset.
568    pub fn with_layout_overrides(
569        mut self,
570        overrides: crate::storage::layout::LayoutOverrides,
571    ) -> Self {
572        self.layout_overrides = overrides;
573        self
574    }
575
576    pub fn with_storage_profile(
577        mut self,
578        selection: crate::storage::profile::StorageProfileSelection,
579    ) -> Result<Self, String> {
580        self.storage_profile = selection.validate()?;
581        Ok(self)
582    }
583
584    /// Resolve `(data_path, TieredLayoutPaths)` for this options bundle.
585    /// Returns `None` when `data_path` is unset (in-memory ephemeral
586    /// instances don't materialise a support tree).
587    pub fn resolve_tiered_layout(
588        &self,
589    ) -> Option<(PathBuf, crate::storage::layout::TieredLayoutPaths)> {
590        let data_path = self.data_path.clone()?;
591        let paths = crate::storage::layout::TieredLayoutPaths::new(
592            &data_path,
593            self.layout,
594            self.layout_overrides.clone(),
595        );
596        Some((data_path, paths))
597    }
598
599    /// Flip the process-global tier-flag toggles to match this options
600    /// bundle's layout, and stash the resolved [`TieredLayoutPaths`] for
601    /// status surfaces. Idempotent. Per-feature env escape hatches
602    /// (`REDDB_META_JSON_SIDECAR=...` and friends) still override what
603    /// this method sets — they are read inside the per-toggle getter, not
604    /// here.
605    ///
606    /// Tier defaults:
607    ///
608    /// | toggle                     | minimal | standard | performance | max |
609    /// |----------------------------|:-------:|:--------:|:-----------:|:---:|
610    /// | `.meta.json` sidecar       |   off   |    off   |     off     |  on |
611    /// | seq-N catalog journal      |   off   |    off   |     off     |  on |
612    /// | `-shm` provisioning        |   off   | **on**   |   **on**    |  on |
613    /// | `fold_pager_meta`          |   off   |    off   |     off     |  on |
614    /// | `fold_dwb_into_wal`        |   off   |    off   |     off     |  on |
615    /// | audit/slow log destination | stderr  |  stderr  |  file       | file|
616    ///
617    /// Retention for the seq-N journal: 32 at `Max`, 4 when the operator
618    /// opts in on a lower tier via `REDDB_SEQN_JOURNAL=1`.
619    pub fn apply_tier_defaults(&self) {
620        use crate::storage::layout::StorageLayout;
621
622        // Opt-in: only flip the process-global toggles when the operator
623        // explicitly chose a layout via `with_layout`. Tests and env
624        // hatches that pre-set toggles before opening a runtime must not
625        // be silently overridden by the implicit `Standard` default.
626        if !self.layout_explicit {
627            if let Some((_, paths)) = self.resolve_tiered_layout() {
628                tier_wiring::stash_layout_paths(paths);
629            }
630            return;
631        }
632
633        let layout = self.layout;
634        // .meta.json sidecar — Max only.
635        crate::physical::set_meta_json_sidecar_enabled(matches!(layout, StorageLayout::Max));
636
637        // Seq-N catalog journal — Max only. Retention = 32 for Max,
638        // OPT_IN baseline (4) for lower tiers (the value is consulted
639        // when an env hatch flips the toggle on outside Max).
640        crate::physical::set_seqn_journal_enabled(matches!(layout, StorageLayout::Max));
641        crate::physical::set_seqn_journal_retention(match layout {
642            StorageLayout::Max => crate::physical::DEFAULT_METADATA_JOURNAL_RETENTION,
643            _ => crate::physical::OPT_IN_METADATA_JOURNAL_RETENTION,
644        });
645
646        // `-shm` provisioning — anything ≥ Standard (gh-475 acceptance).
647        crate::physical::set_shm_provisioning_enabled(matches!(
648            layout,
649            StorageLayout::Standard | StorageLayout::Performance | StorageLayout::Max
650        ));
651
652        // Fold pager meta + fold DWB into WAL — Max only (single
653        // datafile + single recovery substrate is the Max story).
654        crate::physical::set_fold_pager_meta_enabled(matches!(layout, StorageLayout::Max));
655        crate::physical::set_fold_dwb_into_wal_enabled(matches!(layout, StorageLayout::Max));
656
657        // Cache resolved layout paths for `red status` / diagnostics.
658        if let Some((_, paths)) = self.resolve_tiered_layout() {
659            tier_wiring::stash_layout_paths(paths);
660        }
661    }
662}
663
664/// Process-global cache of the most recently applied [`TieredLayoutPaths`].
665/// Read by `red status` to surface resolved log destinations. Written by
666/// `RedDBOptions::apply_tier_defaults` after each open.
667pub mod tier_wiring {
668    use std::sync::Mutex;
669
670    use crate::storage::layout::{LogDestination, TieredLayoutPaths};
671
672    static CURRENT_LAYOUT_PATHS: Mutex<Option<TieredLayoutPaths>> = Mutex::new(None);
673
674    pub fn stash_layout_paths(paths: TieredLayoutPaths) {
675        if let Ok(mut slot) = CURRENT_LAYOUT_PATHS.lock() {
676            *slot = Some(paths);
677        }
678    }
679
680    pub fn current_layout_paths() -> Option<TieredLayoutPaths> {
681        CURRENT_LAYOUT_PATHS
682            .lock()
683            .ok()
684            .and_then(|slot| slot.clone())
685    }
686
687    /// `(audit_log, slow_log)` destinations resolved from the active
688    /// layout. Falls back to `(Stderr, Stderr)` when no layout has been
689    /// applied yet (e.g. ephemeral in-memory paths).
690    pub fn current_log_destinations() -> (LogDestination, LogDestination) {
691        match current_layout_paths() {
692            Some(p) => (p.audit_log_destination, p.slow_log_destination),
693            None => (LogDestination::Stderr, LogDestination::Stderr),
694        }
695    }
696}
697
698#[derive(Debug, Clone, Default)]
699pub struct CollectionStats {
700    pub entities: usize,
701    pub cross_refs: usize,
702    pub segments: usize,
703}
704
705#[derive(Debug, Clone)]
706pub struct CatalogSnapshot {
707    pub name: String,
708    pub total_entities: usize,
709    pub total_collections: usize,
710    pub stats_by_collection: BTreeMap<String, CollectionStats>,
711    pub updated_at: SystemTime,
712}
713
714impl Default for CatalogSnapshot {
715    fn default() -> Self {
716        Self {
717            name: String::new(),
718            total_entities: 0,
719            total_collections: 0,
720            stats_by_collection: BTreeMap::new(),
721            updated_at: UNIX_EPOCH,
722        }
723    }
724}
725
726#[derive(Debug, Clone)]
727pub struct SchemaManifest {
728    pub format_version: u32,
729    pub created_at_unix_ms: u128,
730    pub updated_at_unix_ms: u128,
731    pub options: RedDBOptions,
732    pub collection_count: usize,
733}
734
735impl SchemaManifest {
736    pub fn now(options: RedDBOptions, collection_count: usize) -> Self {
737        let now = SystemTime::now()
738            .duration_since(UNIX_EPOCH)
739            .unwrap_or_default()
740            .as_millis();
741        Self {
742            format_version: REDDB_FORMAT_VERSION,
743            created_at_unix_ms: now,
744            updated_at_unix_ms: now,
745            options,
746            collection_count,
747        }
748    }
749}
750
751#[derive(Debug)]
752pub enum RedDBError {
753    InvalidConfig(String),
754    SchemaVersionMismatch {
755        expected: u32,
756        found: u32,
757    },
758    FeatureNotEnabled(String),
759    NotFound(String),
760    ReadOnly(String),
761    InvalidOperation(String),
762    Engine(String),
763    Catalog(String),
764    Query(String),
765    Validation {
766        message: String,
767        validation: crate::json::Value,
768    },
769    Io(io::Error),
770    VersionUnavailable,
771    /// Operator-pinned cap exceeded (PLAN.md Phase 4.1). The string
772    /// payload should follow the `quota_exceeded:<limit_name>:<current>:<max>`
773    /// shape so HTTP / wire surfaces can map to the right status
774    /// (507 for storage, 429 for rate, 504 for duration, 413 for
775    /// payload).
776    QuotaExceeded(String),
777    /// Issue #769 (PRD #759 / S10) — an aggregating executor
778    /// (`aggregation`, `sort`, `window`) exceeded
779    /// `stream.executor.max_materialized_rows` in its in-memory state.
780    /// Carries the executor that fired, the configured ceiling and the
781    /// live row count at the breach so the client can decide whether to
782    /// redesign the query, raise the limit, or pre-aggregate. Surfaces
783    /// as HTTP 507 with a structured `materialization_limit_exceeded`
784    /// envelope; NDJSON streams emit the same code mid-stream.
785    MaterializationLimitExceeded {
786        executor: &'static str,
787        limit: usize,
788        current: usize,
789    },
790    Internal(String),
791}
792
793impl fmt::Display for RedDBError {
794    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
795        match self {
796            Self::InvalidConfig(msg) => write!(f, "invalid config: {msg}"),
797            Self::SchemaVersionMismatch { expected, found } => {
798                write!(
799                    f,
800                    "schema version mismatch: expected {expected}, found {found}"
801                )
802            }
803            Self::FeatureNotEnabled(msg) => write!(f, "feature disabled: {msg}"),
804            Self::NotFound(msg) => write!(f, "not found: {msg}"),
805            Self::ReadOnly(msg) => write!(f, "read-only violation: {msg}"),
806            Self::InvalidOperation(msg) => write!(f, "INVALID_OPERATION: {msg}"),
807            Self::Engine(msg) => write!(f, "engine error: {msg}"),
808            Self::Catalog(msg) => write!(f, "catalog error: {msg}"),
809            Self::Query(msg) => write!(f, "query error: {msg}"),
810            Self::Validation { message, .. } => write!(f, "validation error: {message}"),
811            Self::Io(err) => write!(f, "io error: {err}"),
812            Self::VersionUnavailable => write!(f, "version information unavailable"),
813            Self::QuotaExceeded(msg) => write!(f, "quota exceeded: {msg}"),
814            Self::MaterializationLimitExceeded {
815                executor,
816                limit,
817                current,
818            } => write!(
819                f,
820                "materialization limit exceeded: executor={executor} current={current} limit={limit}"
821            ),
822            Self::Internal(msg) => write!(f, "internal error: {msg}"),
823        }
824    }
825}
826
827impl std::error::Error for RedDBError {}
828
829impl From<io::Error> for RedDBError {
830    fn from(err: io::Error) -> Self {
831        Self::Io(err)
832    }
833}
834
835impl From<crate::storage::engine::DatabaseError> for RedDBError {
836    fn from(err: crate::storage::engine::DatabaseError) -> Self {
837        Self::Engine(err.to_string())
838    }
839}
840
841impl From<crate::storage::wal::TxError> for RedDBError {
842    fn from(err: crate::storage::wal::TxError) -> Self {
843        Self::Engine(err.to_string())
844    }
845}
846
847impl From<crate::storage::StoreError> for RedDBError {
848    fn from(err: crate::storage::StoreError) -> Self {
849        Self::Catalog(err.to_string())
850    }
851}
852
853impl From<crate::storage::unified::devx::DevXError> for RedDBError {
854    fn from(err: crate::storage::unified::devx::DevXError) -> Self {
855        match err {
856            crate::storage::unified::devx::DevXError::Validation(msg) => Self::InvalidConfig(msg),
857            crate::storage::unified::devx::DevXError::Storage(msg) => Self::Engine(msg),
858            crate::storage::unified::devx::DevXError::NotFound(msg) => Self::NotFound(msg),
859        }
860    }
861}
862
863pub trait CatalogService {
864    fn list_collections(&self) -> Vec<String>;
865    fn collection_stats(&self, collection: &str) -> Option<CollectionStats>;
866    fn catalog_snapshot(&self) -> CatalogSnapshot;
867}
868
869pub trait QueryPlanner {
870    fn plan_cost(&self, query: &str) -> Option<f64>;
871}
872
873pub trait DataOps {
874    fn execute_query(&self, query: &str) -> RedDBResult<()>;
875}
876
877pub mod prelude {
878    pub use super::{
879        Capability, CapabilitySet, CatalogService, CatalogSnapshot, CollectionStats, DataOps,
880        QueryPlanner, RedDBError, RedDBOptions, RedDBResult, SchemaManifest, StorageMode,
881        REDDB_FORMAT_VERSION, REDDB_PROTOCOL_VERSION,
882    };
883}