1use super::impl_config_secret::seed_storage_deploy_config;
13use super::*;
14
15pub(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 pub fn lock_manager(&self) -> std::sync::Arc<crate::runtime::lock_manager::LockManager> {
127 self.inner.lock_manager.clone()
128 }
129
130 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 pub fn memory_budget(&self) -> crate::storage::memory_budget::MemoryBudget {
170 self.inner.memory_budget
171 }
172
173 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 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 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 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 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 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 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 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 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 crate::telemetry::operator_event::install_global_audit_sink(Arc::clone(
530 &runtime.inner.audit_log,
531 ));
532
533 runtime
536 .inner
537 .slow_query_logger
538 .attach_store(Arc::clone(&runtime.inner.slow_query_store));
539
540 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 runtime.rehydrate_tenant_tables();
602 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 {
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 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 crate::runtime::config_matrix::heal_critical_keys(store.as_ref());
794 seed_storage_deploy_config(store.as_ref(), options.storage_profile);
795
796 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 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 {
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 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 {
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 let _ = crate::application::topology_collections::ensure(&runtime);
863
864 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 {
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 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 {
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 {
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 runtime.inner.lifecycle.mark_ready();
993
994 {
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 .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 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 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 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 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 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 (
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 pub fn index_store_ref(&self) -> &super::index_store::IndexStore {
1227 &self.inner.index_store
1228 }
1229
1230 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 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 pub fn set_auth_store(&self, store: Arc<crate::auth::store::AuthStore>) {
1256 *self.inner.auth_store.write() = Some(store);
1257 }
1258
1259 pub fn auth_store(&self) -> Option<Arc<crate::auth::store::AuthStore>> {
1262 self.inner.auth_store.read().clone()
1263 }
1264
1265 pub fn set_oauth_validator(&self, validator: Option<Arc<crate::auth::oauth::OAuthValidator>>) {
1269 *self.inner.oauth_validator.write() = validator;
1270 }
1271
1272 pub fn oauth_validator(&self) -> Option<Arc<crate::auth::oauth::OAuthValidator>> {
1276 self.inner.oauth_validator.read().clone()
1277 }
1278
1279 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 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}