Skip to main content

reddb_server/runtime/
impl_lifecycle.rs

1//! Runtime bootstrap / constructor / handle accessors.
2//!
3//! Extracted verbatim from `impl_core.rs` (impl_core slice 8/10, issue #1629).
4//! Houses the in-memory / options constructors, the ~800-line `with_pool`
5//! constructor (moved verbatim — restructuring it is future work), snapshot
6//! rehydration and materialized-view backing lifecycle, system keyed-collection
7//! bootstrap, and the db / index-store / schema-vocabulary / auth-store /
8//! oauth / browser-token / registry handle accessors.
9//!
10//! - **Free helpers** — `view_records_to_entities`,
11//!   `system_keyed_collection_contract`, `table_row_index_fields`.
12use super::impl_config_secret::seed_storage_deploy_config;
13use super::*;
14
15/// Convert the rows produced by a materialized-view body into
16/// `UnifiedEntity` table rows targeting the backing collection.
17/// Issue #595 slice 9c — feeds `UnifiedStore::refresh_collection`.
18///
19/// Graph fragments and vector hits are ignored: a materialized view
20/// is a relational result set (SELECT-shaped); slices 11+ may extend
21/// this once we have a richer view body shape. Each row materialises
22/// the union of its schema-bound columns + overflow.
23pub(crate) fn view_records_to_entities(
24    table: &str,
25    records: &[crate::storage::query::unified::UnifiedRecord],
26) -> Vec<crate::storage::UnifiedEntity> {
27    use std::collections::HashMap;
28    let table_arc: std::sync::Arc<str> = std::sync::Arc::from(table);
29    let mut out = Vec::with_capacity(records.len());
30    for record in records {
31        let mut named: HashMap<String, reddb_types::Value> = HashMap::new();
32        for (name, value) in record.iter_fields() {
33            named.insert(name.to_string(), value.clone());
34        }
35        let entity = crate::storage::UnifiedEntity::new(
36            crate::storage::EntityId::new(0),
37            crate::storage::EntityKind::TableRow {
38                table: std::sync::Arc::clone(&table_arc),
39                row_id: 0,
40            },
41            crate::storage::EntityData::Row(crate::storage::RowData {
42                columns: Vec::new(),
43                named: Some(named),
44                schema: None,
45            }),
46        );
47        out.push(entity);
48    }
49    out
50}
51
52fn system_keyed_collection_contract(
53    name: &str,
54    model: crate::catalog::CollectionModel,
55) -> crate::physical::CollectionContract {
56    let now = crate::utils::now_unix_millis() as u128;
57    crate::physical::CollectionContract {
58        name: name.to_string(),
59        declared_model: model,
60        schema_mode: crate::catalog::SchemaMode::Dynamic,
61        origin: crate::physical::ContractOrigin::Implicit,
62        version: 1,
63        created_at_unix_ms: now,
64        updated_at_unix_ms: now,
65        default_ttl_ms: None,
66        vector_dimension: None,
67        vector_metric: None,
68        context_index_fields: Vec::new(),
69        declared_columns: Vec::new(),
70        table_def: None,
71        timestamps_enabled: false,
72        context_index_enabled: false,
73        metrics_raw_retention_ms: None,
74        metrics_rollup_policies: Vec::new(),
75        metrics_tenant_identity: None,
76        metrics_namespace: None,
77        append_only: false,
78        subscriptions: Vec::new(),
79        analytics_config: Vec::new(),
80        session_key: None,
81        session_gap_ms: None,
82        retention_duration_ms: None,
83        analytical_storage: None,
84
85        ai_policy: None,
86    }
87}
88
89pub(crate) fn table_row_index_fields(
90    entity: &crate::storage::unified::entity::UnifiedEntity,
91) -> Vec<(String, reddb_types::Value)> {
92    let crate::storage::EntityData::Row(row) = &entity.data else {
93        return Vec::new();
94    };
95    if let Some(named) = &row.named {
96        return named
97            .iter()
98            .map(|(name, value)| (name.clone(), value.clone()))
99            .collect();
100    }
101    if let Some(schema) = &row.schema {
102        return schema
103            .iter()
104            .zip(row.columns.iter())
105            .map(|(name, value)| (name.clone(), value.clone()))
106            .collect();
107    }
108    Vec::new()
109}
110
111impl RedDBRuntime {
112    pub fn in_memory() -> RedDBResult<Self> {
113        Self::with_options(RedDBOptions::in_memory())
114    }
115
116    pub fn flush(&self) -> RedDBResult<()> {
117        self.inner
118            .db
119            .flush()
120            .map_err(|err| RedDBError::Internal(err.to_string()))
121    }
122
123    /// Handle to the intent-lock manager for tests + introspection.
124    /// Production code acquires via `LockerGuard::new(rt.lock_manager())`
125    /// rather than touching the manager directly.
126    pub fn lock_manager(&self) -> std::sync::Arc<crate::runtime::lock_manager::LockManager> {
127        self.inner.lock_manager.clone()
128    }
129
130    /// Process-local governance registry for managed policy/config guardrails.
131    pub fn config_registry(&self) -> std::sync::Arc<crate::auth::registry::ConfigRegistry> {
132        self.inner.config_registry.clone()
133    }
134
135    pub fn query_audit(&self) -> std::sync::Arc<crate::runtime::query_audit::QueryAuditStream> {
136        self.inner.query_audit.clone()
137    }
138
139    pub fn control_events_require_persistence(&self) -> bool {
140        self.inner.control_event_config.require_persistence()
141    }
142
143    pub fn control_event_config(&self) -> crate::runtime::control_events::ControlEventConfig {
144        self.inner.control_event_config
145    }
146
147    pub fn control_event_ledger(
148        &self,
149    ) -> Arc<dyn crate::runtime::control_events::ControlEventLedger> {
150        self.inner.control_event_ledger.read().clone()
151    }
152
153    #[doc(hidden)]
154    pub fn replace_control_event_ledger_for_tests(
155        &self,
156        ledger: Arc<dyn crate::runtime::control_events::ControlEventLedger>,
157    ) {
158        *self.inner.control_event_ledger.write() = ledger;
159    }
160
161    #[inline(never)]
162    pub fn with_options(options: RedDBOptions) -> RedDBResult<Self> {
163        Self::with_pool(options, ConnectionPoolConfig::default())
164    }
165
166    /// The memory budget resolved at boot (ADR 0073 §1). Immutable for the
167    /// process lifetime; echoed by the boot log and the `red.stats` budget
168    /// section.
169    pub fn memory_budget(&self) -> crate::storage::memory_budget::MemoryBudget {
170        self.inner.memory_budget
171    }
172
173    /// The one shared accounting pool (ADR 0073 §2). Per-pool shares are fixed
174    /// at boot; live usage is refreshed by
175    /// [`RedDBRuntime::refresh_memory_accounting`].
176    pub fn memory_accounting(&self) -> &crate::storage::memory_pools::MemoryAccounting {
177        &self.inner.memory_accounting
178    }
179
180    pub fn with_pool(
181        options: RedDBOptions,
182        pool_config: ConnectionPoolConfig,
183    ) -> RedDBResult<Self> {
184        options
185            .replication
186            .validate_failover_profile()
187            .map_err(RedDBError::InvalidConfig)?;
188        // PLAN.md Phase 9.1 — capture wall-clock before storage
189        // open so the cold-start phase markers can be backfilled
190        // once Lifecycle is constructed below. Storage open
191        // encapsulates auto-restore + WAL replay; we treat the
192        // whole window as one combined "restore" + "wal_replay"
193        // phase split at the same boundary because the storage
194        // layer doesn't yet emit a finer signal.
195        let boot_open_start_ms = std::time::SystemTime::now()
196            .duration_since(std::time::UNIX_EPOCH)
197            .map(|d| d.as_millis() as u64)
198            .unwrap_or(0);
199        // ADR 0073 §1 — resolve the one memory budget before anything is
200        // opened, so a nonsensical operator value fails the boot before the
201        // process touches disk. The log line fires once per process.
202        let memory_budget = crate::storage::memory_budget::resolve_for_boot(
203            options.storage_profile.deploy_profile,
204            options.memory_budget_bytes,
205        )
206        .map_err(|err| RedDBError::InvalidConfig(err.to_string()))?;
207        crate::storage::memory_budget::log_resolved_once(&memory_budget);
208        // ADR 0073 §2 — divide the budget among the pools before any of them
209        // is constructed. `BudgetShares::resolve` asserts Σ(shares) ≤ budget;
210        // one boot log line per pool names its share.
211        let memory_shares = crate::storage::memory_pools::BudgetShares::resolve(
212            memory_budget,
213            options.storage_profile.deploy_profile,
214        );
215        memory_shares.log_once();
216        let embedded_single_file = options.storage_profile.deploy_profile
217            == crate::storage::DeployProfile::Embedded
218            && options.storage_profile.packaging == crate::storage::StoragePackaging::SingleFile;
219        let db = Arc::new(RedDB::open_with_options(&options).map_err(|err| {
220            match err.downcast::<crate::storage::StoreError>() {
221                Ok(store_err) => RedDBError::from(*store_err),
222                Err(err) => RedDBError::Internal(err.to_string()),
223            }
224        })?);
225        // The RAM tier is sized from its share. L2's disk extent is not memory
226        // and keeps its own ceiling; only L1 and L2's resident metadata are
227        // accounted against `blob_cache_l1`.
228        let result_blob_cache_config = if embedded_single_file {
229            crate::storage::cache::BlobCacheConfig::default()
230        } else {
231            crate::storage::cache::BlobCacheConfig::default().with_l2_path(
232                reddb_file::layout::result_cache_l2_path(
233                    &options.resolved_path(reddb_file::default_database_path()),
234                ),
235            )
236        }
237        .with_l1_bytes_max(memory_shares.blob_cache_l1_bytes());
238        let result_blob_cache =
239            crate::storage::cache::BlobCache::open_with_l2(result_blob_cache_config).map_err(
240                |err| RedDBError::Internal(format!("open result Blob Cache L2 failed: {err:?}")),
241            )?;
242        let storage_ready_ms = std::time::SystemTime::now()
243            .duration_since(std::time::UNIX_EPOCH)
244            .map(|d| d.as_millis() as u64)
245            .unwrap_or(0);
246
247        let no_auth_boot = crate::service_cli::no_auth_active(&options);
248        let runtime = Self {
249            background_tasks: Some(Arc::new(background_tasks::RuntimeBackgroundTasks::default())),
250            inner: Arc::new(RuntimeInner {
251                functions: parking_lot::RwLock::new(
252                    super::function_catalog::FunctionCatalog::load(&db)?,
253                ),
254                db: db.clone(),
255                layout: PhysicalLayout::from_options(&options),
256                embedded_single_file,
257                memory_budget,
258                memory_accounting: Arc::new(
259                    crate::storage::memory_pools::MemoryAccounting::from_shares(
260                        memory_budget,
261                        memory_shares,
262                    ),
263                ),
264                memory_reservations: parking_lot::Mutex::new(
265                    super::memory_admission::MemoryReservations::default(),
266                ),
267                indices: IndexCatalog::register_default_vector_graph(
268                    options.has_capability(crate::api::Capability::Table),
269                    options.has_capability(crate::api::Capability::Graph),
270                ),
271                pool_config,
272                pool: Mutex::new(PoolState::default()),
273                started_at_unix_ms: SystemTime::now()
274                    .duration_since(UNIX_EPOCH)
275                    .unwrap_or_default()
276                    .as_millis(),
277                probabilistic: super::probabilistic_store::ProbabilisticStore::new(),
278                index_store: super::index_store::IndexStore::new(),
279                cdc: crate::replication::cdc::CdcBuffer::new(100_000),
280                checkpoint_projection_stats: super::CheckpointProjectionStats::default(),
281                scrub_state: parking_lot::Mutex::new(super::ScrubRuntimeState::default()),
282                checkpoint_columnar_emission_budget_chunks: options
283                    .checkpoint_columnar_emission_budget_chunks,
284                columnar_projection_size_floor_rows: options.columnar_projection_size_floor_rows,
285                backup_scheduler: crate::replication::scheduler::BackupScheduler::new(3600),
286                query_cache: parking_lot::RwLock::new(
287                    crate::storage::query::planner::cache::PlanCache::new(1000),
288                ),
289                result_cache: parking_lot::RwLock::new((
290                    HashMap::new(),
291                    std::collections::VecDeque::new(),
292                )),
293                result_blob_cache,
294                result_blob_entries: parking_lot::RwLock::new((
295                    HashMap::new(),
296                    std::collections::VecDeque::new(),
297                )),
298                ask_answer_cache_entries: parking_lot::RwLock::new((
299                    HashSet::new(),
300                    std::collections::VecDeque::new(),
301                )),
302                result_cache_shadow_divergences: std::sync::atomic::AtomicU64::new(0),
303                result_cache_hits: std::sync::atomic::AtomicU64::new(0),
304                result_cache_misses: std::sync::atomic::AtomicU64::new(0),
305                result_cache_evictions: std::sync::atomic::AtomicU64::new(0),
306                ask_daily_spend: parking_lot::RwLock::new(HashMap::new()),
307                queue_message_locks: parking_lot::RwLock::new(HashMap::new()),
308                rmw_locks: RmwLockTable::new(),
309                planner_dirty_tables: parking_lot::RwLock::new(HashSet::new()),
310                ec_registry: Arc::new(crate::ec::config::EcRegistry::new()),
311                config_registry: Arc::new(crate::auth::registry::ConfigRegistry::new()),
312                ec_worker: crate::ec::worker::EcWorker::new(),
313                auth_store: parking_lot::RwLock::new(None),
314                oauth_validator: parking_lot::RwLock::new(None),
315                browser_token_authority: parking_lot::RwLock::new(None),
316                views: parking_lot::RwLock::new(HashMap::new()),
317                materialized_views: parking_lot::RwLock::new(
318                    crate::storage::cache::result::MaterializedViewCache::new(),
319                ),
320                retention_sweeper: parking_lot::RwLock::new(
321                    crate::runtime::retention_sweeper::RetentionSweeperState::new(),
322                ),
323                transaction_state: crate::runtime::transaction_state::TransactionState::new(),
324                env_config_overrides: crate::runtime::config_overlay::collect_env_overrides(),
325                lock_manager: Arc::new({
326                    // Sourced from the matrix: Tier B key
327                    // `concurrency.locking.deadlock_timeout_ms`
328                    // (default 5000). Env var wins at boot so
329                    // operators can tune without touching red_config.
330                    let env = crate::runtime::config_overlay::collect_env_overrides();
331                    let timeout_ms = env
332                        .get("concurrency.locking.deadlock_timeout_ms")
333                        .and_then(|raw| raw.parse::<u64>().ok())
334                        .unwrap_or_else(|| {
335                            match crate::runtime::config_matrix::default_for(
336                                "concurrency.locking.deadlock_timeout_ms",
337                            ) {
338                                Some(crate::serde_json::Value::Number(n)) => n as u64,
339                                _ => 5000,
340                            }
341                        });
342                    let cfg = crate::runtime::lock_manager::LockConfig {
343                        default_timeout: std::time::Duration::from_millis(timeout_ms),
344                        ..Default::default()
345                    };
346                    crate::runtime::lock_manager::LockManager::new(cfg)
347                }),
348                rls_policies: parking_lot::RwLock::new(HashMap::new()),
349                rls_enabled_tables: parking_lot::RwLock::new(HashSet::new()),
350                foreign_tables: Arc::new(crate::storage::fdw::ForeignTableRegistry::with_builtins()),
351                pending_tombstones: parking_lot::RwLock::new(HashMap::new()),
352                pending_versioned_updates: parking_lot::RwLock::new(HashMap::new()),
353                pending_queue_dedup: parking_lot::RwLock::new(HashMap::new()),
354                pending_kv_watch_events: parking_lot::RwLock::new(HashMap::new()),
355                pending_vault_writes: parking_lot::RwLock::new(HashMap::new()),
356                pending_store_wal_actions: parking_lot::RwLock::new(HashMap::new()),
357                pending_claim_locks: parking_lot::RwLock::new(HashMap::new()),
358                queue_wait_registry: std::sync::Arc::new(
359                    crate::runtime::queue_wait_registry::QueueWaitRegistry::new(),
360                ),
361                pending_queue_wakes: parking_lot::RwLock::new(HashMap::new()),
362                tenant_tables: parking_lot::RwLock::new(HashMap::new()),
363                ddl_epoch: std::sync::atomic::AtomicU64::new(0),
364                write_gate: Arc::new(crate::runtime::write_gate::WriteGate::from_options(
365                    &options,
366                )),
367                lifecycle: crate::runtime::lifecycle::Lifecycle::new(),
368                resource_limits: crate::runtime::resource_limits::ResourceLimits::from_env(),
369                audit_log: {
370                    // Default audit-log path for the in-memory case
371                    // sits in the system temp dir; persistent runs
372                    // place it next to the resolved data file.
373                    //
374                    // gh-471 iter 2: route through the resolved
375                    // `LogDestination`. Performance/Max tiers emit a
376                    // file-backed log destination under the file-owned
377                    // support-directory logs tier;
378                    // lower tiers / ephemeral runs report `Stderr`
379                    // and we keep the legacy file-next-to-data sink.
380                    // #1375 — single-file embedded mode keeps the data
381                    // directory to exactly the `.rdb` artifact, so the audit
382                    // log must NOT land as a sibling. Route it to a
383                    // process-unique temp location even when a data path is
384                    // set; only the non-embedded case uses the data dir.
385                    let data_path = if embedded_single_file {
386                        std::env::temp_dir()
387                            .join("reddb-embedded-runtime")
388                            .join(format!("audit-{}", std::process::id()))
389                    } else {
390                        options
391                            .data_path
392                            .clone()
393                            .unwrap_or_else(|| std::env::temp_dir().join("reddb"))
394                    };
395                    let (audit_dest, _) = crate::api::tier_wiring::current_log_destinations();
396                    if !matches!(audit_dest, crate::storage::layout::LogDestination::File(_))
397                        && (embedded_single_file
398                            || options
399                                .metadata
400                                .contains_key(crate::api::EPHEMERAL_RUNTIME_METADATA_KEY))
401                    {
402                        // The Stderr/Syslog lower-tier sink resolves to a
403                        // `for_data_path` sibling that collides across concurrent
404                        // temp-dir runtimes — nextest's process-per-test model
405                        // truncates one shared file, flaking audit assertions.
406                        // Pin a unique sibling for these short-lived ephemeral /
407                        // single-file embedded runtimes. The file-owned support-
408                        // dir tier (`File`) is already per-data unique, so leave
409                        // it to `for_destination` (#1375: the embedded audit then
410                        // still never lands a sibling next to the `.rdb`).
411                        let audit_path = reddb_file::layout::sibling_path(
412                            &data_path,
413                            &reddb_file::layout::sidecar_file_name(&data_path, "audit.log"),
414                        );
415                        Arc::new(crate::runtime::audit_log::AuditLogger::with_path(
416                            audit_path,
417                        ))
418                    } else {
419                        Arc::new(crate::runtime::audit_log::AuditLogger::for_destination(
420                            &audit_dest,
421                            &data_path,
422                        ))
423                    }
424                },
425                control_event_ledger: parking_lot::RwLock::new(Arc::new(
426                    crate::runtime::control_events::RuntimeLedger::new(db.store()),
427                )),
428                control_event_config: options.control_events,
429                query_audit: Arc::new(crate::runtime::query_audit::QueryAuditStream::new(
430                    db.store(),
431                    options.query_audit.clone(),
432                )),
433                lease_lifecycle: std::sync::OnceLock::new(),
434                replica_apply_metrics: std::sync::Arc::new(
435                    crate::replication::logical::ReplicaApplyMetrics::default(),
436                ),
437                replica_link_metrics: std::sync::Arc::new(
438                    crate::replication::reconnect::ReplicaLinkMetrics::default(),
439                ),
440                quota_bucket: crate::runtime::quota_bucket::QuotaBucket::from_env(),
441                schema_vocabulary: parking_lot::RwLock::new(
442                    crate::runtime::schema_vocabulary::SchemaVocabulary::new(),
443                ),
444                slow_query_logger: {
445                    // Issue #205 — slow-query sink lives in the same
446                    // directory the audit log uses, so backup/restore
447                    // ships them together. Threshold + sample-pct
448                    // default conservatively (1 s, 100% sampling) so
449                    // emitted lines are rare and complete. Operators
450                    // tune via env / config matrix in a follow-up.
451                    //
452                    // gh-471 iter 2: same routing as the audit log —
453                    // `LogDestination::File(...)` for Performance/Max
454                    // lands under the file-owned support-directory logs tier;
455                    // lower tiers fall back to `red-slow.log` in the
456                    // data directory.
457                    // #1375 — see the audit-log note above: single-file mode
458                    // never writes the slow-query log as a sibling of the
459                    // `.rdb`. Route to a process-unique temp dir when embedded,
460                    // regardless of the data path.
461                    let fallback_dir = if embedded_single_file {
462                        std::env::temp_dir()
463                            .join("reddb-embedded-runtime")
464                            .join(format!("slow-{}", std::process::id()))
465                    } else {
466                        options
467                            .data_path
468                            .as_ref()
469                            .and_then(|p| p.parent().map(std::path::PathBuf::from))
470                            .unwrap_or_else(|| std::env::temp_dir().join("reddb"))
471                    };
472                    let threshold_ms = std::env::var("RED_SLOW_QUERY_THRESHOLD_MS")
473                        .ok()
474                        .and_then(|s| s.parse::<u64>().ok())
475                        .unwrap_or(1000);
476                    let sample_pct = std::env::var("RED_SLOW_QUERY_SAMPLE_PCT")
477                        .ok()
478                        .and_then(|s| s.parse::<u8>().ok())
479                        .unwrap_or(100);
480                    let (_, slow_dest) = crate::api::tier_wiring::current_log_destinations();
481                    crate::telemetry::slow_query_logger::SlowQueryLogger::for_destination(
482                        &slow_dest,
483                        &fallback_dir,
484                        threshold_ms,
485                        sample_pct,
486                    )
487                },
488                slow_query_store: crate::telemetry::slow_query_store::SlowQueryStore::new(
489                    crate::telemetry::slow_query_store::DEFAULT_CAP,
490                ),
491                kv_stats: crate::runtime::KvStatsCounters::default(),
492                metrics_ingest_stats: crate::runtime::MetricsIngestCounters::default(),
493                metrics_tenant_activity_stats:
494                    crate::runtime::MetricsTenantActivityCounters::default(),
495                claim_telemetry: Arc::new(
496                    crate::runtime::claim_telemetry::ClaimTelemetryCounters::default(),
497                ),
498                queue_telemetry: Arc::new(
499                    crate::runtime::queue_telemetry::QueueTelemetryCounters::default(),
500                ),
501                query_latency_telemetry: Arc::new(
502                    crate::runtime::query_latency_telemetry::QueryLatencyTelemetry::default(),
503                ),
504                occupancy_sampler: Arc::new(
505                    crate::runtime::occupancy_sampler::OccupancySampler::new(),
506                ),
507                node_load_telemetry: Arc::new(
508                    crate::runtime::node_load_telemetry::NodeLoadTelemetry::default(),
509                ),
510                queue_presence: Arc::new(
511                    crate::storage::queue::presence::ConsumerPresenceRegistry::new(),
512                ),
513                vector_introspection: Arc::new(
514                    crate::storage::vector::introspection::VectorIntrospectionRegistry::new(),
515                ),
516                kv_tag_index: crate::runtime::KvTagIndex::default(),
517                chain_tip_cache: parking_lot::Mutex::new(HashMap::new()),
518                chain_integrity_broken: parking_lot::Mutex::new(HashMap::new()),
519                integrity_tombstones: parking_lot::Mutex::new(Vec::new()),
520                integrity_tombstones_state: std::sync::atomic::AtomicU8::new(0),
521            }),
522        };
523
524        // Issue #205 — install the process-wide OperatorEvent sink so
525        // emit sites buried in storage / replication / signal handlers
526        // can record without threading an `&AuditLogger` through every
527        // call stack. First registration wins; subsequent in-memory
528        // runtimes (test harnesses) fall through to tracing+eprintln.
529        crate::telemetry::operator_event::install_global_audit_sink(Arc::clone(
530            &runtime.inner.audit_log,
531        ));
532
533        // Issue #1238 — wire the slow-query telemetry substrate (ADR 0060).
534        // The logger dual-writes: file sink (existing) + ring store (new).
535        runtime
536            .inner
537            .slow_query_logger
538            .attach_store(Arc::clone(&runtime.inner.slow_query_store));
539
540        // PLAN.md Phase 9.1 — backfill cold-start phase markers
541        // from the wall-clock captured before storage open. The
542        // entire `RedDB::open_with_options` call covers both
543        // auto-restore (when configured) and WAL replay. We
544        // record both phases against the same boundary today;
545        // a follow-up will split them once the storage layer
546        // surfaces a finer-grained event.
547        runtime
548            .inner
549            .lifecycle
550            .set_restore_started_at_ms(boot_open_start_ms);
551        runtime
552            .inner
553            .lifecycle
554            .set_restore_ready_at_ms(storage_ready_ms);
555        runtime
556            .inner
557            .lifecycle
558            .set_wal_replay_started_at_ms(boot_open_start_ms);
559        runtime
560            .inner
561            .lifecycle
562            .set_wal_replay_ready_at_ms(storage_ready_ms);
563
564        let restored_cdc_lsn = runtime
565            .inner
566            .db
567            .replication
568            .as_ref()
569            .map(|repl| {
570                repl.logical_wal_spool
571                    .as_ref()
572                    .map(|spool| spool.current_lsn())
573                    .unwrap_or(0)
574            })
575            .unwrap_or(0)
576            .max(runtime.config_u64("red.config.timeline.last_archived_lsn", 0));
577        runtime.inner.cdc.set_current_lsn(restored_cdc_lsn);
578        runtime.rehydrate_snapshot_xid_floor();
579        assert!(
580            runtime
581                .inner
582                .db
583                .store()
584                .snapshot_manager
585                .set(runtime.snapshot_manager())
586                .is_ok(),
587            "a new runtime owns the store transaction outcomes"
588        );
589        runtime
590            .bootstrap_system_keyed_collections()
591            .map_err(|err| RedDBError::Internal(format!("bootstrap system collections: {err}")))?;
592        runtime.rehydrate_declared_column_schemas();
593        runtime.rehydrate_runtime_index_registry()?;
594        runtime
595            .load_probabilistic_state()
596            .map_err(|err| RedDBError::Internal(format!("load probabilistic state: {err}")))?;
597
598        // Phase 2.5.4: replay `tenant_tables.{table}.column` markers so
599        // tables declared via `TENANT BY (col)` survive restart. Each
600        // entry re-registers the auto-policy and flips RLS on again.
601        runtime.rehydrate_tenant_tables();
602        // Issue #593 slice 9a — replay persisted materialized-view
603        // descriptors so `CREATE MATERIALIZED VIEW v AS …` survives a
604        // restart. Runs after the system-keyed collections bootstrap
605        // and before the API opens.
606        runtime.rehydrate_materialized_view_descriptors();
607        if let Some(repl) = &runtime.inner.db.replication {
608            repl.wal_buffer.set_current_lsn(restored_cdc_lsn);
609        }
610
611        // Save system info to red_config on boot
612        {
613            let sys = SystemInfo::collect();
614            runtime.inner.db.store().set_config_tree(
615                "red.system",
616                &crate::serde_json::json!({
617                    "pid": sys.pid,
618                    "cpu_cores": sys.cpu_cores,
619                    "total_memory_bytes": sys.total_memory_bytes,
620                    "available_memory_bytes": sys.available_memory_bytes,
621                    "os": sys.os,
622                    "arch": sys.arch,
623                    "hostname": sys.hostname,
624                    "started_at": SystemTime::now()
625                        .duration_since(UNIX_EPOCH)
626                        .unwrap_or_default()
627                        .as_millis() as u64
628                }),
629            );
630
631            // Seed defaults on first boot (only if red_config is empty or missing defaults)
632            let store = runtime.inner.db.store();
633            if store
634                .get_collection("red_config")
635                .map(|m| m.query_all(|_| true).len())
636                .unwrap_or(0)
637                <= 10
638            {
639                store.set_config_tree("red.ai", &crate::json!({
640                    "default": crate::json!({
641                        "provider": "openai",
642                        "model": crate::ai::DEFAULT_OPENAI_PROMPT_MODEL
643                    }),
644                    "max_embedding_inputs": 256,
645                    "max_prompt_batch": 256,
646                    "timeout": crate::json!({ "connect_secs": 10, "read_secs": 90, "write_secs": 30 })
647                }));
648                store.set_config_tree(
649                    "red.server",
650                    &crate::json!({
651                        "max_scan_limit": 1000,
652                        "max_body_size": 1048576,
653                        "read_timeout_ms": 5000,
654                        "write_timeout_ms": 5000
655                    }),
656                );
657                store.set_config_tree(
658                    "red.storage",
659                    &crate::json!({
660                        "page_size": 4096,
661                        "page_cache_capacity": 100000,
662                        "auto_checkpoint_pages": 1000,
663                        "snapshot_retention": 16,
664                        "verify_checksums": true,
665                        "segment": crate::json!({
666                            "max_entities": 100000,
667                            "max_bytes": 268435456_u64,
668                            "compression_level": 6
669                        }),
670                        "hnsw": crate::json!({ "m": 16, "ef_construction": 100, "ef_search": 50 }),
671                        "ivf": crate::json!({ "n_lists": 100, "n_probes": 10 }),
672                        "bm25": crate::json!({ "k1": 1.2, "b": 0.75 })
673                    }),
674                );
675                store.set_config_tree(
676                    "red.search",
677                    &crate::json!({
678                        "rag": crate::json!({
679                            "max_chunks_per_source": 10,
680                            "max_total_chunks": 25,
681                            "similarity_threshold": 0.8,
682                            "graph_depth": 2,
683                            "min_relevance": 0.3
684                        }),
685                        "fusion": crate::json!({
686                            "vector_weight": 0.5,
687                            "graph_weight": 0.3,
688                            "table_weight": 0.2,
689                            "dedup_threshold": 0.85
690                        })
691                    }),
692                );
693                store.set_config_tree(
694                    "red.auth",
695                    &crate::json!({
696                        "enabled": false,
697                        "session_ttl_secs": 3600,
698                        "require_auth": false
699                    }),
700                );
701                store.set_config_tree(
702                    "red.query",
703                    &crate::json!({
704                        "connection_pool": crate::json!({ "max_connections": 64, "max_idle": 16 }),
705                        "max_recursion_depth": 1000
706                    }),
707                );
708                store.set_config_tree(
709                    "red.indexes",
710                    &crate::json!({
711                        "auto_select": true,
712                        "bloom_filter": crate::json!({
713                            "enabled": true,
714                            "false_positive_rate": 0.01,
715                            "prune_on_scan": true
716                        }),
717                        "hash": crate::json!({ "enabled": true }),
718                        "bitmap": crate::json!({ "enabled": true, "max_cardinality": 1000 }),
719                        "spatial": crate::json!({ "enabled": true })
720                    }),
721                );
722                store.set_config_tree(
723                    "red.probabilistic",
724                    &crate::json!({
725                        "hll_registers": 16384,
726                        "sketch_default_width": 1000,
727                        "sketch_default_depth": 5,
728                        "filter_default_capacity": 100000
729                    }),
730                );
731                store.set_config_tree(
732                    "red.timeseries",
733                    &crate::json!({
734                        "default_chunk_size": 1024,
735                        "compression": crate::json!({
736                            "timestamps": "delta_of_delta",
737                            "values": "gorilla_xor"
738                        }),
739                        "default_retention_days": 0
740                    }),
741                );
742                store.set_config_tree(
743                    "red.queue",
744                    &crate::json!({
745                        "default_max_size": 0,
746                        "default_max_attempts": 3,
747                        "visibility_timeout_ms": 30000,
748                        "consumer_idle_timeout_ms": 60000
749                    }),
750                );
751                store.set_config_tree(
752                    "red.backup",
753                    &crate::json!({
754                        "enabled": false,
755                        "interval_secs": 3600,
756                        "retention_count": 24,
757                        "upload": false,
758                        "backend": "local"
759                    }),
760                );
761                store.set_config_tree(
762                    "red.wal",
763                    &crate::json!({
764                        "archive": crate::json!({
765                            "enabled": false,
766                            "retention_hours": 168,
767                            "prefix": reddb_file::backup_wal_prefix("")
768                        })
769                    }),
770                );
771                store.set_config_tree(
772                    "red.cdc",
773                    &crate::json!({
774                        "enabled": true,
775                        "buffer_size": 100000
776                    }),
777                );
778                store.set_config_tree(
779                    "red.config.secret",
780                    &crate::json!({
781                        "auto_encrypt": true,
782                        "auto_decrypt": true
783                    }),
784                );
785            }
786
787            // Perf-parity config matrix: heal the Tier A (critical)
788            // keys unconditionally on every boot. Idempotent — only
789            // writes the default when the key is missing. Keeps
790            // `SHOW CONFIG` showing every guarantee the operator has
791            // (durability.mode, concurrency.locking.enabled, …) even
792            // on long-running datadirs that predate the matrix.
793            crate::runtime::config_matrix::heal_critical_keys(store.as_ref());
794            seed_storage_deploy_config(store.as_ref(), options.storage_profile);
795
796            // Phase 5 — Lehman-Yao runtime flag. Read the Tier A
797            // `storage.btree.lehman_yao` value from the matrix (env
798            // > file > red_config > default) and publish it to the
799            // storage layer's atomic so the B-tree read / split
800            // paths can branch without re-reading the config on
801            // every hot-path call.
802            let lehman_yao = runtime.config_bool("storage.btree.lehman_yao", true);
803            crate::storage::engine::btree::lehman_yao::set_enabled(lehman_yao);
804            if lehman_yao {
805                tracing::info!(
806                    "storage.btree.lehman_yao=true — lock-free concurrent descent enabled"
807                );
808            }
809
810            // Config file overlay — mounted `/etc/reddb/config.json`
811            // (override path via REDDB_CONFIG_FILE). Writes keys with
812            // write-if-absent semantics so a later user `SET CONFIG`
813            // always wins. Missing file = silent no-op.
814            let overlay_path = crate::runtime::config_overlay::config_file_path();
815            let _ =
816                crate::runtime::config_overlay::apply_config_file(store.as_ref(), &overlay_path);
817        }
818
819        // VCS ("Git for Data") — create the `red_*` metadata
820        // collections on first boot. Idempotent: `get_or_create_collection`
821        // is a no-op if the collection already exists.
822        {
823            let store = runtime.inner.db.store();
824            for name in crate::application::vcs_collections::ALL {
825                let _ = store.get_or_create_collection(*name);
826            }
827            // Seed VCS config namespace with sensible defaults on first
828            // boot, matching the pattern used by red.ai / red.storage.
829            store.set_config_tree(
830                crate::application::vcs_collections::CONFIG_NAMESPACE,
831                &crate::json!({
832                    "default_branch": "main",
833                    "author": crate::json!({
834                        "name": "reddb",
835                        "email": "reddb@localhost"
836                    }),
837                    "protected_branches": crate::json!(["main"]),
838                    "closure": crate::json!({
839                        "enabled": true,
840                        "lazy": true
841                    }),
842                    "merge": crate::json!({
843                        "default_strategy": "auto",
844                        "fast_forward": true
845                    })
846                }),
847            );
848        }
849
850        // Migrations — create the `red_migrations` / `red_migration_deps`
851        // system collections on first boot. Idempotent.
852        {
853            let store = runtime.inner.db.store();
854            for name in crate::application::migration_collections::ALL {
855                let _ = store.get_or_create_collection(*name);
856            }
857        }
858
859        // Topology graph (#803) — ensure the built-in `red.topology.cluster`
860        // graph collection (declared WITH ANALYTICS) and its metadata sidecar
861        // exist. Idempotent and survives restarts via the WAL-backed contract.
862        let _ = crate::application::topology_collections::ensure(&runtime);
863
864        // #1369 — reserve a fixed internal-id floor so the first user-inserted
865        // entity always receives a stable, documented `rid` (FIRST_USER_ENTITY_ID),
866        // independent of how many internal collection-descriptor / config-default
867        // entities the boot sequence seeded above. `register_entity_id` only ever
868        // raises the allocator, so a database that already holds user data
869        // (counter past the floor) is untouched; a freshly-seeded database jumps
870        // straight to the floor.
871        runtime
872            .inner
873            .db
874            .store()
875            .register_entity_id(crate::storage::EntityId::new(
876                crate::storage::FIRST_USER_ENTITY_ID - 1,
877            ));
878
879        // Start background maintenance thread (context index refresh +
880        // session purge). Held by a WEAK reference to `RuntimeInner`
881        // to avoid an ownership cycle. The last foreground owner unparks
882        // and joins the worker before releasing its inner reference, so
883        // an active weak upgrade cannot retain the pager lock after close.
884        // The 60-second work cadence is tracked via `last_work`.
885        //
886        // The previous version captured `rt = runtime.clone()` by
887        // strong reference and ran an unterminated `loop`, which held
888        // Arc<RuntimeInner> forever — reopening a persistent database
889        // in the same process failed with "Database is locked" because
890        // the pager could never drop. See the regression test
891        // `finding_1_select_after_bulk_insert_persistent_reopen`.
892        {
893            let weak = Arc::downgrade(&runtime.inner);
894            runtime
895                .background_tasks
896                .as_ref()
897                .expect("foreground runtime")
898                .spawn(
899                    std::thread::Builder::new().name("reddb-maintenance".into()),
900                    move |stopped| {
901                        let tick = std::time::Duration::from_millis(200);
902                        let work_interval = std::time::Duration::from_secs(60);
903                        let mut last_work = std::time::Instant::now();
904                        loop {
905                            std::thread::park_timeout(tick);
906                            if stopped.load(AtomicOrdering::Acquire) {
907                                break;
908                            }
909                            let Some(inner) = weak.upgrade() else {
910                                // All strong references dropped — the
911                                // runtime is gone, exit cleanly.
912                                break;
913                            };
914                            if last_work.elapsed() >= work_interval {
915                                let _stats = inner.db.store().context_index().stats();
916                                last_work = std::time::Instant::now();
917                            }
918                        }
919                    },
920                );
921        }
922
923        // Start backup scheduler if enabled via red_config
924        {
925            let store = runtime.inner.db.store();
926            let mut backup_enabled = false;
927            let mut backup_interval = 3600u64;
928
929            if let Some(manager) = store.get_collection("red_config") {
930                manager.for_each_entity(|entity| {
931                    if let Some(row) = entity.data.as_row() {
932                        let key = row.get_field("key").and_then(|v| match v {
933                            reddb_types::Value::Text(s) => Some(s.as_ref()),
934                            _ => None,
935                        });
936                        let val = row.get_field("value");
937                        if key == Some("red.config.backup.enabled") {
938                            backup_enabled = match val {
939                                Some(reddb_types::Value::Boolean(true)) => true,
940                                Some(reddb_types::Value::Text(s)) => &**s == "true",
941                                _ => false,
942                            };
943                        } else if key == Some("red.config.backup.interval_secs") {
944                            if let Some(reddb_types::Value::Integer(n)) = val {
945                                backup_interval = *n as u64;
946                            }
947                        }
948                    }
949                    true
950                });
951            }
952
953            if backup_enabled {
954                runtime.inner.backup_scheduler.set_interval(backup_interval);
955                let rt = runtime.clone();
956                runtime
957                    .inner
958                    .backup_scheduler
959                    .start(move || rt.trigger_backup().map_err(|e| format!("{}", e)));
960            }
961        }
962
963        // Load EC registry from red_config and start worker
964        {
965            runtime
966                .inner
967                .ec_registry
968                .load_from_config_store(runtime.inner.db.store().as_ref());
969            if !runtime.inner.ec_registry.async_configs().is_empty() {
970                runtime.inner.ec_worker.start(
971                    Arc::clone(&runtime.inner.ec_registry),
972                    Arc::clone(&runtime.inner.db.store()),
973                );
974            }
975        }
976
977        if let (true, crate::replication::ReplicationRole::Replica { primary_addr }) = (
978            runtime.inner.db.options().replica_loop_enabled,
979            runtime.inner.db.options().replication.role.clone(),
980        ) {
981            let rt = runtime.clone();
982            std::thread::Builder::new()
983                .name("reddb-replica".into())
984                .spawn(move || rt.run_replica_loop(primary_addr))
985                .ok();
986        }
987
988        // PLAN.md Phase 1 — Lifecycle Contract. Mark Ready once every
989        // boot stage above has completed (WAL replay, restore-from-
990        // remote, replica-loop spawn). Health probes flip from 503 to
991        // 200 here; shutdown begins from this state.
992        runtime.inner.lifecycle.mark_ready();
993
994        // Issue #583 slice 10 — ContinuousMaterializedView scheduler.
995        // Low-priority background ticker that drains the cache's
996        // `claim_due_at` set every ~50ms. Holds only a Weak<RuntimeInner>
997        // to avoid an ownership cycle; foreground close unparks and drains
998        // an active refresh before releasing storage. Materialized views without
999        // a `REFRESH EVERY` clause stay on the manual-refresh path
1000        // and are skipped by `claim_due_at`, so the loop is a no-op
1001        // when no scheduled views exist.
1002        {
1003            let weak_inner = Arc::downgrade(&runtime.inner);
1004            runtime
1005                .background_tasks
1006                .as_ref()
1007                .expect("foreground runtime")
1008                .spawn(
1009                    std::thread::Builder::new()
1010                        .name("reddb-mv-scheduler".into())
1011                        // Rust's default for spawned threads is 2 MiB, which is too
1012                        // small for the query-execution path that refresh_due_materialized_views
1013                        // runs (StatementExecutionFrame + guards per view). Match the
1014                        // 8 MiB that the OS gives the main thread.
1015                        .stack_size(8 * 1024 * 1024),
1016                    move |stopped| loop {
1017                        std::thread::park_timeout(std::time::Duration::from_millis(50));
1018                        if stopped.load(AtomicOrdering::Acquire) {
1019                            break;
1020                        }
1021                        let Some(inner) = weak_inner.upgrade() else {
1022                            break;
1023                        };
1024                        let rt = RedDBRuntime {
1025                            background_tasks: None,
1026                            inner,
1027                        };
1028                        rt.refresh_due_materialized_views();
1029                    },
1030                );
1031        }
1032
1033        // Issue #584 slice 12 — DeclarativeRetention background sweeper.
1034        // Low-priority ticker that physically reclaims rows whose
1035        // timestamp has fallen beyond the retention window. Holds a
1036        // `Weak<RuntimeInner>`; foreground close drains an active tick
1037        // before releasing storage (shutdown leaves storage consistent
1038        // because each tick goes through the standard DELETE path —
1039        // there is no half-finished mutation state to clean up). The
1040        // tick interval is intentionally longer than the MV scheduler
1041        // (500ms) because retention is order-of-seconds at minimum.
1042        if !runtime.write_gate().is_read_only() {
1043            let weak_inner = Arc::downgrade(&runtime.inner);
1044            runtime
1045                .background_tasks
1046                .as_ref()
1047                .expect("foreground runtime")
1048                .spawn(
1049                    std::thread::Builder::new().name("reddb-retention-sweeper".into()),
1050                    move |stopped| loop {
1051                        std::thread::park_timeout(std::time::Duration::from_millis(500));
1052                        if stopped.load(AtomicOrdering::Acquire) {
1053                            break;
1054                        }
1055                        let Some(inner) = weak_inner.upgrade() else {
1056                            break;
1057                        };
1058                        let rt = RedDBRuntime {
1059                            background_tasks: None,
1060                            inner,
1061                        };
1062                        rt.sweep_retention_tick(
1063                            crate::runtime::retention_sweeper::DEFAULT_SWEEPER_BATCH,
1064                        );
1065                    },
1066                );
1067        }
1068
1069        // Anonymous access is the most consequential posture an operator can
1070        // choose. It was only ever announced on stderr; leave it in the
1071        // chained ledger too, so an audit of the store itself shows it.
1072        if no_auth_boot {
1073            use crate::runtime::control_events::{
1074                ActorRef, ControlEvent, ControlEventCtx, EventKind, Outcome,
1075            };
1076            let ledger = std::sync::Arc::clone(&runtime.inner.control_event_ledger.read());
1077            let ctx = ControlEventCtx {
1078                actor: ActorRef::System("boot"),
1079                scope: None,
1080                request_id: None,
1081                trace_id: None,
1082            };
1083            let event = ControlEvent {
1084                kind: EventKind::AuthDisabled,
1085                outcome: Outcome::Allowed,
1086                action: std::borrow::Cow::Borrowed("auth.disabled"),
1087                resource: Some("server".to_string()),
1088                reason: Some("--no-auth / --dev".to_string()),
1089                matched_policy_id: None,
1090                fields: std::collections::HashMap::new(),
1091            };
1092            if let Err(err) = ledger.emit(&ctx, event) {
1093                tracing::warn!(
1094                    target: "reddb::security",
1095                    error = %err,
1096                    "could not record the auth-disabled boot in the control-event ledger"
1097                );
1098            }
1099        }
1100
1101        Ok(runtime)
1102    }
1103
1104    fn rehydrate_snapshot_xid_floor(&self) {
1105        let store = self.inner.db.store();
1106        for collection in store.list_collections() {
1107            let Some(manager) = store.get_collection(&collection) else {
1108                continue;
1109            };
1110            for entity in manager.query_all(|_| true) {
1111                let snapshot_manager = self.inner.transaction_state.snapshot_manager();
1112                snapshot_manager.observe_committed_xid(entity.xmin);
1113                snapshot_manager.observe_committed_xid(entity.xmax);
1114            }
1115        }
1116    }
1117
1118    /// Provision an empty Table-shaped collection that backs a
1119    /// `CREATE MATERIALIZED VIEW v` (issue #594 slice 9b of #575).
1120    /// `SELECT FROM v` reads this collection directly; the rewriter is
1121    /// configured to skip materialized views so the body is no longer
1122    /// substituted. REFRESH still writes to the cache slot — wiring it
1123    /// into this backing collection is the job of slice 9c.
1124    ///
1125    /// Idempotent: re-running for the same name leaves the existing
1126    /// collection in place (mirrors `CREATE TABLE IF NOT EXISTS`
1127    /// semantics). This keeps `CREATE OR REPLACE MATERIALIZED VIEW v`
1128    /// cheap — the body change does not invalidate already-buffered
1129    /// rows. Until 9c lands the backing is always empty anyway.
1130    pub(crate) fn ensure_materialized_view_backing(&self, name: &str) -> RedDBResult<()> {
1131        let store = self.inner.db.store();
1132        let mut changed = false;
1133        if store.get_collection(name).is_none() {
1134            store.get_or_create_collection(name);
1135            changed = true;
1136        }
1137        if self.inner.db.collection_contract(name).is_none() {
1138            self.inner
1139                .db
1140                .save_collection_contract(system_keyed_collection_contract(
1141                    name,
1142                    crate::catalog::CollectionModel::Table,
1143                ))
1144                .map_err(|err| RedDBError::Internal(err.to_string()))?;
1145            changed = true;
1146        }
1147        if changed {
1148            self.inner
1149                .db
1150                .persist_metadata()
1151                .map_err(|err| RedDBError::Internal(err.to_string()))?;
1152        }
1153        Ok(())
1154    }
1155
1156    /// Inverse of [`ensure_materialized_view_backing`] — drops the
1157    /// backing collection on `DROP MATERIALIZED VIEW v`. No-op when
1158    /// the collection was never created (e.g. a `DROP MATERIALIZED
1159    /// VIEW IF EXISTS v` against an unknown name).
1160    pub(crate) fn drop_materialized_view_backing(&self, name: &str) -> RedDBResult<()> {
1161        let store = self.inner.db.store();
1162        if store.get_collection(name).is_none() {
1163            return Ok(());
1164        }
1165        store
1166            .drop_collection(name)
1167            .map_err(|err| RedDBError::Internal(err.to_string()))?;
1168        // The contract may have been dropped already (DROP TABLE path)
1169        // — ignore "not found" errors by checking presence first.
1170        if self.inner.db.collection_contract(name).is_some() {
1171            self.inner
1172                .db
1173                .remove_collection_contract(name)
1174                .map_err(|err| RedDBError::Internal(err.to_string()))?;
1175        }
1176        self.invalidate_result_cache();
1177        self.inner
1178            .db
1179            .persist_metadata()
1180            .map_err(|err| RedDBError::Internal(err.to_string()))?;
1181        Ok(())
1182    }
1183
1184    fn bootstrap_system_keyed_collections(&self) -> RedDBResult<()> {
1185        let mut changed = false;
1186        for (name, model) in [
1187            ("red.config", crate::catalog::CollectionModel::Config),
1188            ("red.vault", crate::catalog::CollectionModel::Vault),
1189            // Issue #593 — materialized-view catalog. One row per
1190            // `CREATE MATERIALIZED VIEW`; rehydrated at boot before
1191            // the API opens.
1192            (
1193                crate::runtime::continuous_materialized_view::CATALOG_COLLECTION,
1194                crate::catalog::CollectionModel::Config,
1195            ),
1196        ] {
1197            if self.inner.db.store().get_collection(name).is_none() {
1198                self.inner.db.store().get_or_create_collection(name);
1199                changed = true;
1200            }
1201            if self.inner.db.collection_contract(name).is_none() {
1202                self.inner
1203                    .db
1204                    .save_collection_contract(system_keyed_collection_contract(name, model))
1205                    .map_err(|err| RedDBError::Internal(err.to_string()))?;
1206                changed = true;
1207            }
1208        }
1209        if changed {
1210            self.inner
1211                .db
1212                .persist_metadata()
1213                .map_err(|err| RedDBError::Internal(err.to_string()))?;
1214        }
1215        Ok(())
1216    }
1217
1218    pub fn db(&self) -> Arc<RedDB> {
1219        Arc::clone(&self.inner.db)
1220    }
1221
1222    /// Direct access to the runtime's secondary-index store.
1223    /// Used by bulk-insert entry points (gRPC binary bulk, HTTP bulk,
1224    /// wire bulk) that need to push new rows through the per-index
1225    /// maintenance hook after `store.bulk_insert` returns.
1226    pub fn index_store_ref(&self) -> &super::index_store::IndexStore {
1227        &self.inner.index_store
1228    }
1229
1230    /// Apply a DDL event to the schema-vocabulary reverse index
1231    /// (issue #120). Called by DDL execution paths after the catalog
1232    /// mutation has succeeded so the index never holds entries for
1233    /// half-applied DDL.
1234    pub(crate) fn schema_vocabulary_apply(
1235        &self,
1236        event: crate::runtime::schema_vocabulary::DdlEvent,
1237    ) {
1238        self.inner.schema_vocabulary.write().on_ddl(event);
1239    }
1240
1241    /// Lookup `token` in the schema-vocabulary reverse index. Returns
1242    /// an owned `Vec<VocabHit>` because the underlying read lock
1243    /// cannot be borrowed across the call boundary; the slice from
1244    /// `SchemaVocabulary::lookup` is cloned per hit.
1245    pub fn schema_vocabulary_lookup(
1246        &self,
1247        token: &str,
1248    ) -> Vec<crate::runtime::schema_vocabulary::VocabHit> {
1249        self.inner.schema_vocabulary.read().lookup(token).to_vec()
1250    }
1251
1252    /// Inject an AuthStore into the runtime. Called by server boot
1253    /// after the vault has been bootstrapped, so that `Value::Secret`
1254    /// auto-encrypt/decrypt can reach the vault AES key.
1255    pub fn set_auth_store(&self, store: Arc<crate::auth::store::AuthStore>) {
1256        *self.inner.auth_store.write() = Some(store);
1257    }
1258
1259    /// Snapshot the current AuthStore (if any). Used by the wire listener
1260    /// to validate bearer tokens issued via HTTP `/auth/login`.
1261    pub fn auth_store(&self) -> Option<Arc<crate::auth::store::AuthStore>> {
1262        self.inner.auth_store.read().clone()
1263    }
1264
1265    /// Inject an `OAuthValidator` into the runtime. When set, HTTP and
1266    /// wire transports try OAuth JWT validation before falling back to
1267    /// the local AuthStore lookup. Pass `None` to disable.
1268    pub fn set_oauth_validator(&self, validator: Option<Arc<crate::auth::oauth::OAuthValidator>>) {
1269        *self.inner.oauth_validator.write() = validator;
1270    }
1271
1272    /// Returns a clone of the configured `OAuthValidator` Arc, if any.
1273    /// Hot path: called per HTTP request when an Authorization header
1274    /// is present, so we hand back a cheap Arc clone.
1275    pub fn oauth_validator(&self) -> Option<Arc<crate::auth::oauth::OAuthValidator>> {
1276        self.inner.oauth_validator.read().clone()
1277    }
1278
1279    /// Inject the browser-token authority (issue #936). When set, the
1280    /// RedWire WS handshake accepts the short-lived access JWT it mints
1281    /// (alongside, and tried before, the federated OAuth validator), and
1282    /// the `/auth/browser/*` HTTP endpoints can issue/rotate the pair.
1283    /// `None` leaves the browser credential flow inert.
1284    pub fn set_browser_token_authority(
1285        &self,
1286        authority: Option<Arc<crate::auth::browser_token::BrowserTokenAuthority>>,
1287    ) {
1288        *self.inner.browser_token_authority.write() = authority;
1289    }
1290
1291    /// Snapshot the browser-token authority, if wired. Read on the WS
1292    /// handshake path and by the `/auth/browser/*` handlers; a cheap Arc
1293    /// clone keeps the lock hold short.
1294    pub fn browser_token_authority(
1295        &self,
1296    ) -> Option<Arc<crate::auth::browser_token::BrowserTokenAuthority>> {
1297        self.inner.browser_token_authority.read().clone()
1298    }
1299}