1use super::*;
2use crate::storage::unified::metadata::{MetadataFilter, MetadataValue};
3
4impl RedDB {
5 pub fn is_replica_role(&self) -> bool {
6 matches!(
7 self.options.replication.role,
8 crate::replication::ReplicationRole::Replica { .. }
9 )
10 }
11
12 pub fn enforce_retention_policy(&self) -> Result<(), Box<dyn std::error::Error>> {
13 if self.options.read_only || self.is_replica_role() {
14 return Ok(());
15 }
16
17 if self.options.mode == StorageMode::Persistent {
20 let Some(path) = self.path() else {
21 return Ok(());
22 };
23
24 let _publication = self.store.vault_publication_lock.lock();
25 let Ok(mut metadata) = self.load_or_bootstrap_physical_metadata(true) else {
26 return Ok(());
27 };
28
29 self.prune_export_registry(&mut metadata.exports);
30 metadata.save_for_data_path(path)?;
31 }
32
33 let _ = self.sweep_ttl_expired_entities()?;
34
35 Ok(())
36 }
37
38 pub(crate) fn ttl_expired_entities_now(
39 &self,
40 ) -> Result<Vec<(String, EntityId)>, Box<dyn std::error::Error>> {
41 self.ttl_expired_entities_at(current_unix_ms())
42 }
43
44 pub fn replica_allows_entity_at_read(
45 &self,
46 collection: &str,
47 entity: &crate::storage::UnifiedEntity,
48 ) -> bool {
49 if !self.is_replica_role() {
50 return true;
51 }
52 !self.entity_expired_at(collection, entity, current_unix_ms())
53 }
54
55 fn sweep_ttl_expired_entities(&self) -> Result<usize, Box<dyn std::error::Error>> {
56 let to_delete = self.ttl_expired_entities_now()?;
57
58 let mut deleted = 0usize;
59 for (collection, id) in to_delete {
60 match self.store.delete(&collection, id) {
61 Ok(true) => deleted = deleted.saturating_add(1),
62 Ok(false) => {}
63 Err(err) => {
64 return Err(format!(
65 "failed deleting expired entity {id} from collection '{collection}': {err:?}"
66 )
67 .into());
68 }
69 }
70 }
71
72 Ok(deleted)
73 }
74
75 fn ttl_expired_entities_at(
76 &self,
77 now_ms: u64,
78 ) -> Result<Vec<(String, EntityId)>, Box<dyn std::error::Error>> {
79 let mut to_delete = Vec::<(String, EntityId)>::new();
80
81 let mut absolute_expired = self.expired_entities_by_expires_at(now_ms)?;
82 to_delete.append(&mut absolute_expired);
83
84 let mut relative_expired = self.expired_entities_by_ttl(now_ms)?;
85 to_delete.append(&mut relative_expired);
86
87 to_delete.sort_unstable();
88 to_delete.dedup();
89
90 Ok(to_delete)
91 }
92
93 fn entity_expired_at(
94 &self,
95 collection: &str,
96 entity: &crate::storage::UnifiedEntity,
97 now_ms: u64,
98 ) -> bool {
99 let Some(metadata) = self.store.get_metadata(collection, entity.id) else {
100 return false;
101 };
102
103 if metadata
104 .get("_expires_at")
105 .and_then(Self::metadata_u64)
106 .is_some_and(|expires_at_ms| expires_at_ms <= now_ms)
107 {
108 return true;
109 }
110
111 let ttl_ms = metadata.get("_ttl_ms").and_then(Self::metadata_u64);
112 let ttl_secs = if ttl_ms.is_none() {
113 metadata.get("_ttl").and_then(|value| {
114 Self::metadata_u64(value).and_then(|value_secs| value_secs.checked_mul(1000))
115 })
116 } else {
117 None
118 };
119
120 let Some(ttl_ms) = ttl_ms.or(ttl_secs) else {
121 return false;
122 };
123 entity
124 .created_at
125 .saturating_mul(1000)
126 .saturating_add(ttl_ms)
127 <= now_ms
128 }
129
130 fn expired_entities_by_expires_at(
131 &self,
132 now_ms: u64,
133 ) -> Result<Vec<(String, EntityId)>, Box<dyn std::error::Error>> {
134 let mut ids = self.store.filter_metadata_all(&[(
135 "_expires_at".to_string(),
136 MetadataFilter::Le(MetadataValue::Timestamp(now_ms)),
137 )]);
138
139 if let Ok(now_ms_i64) = i64::try_from(now_ms) {
140 ids.extend(self.store.filter_metadata_all(&[(
141 "_expires_at".to_string(),
142 MetadataFilter::Le(MetadataValue::Int(now_ms_i64)),
143 )]));
144 }
145
146 let now_ms_f64 = now_ms as f64;
147 if now_ms_f64.is_finite() {
148 ids.extend(self.store.filter_metadata_all(&[(
149 "_expires_at".to_string(),
150 MetadataFilter::Le(MetadataValue::Float(now_ms_f64)),
151 )]));
152 }
153
154 Ok(ids)
155 }
156
157 fn expired_entities_by_ttl(
158 &self,
159 now_ms: u64,
160 ) -> Result<Vec<(String, EntityId)>, Box<dyn std::error::Error>> {
161 let mut candidates = Vec::<(String, EntityId)>::new();
162
163 let ttl_ms_candidates = self
164 .store
165 .filter_metadata_all(&[("_ttl_ms".to_string(), MetadataFilter::IsNotNull)]);
166 candidates.extend(ttl_ms_candidates);
167
168 let ttl_candidates = self
169 .store
170 .filter_metadata_all(&[("_ttl".to_string(), MetadataFilter::IsNotNull)]);
171 candidates.extend(ttl_candidates);
172
173 if candidates.is_empty() {
174 return Ok(Vec::new());
175 }
176
177 candidates.sort_unstable();
178 candidates.dedup();
179
180 let mut expired = Vec::<(String, EntityId)>::new();
181 for (collection, entity_id) in candidates {
182 let Some(entity) = self.store.get(&collection, entity_id) else {
183 continue;
184 };
185
186 let Some(metadata) = self.store.get_metadata(&collection, entity_id) else {
187 continue;
188 };
189
190 let ttl_ms = metadata.get("_ttl_ms").and_then(Self::metadata_u64);
191 let ttl_secs = if ttl_ms.is_none() {
192 metadata.get("_ttl").and_then(|value| {
193 Self::metadata_u64(value).and_then(|value_secs| value_secs.checked_mul(1000))
194 })
195 } else {
196 None
197 };
198
199 let Some(ttl_ms) = ttl_ms.or(ttl_secs) else {
200 continue;
201 };
202
203 let created_at_ms = entity.created_at.saturating_mul(1000);
204 let expiry_ms = created_at_ms.saturating_add(ttl_ms);
205 if expiry_ms <= now_ms {
206 expired.push((collection, entity_id));
207 }
208 }
209
210 Ok(expired)
211 }
212
213 fn metadata_u64(value: &MetadataValue) -> Option<u64> {
214 match value {
215 MetadataValue::Int(v) if *v >= 0 => Some(*v as u64),
216 MetadataValue::Timestamp(v) => Some(*v),
217 MetadataValue::Float(v) => {
218 if !v.is_finite() || !v.is_sign_positive() || v.fract().abs() >= f64::EPSILON {
219 return None;
220 }
221 if *v > u64::MAX as f64 {
222 return None;
223 }
224 Some(v.trunc() as u64)
225 }
226 MetadataValue::String(v) => v.parse::<u64>().ok(),
227 _ => None,
228 }
229 }
230
231 pub fn node(&self, collection: impl Into<String>, label: impl Into<String>) -> NodeBuilder {
244 NodeBuilder::new(
245 self.store.clone(),
246 self.preprocessors.clone(),
247 collection,
248 label,
249 )
250 }
251
252 pub fn edge(&self, collection: impl Into<String>, label: impl Into<String>) -> EdgeBuilder {
264 EdgeBuilder::new(
265 self.store.clone(),
266 self.preprocessors.clone(),
267 collection,
268 label,
269 )
270 }
271
272 pub fn vector(&self, collection: impl Into<String>) -> VectorBuilder {
283 VectorBuilder::new(self.store.clone(), self.preprocessors.clone(), collection)
284 }
285
286 pub fn row(&self, table: impl Into<String>, columns: Vec<(&str, Value)>) -> RowBuilder {
297 RowBuilder::new(
298 self.store.clone(),
299 self.preprocessors.clone(),
300 table,
301 columns,
302 )
303 }
304
305 pub fn doc(&self, collection: impl Into<String>) -> DocumentBuilder {
319 DocumentBuilder::new(self.store.clone(), self.preprocessors.clone(), collection)
320 }
321
322 pub fn kv(
333 &self,
334 collection: impl Into<String>,
335 key: impl Into<String>,
336 value: Value,
337 ) -> KvBuilder {
338 KvBuilder::new(
339 self.store.clone(),
340 self.preprocessors.clone(),
341 collection,
342 key,
343 value,
344 )
345 }
346
347 pub fn get_kv(&self, collection: &str, key: &str) -> Option<(Value, EntityId)> {
352 let manager = self.store.get_collection(collection)?;
353 if let Some(value) = manager.project_key_entity(key, None, false, |entity| {
354 (
355 entity
356 .data
357 .as_row()
358 .and_then(|row| row.get_field("value"))
359 .cloned()
360 .unwrap_or(Value::Null),
361 entity.id,
362 )
363 }) {
364 return Some(value);
365 }
366 let columns = manager.column_schema();
367 let key_index = columns
368 .as_ref()
369 .and_then(|names| names.iter().position(|name| name == "key"));
370 let value_index = columns
371 .as_ref()
372 .and_then(|names| names.iter().position(|name| name == "value"));
373 let mut found = None;
374 manager.for_each_entity(|entity| {
375 let EntityData::Row(row) = &entity.data else {
376 return true;
377 };
378 let legacy = row.named.is_none() && row.schema.is_none();
379 let stored_key = row.get_field("key").or_else(|| {
380 if legacy {
381 key_index.and_then(|index| row.columns.get(index))
382 } else {
383 None
384 }
385 });
386 if matches!(stored_key, Some(Value::Text(stored)) if stored.as_ref() == key) {
387 let value = row.get_field("value").or_else(|| {
388 if legacy {
389 value_index.and_then(|index| row.columns.get(index))
390 } else {
391 None
392 }
393 });
394 found = Some((value.cloned().unwrap_or(Value::Null), entity.id));
395 return false;
396 }
397 true
398 });
399 found
400 }
401
402 pub fn delete_kv(
404 &self,
405 collection: &str,
406 key: &str,
407 ) -> Result<bool, super::super::error::DevXError> {
408 let Some((_, id)) = self.get_kv(collection, key) else {
409 return Ok(false);
410 };
411 self.store
412 .delete(collection, id)
413 .map_err(|err| super::super::error::DevXError::Storage(format!("{err:?}")))?;
414 Ok(true)
415 }
416
417 pub(crate) fn with_initialized_metadata(self) -> Result<Self, Box<dyn std::error::Error>> {
418 if self.options.mode == StorageMode::Persistent {
419 match self.load_or_bootstrap_physical_metadata(false) {
421 Ok(metadata) => {
422 crate::reserved_fields::validate_physical_metadata_contracts(&metadata)
423 .map_err(|err| err.to_string())?;
424 }
425 Err(error)
426 if error
427 .downcast_ref::<std::io::Error>()
428 .is_some_and(crate::physical::is_collection_contract_decode_error) =>
429 {
430 return Err(error)
431 }
432 Err(_) => {}
433 }
434 }
438 self.load_collection_ttl_defaults_from_metadata();
439 if self.options.storage_profile.deploy_profile == crate::storage::DeployProfile::Embedded
446 && self.options.storage_profile.packaging
447 == crate::storage::StoragePackaging::SingleFile
448 && !self.paged_mode
449 {
450 self.seed_contract_cache_from_store_aux()?;
451 }
452 self.load_hypertables_from_metadata();
456 Ok(self)
460 }
461
462 pub(crate) fn persist_metadata(&self) -> Result<(), Box<dyn std::error::Error>> {
463 if self.options.mode != StorageMode::Persistent || self.options.read_only {
464 return Ok(());
465 }
466 if !self.paged_mode
469 && self.options.storage_profile.deploy_profile
470 == crate::storage::DeployProfile::Embedded
471 && self.options.storage_profile.packaging
472 == crate::storage::StoragePackaging::SingleFile
473 {
474 return Ok(());
475 }
476 let Some(path) = self.path() else {
477 return Ok(());
478 };
479
480 let _publication = self.store.vault_publication_lock.lock();
481 let previous = self.load_or_bootstrap_physical_metadata(false).ok();
482 let collection_roots = self.physical_collection_roots();
483 let indexes = self
484 .native_physical_state()
485 .map(|state| self.physical_index_state_from_native_state(&state, previous.as_ref()))
486 .unwrap_or_else(|| self.physical_index_state());
487 let mut metadata = PhysicalMetadataFile::from_state(
488 self.options.clone(),
489 self.catalog_snapshot(),
490 collection_roots,
491 indexes,
492 previous.as_ref(),
493 );
494 metadata.collection_ttl_defaults_ms = self.collection_ttl_defaults_snapshot();
495 metadata.hypertables = self.hypertable_registry_snapshot();
499 metadata.save_for_data_path(path)?;
500 self.persist_native_physical_header(&metadata)?;
501 Ok(())
502 }
503
504 fn bootstrap_metadata_from_native_state(&self) -> Result<bool, Box<dyn std::error::Error>> {
505 if self.options.mode != StorageMode::Persistent || self.options.read_only {
506 return Ok(false);
507 }
508 let Some(path) = self.path() else {
509 return Ok(false);
510 };
511 let Some(native_state) = self.native_physical_state() else {
512 return Ok(false);
513 };
514 if !Self::native_state_is_bootstrap_complete(&native_state) {
515 return Ok(false);
516 }
517
518 let _publication = self.store.vault_publication_lock.lock();
519 let previous = PhysicalMetadataFile::load_for_data_path(path).ok();
520 let metadata = self.metadata_from_native_state(&native_state, previous.as_ref());
521 metadata.save_for_data_path(path)?;
522 self.persist_native_physical_header(&metadata)?;
523 Ok(true)
524 }
525
526 pub fn rebuild_physical_metadata_from_native_state(
529 &self,
530 ) -> Result<bool, Box<dyn std::error::Error>> {
531 self.bootstrap_metadata_from_native_state()
532 }
533
534 pub(crate) fn native_state_is_bootstrap_complete(native_state: &NativePhysicalState) -> bool {
535 let registry_complete = native_state.registry.as_ref().map(|registry| {
536 registry.collections_complete
537 && registry.indexes_complete
538 && registry.graph_projections_complete
539 && registry.analytics_jobs_complete
540 && registry.vector_artifacts_complete
541 });
542 let recovery_complete = native_state
543 .recovery
544 .as_ref()
545 .map(|recovery| recovery.snapshots_complete && recovery.exports_complete);
546 let catalog_complete = native_state
547 .catalog
548 .as_ref()
549 .map(|catalog| catalog.collections_complete);
550
551 registry_complete == Some(true)
552 && recovery_complete == Some(true)
553 && catalog_complete == Some(true)
554 }
555
556 pub(crate) fn load_or_bootstrap_physical_metadata(
557 &self,
558 persist_bootstrapped: bool,
559 ) -> Result<PhysicalMetadataFile, Box<dyn std::error::Error>> {
560 if self.options.mode != StorageMode::Persistent {
561 return Err("physical metadata requires persistent mode".into());
562 }
563 let Some(path) = self.path() else {
564 return Err("database path is not available".into());
565 };
566 let _publication = self.store.vault_publication_lock.lock();
567 let native_state = self.native_physical_state();
568
569 let metadata = if persist_bootstrapped && !self.options.read_only {
570 PhysicalMetadataFile::load_for_data_path(path)
571 } else {
572 PhysicalMetadataFile::inspect_for_data_path(path)
573 };
574 match metadata {
575 Ok(metadata) => {
576 if let Some(native_state) = native_state.as_ref() {
577 let inspection = Self::inspect_native_header_against_metadata(
578 native_state.header,
579 &metadata,
580 );
581 if Self::repair_policy_for_inspection(&inspection)
582 == NativeHeaderRepairPolicy::NativeAheadOfMetadata
583 {
584 let bootstrapped =
585 self.metadata_from_native_state(native_state, Some(&metadata));
586 if persist_bootstrapped && !self.options.read_only {
587 bootstrapped.save_for_data_path(path)?;
588 self.persist_native_physical_header(&bootstrapped)?;
589 }
590 return Ok(bootstrapped);
591 }
592 }
593 Ok(metadata)
594 }
595 Err(err) => {
596 if crate::physical::is_collection_contract_decode_error(&err) {
597 return Err(err.into());
598 }
599 let Some(native_state) = native_state else {
600 return Err(err.into());
601 };
602 let is_fresh_empty = native_state.header.sequence == 0
618 && native_state.registry.is_none()
619 && native_state.catalog.is_none()
620 && native_state.recovery.is_none();
621 if !is_fresh_empty && !Self::native_state_is_bootstrap_complete(&native_state) {
622 return Err(err.into());
623 }
624 let metadata = self.metadata_from_native_state(&native_state, None);
625 if persist_bootstrapped && !self.options.read_only {
626 metadata.save_for_data_path(path)?;
627 self.persist_native_physical_header(&metadata)?;
628 }
629 Ok(metadata)
630 }
631 }
632 }
633
634 pub(crate) fn physical_metadata_preference(&self) -> Option<&'static str> {
635 let path = self.path()?;
636 let native_state = self.native_physical_state();
637 let metadata = PhysicalMetadataFile::inspect_for_data_path(path).ok();
638
639 match (metadata, native_state) {
640 (Some(metadata), Some(native_state)) => {
641 let inspection =
642 Self::inspect_native_header_against_metadata(native_state.header, &metadata);
643 match Self::repair_policy_for_inspection(&inspection) {
644 NativeHeaderRepairPolicy::InSync => Some("sidecar_current"),
645 NativeHeaderRepairPolicy::RepairNativeFromMetadata => Some("sidecar_current"),
646 NativeHeaderRepairPolicy::NativeAheadOfMetadata => Some("native_ahead"),
647 }
648 }
649 (Some(_), None) => Some("sidecar_only"),
650 (None, Some(_)) => Some("sidecar_missing_native_available"),
651 (None, None) => Some("sidecar_missing_no_native"),
652 }
653 }
654
655 fn metadata_from_native_state(
656 &self,
657 native_state: &NativePhysicalState,
658 previous: Option<&PhysicalMetadataFile>,
659 ) -> PhysicalMetadataFile {
660 let now = SystemTime::now()
661 .duration_since(UNIX_EPOCH)
662 .unwrap_or_default()
663 .as_millis();
664 let catalog = self.catalog_snapshot();
665 let catalog_name = catalog.name.clone();
666 let catalog_total_entities = catalog.total_entities;
667 let catalog_total_collections = catalog.total_collections;
668 let indexes = self.physical_index_state();
669
670 let mut manifest =
671 crate::api::SchemaManifest::now(self.options.clone(), catalog.total_collections);
672 manifest.updated_at_unix_ms = now;
673
674 let manifest_events = native_state
675 .manifest
676 .as_ref()
677 .map(|summary| {
678 summary
679 .recent_events
680 .iter()
681 .map(|event| crate::physical::ManifestEvent {
682 collection: event.collection.clone(),
683 object_key: event.object_key.clone(),
684 kind: match event.kind.as_str() {
685 "insert" => crate::physical::ManifestEventKind::Insert,
686 "update" => crate::physical::ManifestEventKind::Update,
687 "remove" => crate::physical::ManifestEventKind::Remove,
688 _ => crate::physical::ManifestEventKind::Checkpoint,
689 },
690 block: crate::physical::BlockReference {
691 index: event.block_index,
692 checksum: event.block_checksum,
693 },
694 snapshot_min: event.snapshot_min,
695 snapshot_max: event.snapshot_max,
696 })
697 .collect()
698 })
699 .unwrap_or_default();
700
701 let graph_projections = native_state
702 .registry
703 .as_ref()
704 .and_then(|registry| {
705 registry.graph_projections_complete.then(|| {
706 registry
707 .graph_projections
708 .iter()
709 .map(|projection| crate::physical::PhysicalGraphProjection {
710 name: projection.name.clone(),
711 created_at_unix_ms: projection.created_at_unix_ms,
712 updated_at_unix_ms: projection.updated_at_unix_ms,
713 state: "materialized".to_string(),
714 source: projection.source.clone(),
715 node_labels: projection.node_labels.clone(),
716 node_types: projection.node_types.clone(),
717 edge_labels: projection.edge_labels.clone(),
718 last_materialized_sequence: projection.last_materialized_sequence,
719 })
720 .collect()
721 })
722 })
723 .or_else(|| previous.map(|metadata| metadata.graph_projections.clone()))
724 .unwrap_or_default();
725
726 let analytics_jobs = native_state
727 .registry
728 .as_ref()
729 .and_then(|registry| {
730 registry.analytics_jobs_complete.then(|| {
731 registry
732 .analytics_jobs
733 .iter()
734 .map(|job| crate::physical::PhysicalAnalyticsJob {
735 id: job.id.clone(),
736 kind: job.kind.clone(),
737 state: job.state.clone(),
738 projection: job.projection.clone(),
739 created_at_unix_ms: job.created_at_unix_ms,
740 updated_at_unix_ms: job.updated_at_unix_ms,
741 last_run_sequence: job.last_run_sequence,
742 metadata: job.metadata.clone(),
743 })
744 .collect()
745 })
746 })
747 .or_else(|| previous.map(|metadata| metadata.analytics_jobs.clone()))
748 .unwrap_or_default();
749
750 let exports = native_state
751 .recovery
752 .as_ref()
753 .and_then(|recovery| {
754 recovery.exports_complete.then(|| {
755 recovery
756 .exports
757 .iter()
758 .map(|export| crate::physical::ExportDescriptor {
759 name: export.name.clone(),
760 created_at_unix_ms: export.created_at_unix_ms,
761 snapshot_id: export.snapshot_id,
762 superblock_sequence: export.superblock_sequence,
763 data_path: self
764 .path()
765 .map(|path| {
766 crate::physical::PhysicalMetadataFile::export_data_path_for(
767 path,
768 &export.name,
769 )
770 .display()
771 .to_string()
772 })
773 .unwrap_or_default(),
774 metadata_path: self
775 .path()
776 .map(|path| {
777 let export_data_path =
778 crate::physical::PhysicalMetadataFile::export_data_path_for(
779 path,
780 &export.name,
781 );
782 crate::physical::PhysicalMetadataFile::metadata_path_for(
783 &export_data_path,
784 )
785 .display()
786 .to_string()
787 })
788 .unwrap_or_default(),
789 collection_count: export.collection_count as usize,
790 total_entities: export.total_entities as usize,
791 })
792 .collect()
793 })
794 })
795 .or_else(|| previous.map(|metadata| metadata.exports.clone()))
796 .unwrap_or_default();
797
798 let snapshots = native_state
799 .recovery
800 .as_ref()
801 .and_then(|recovery| {
802 recovery.snapshots_complete.then(|| {
803 recovery
804 .snapshots
805 .iter()
806 .map(|snapshot| crate::physical::SnapshotDescriptor {
807 snapshot_id: snapshot.snapshot_id,
808 created_at_unix_ms: snapshot.created_at_unix_ms,
809 superblock_sequence: snapshot.superblock_sequence,
810 collection_count: snapshot.collection_count as usize,
811 total_entities: snapshot.total_entities as usize,
812 })
813 .collect()
814 })
815 })
816 .or_else(|| previous.map(|metadata| metadata.snapshots.clone()))
817 .unwrap_or_else(|| {
818 vec![crate::physical::SnapshotDescriptor {
819 snapshot_id: native_state.header.sequence,
820 created_at_unix_ms: now,
821 superblock_sequence: native_state.header.sequence,
822 collection_count: catalog_total_collections,
823 total_entities: catalog_total_entities,
824 }]
825 });
826
827 let catalog_stats = native_state
828 .catalog
829 .as_ref()
830 .and_then(|native_catalog| {
831 native_catalog.collections_complete.then(|| {
832 native_catalog
833 .collections
834 .iter()
835 .map(|collection| {
836 (
837 collection.name.clone(),
838 crate::api::CollectionStats {
839 entities: collection.entities as usize,
840 cross_refs: collection.cross_refs as usize,
841 segments: collection.segments as usize,
842 },
843 )
844 })
845 .collect::<BTreeMap<_, _>>()
846 })
847 })
848 .or_else(|| previous.map(|metadata| metadata.catalog.stats_by_collection.clone()))
849 .unwrap_or_else(|| catalog.stats_by_collection.clone());
850
851 PhysicalMetadataFile {
852 protocol_version: crate::physical::PHYSICAL_METADATA_PROTOCOL_VERSION.to_string(),
853 generated_at_unix_ms: now,
854 last_loaded_from: Some("native_bootstrap".to_string()),
855 last_healed_at_unix_ms: Some(now),
856 manifest,
857 catalog: crate::api::CatalogSnapshot {
858 name: catalog_name,
859 total_entities: native_state
860 .catalog
861 .as_ref()
862 .map(|summary| summary.total_entities as usize)
863 .unwrap_or(catalog_total_entities),
864 total_collections: native_state
865 .catalog
866 .as_ref()
867 .map(|summary| summary.collection_count as usize)
868 .unwrap_or(catalog_total_collections),
869 stats_by_collection: catalog_stats,
870 updated_at: SystemTime::now(),
871 },
872 manifest_events,
873 collection_ttl_defaults_ms: previous
874 .map(|metadata| metadata.collection_ttl_defaults_ms.clone())
875 .unwrap_or_default(),
876 collection_contracts: previous
877 .map(|metadata| metadata.collection_contracts.clone())
878 .unwrap_or_default(),
879 hypertables: previous
880 .map(|metadata| metadata.hypertables.clone())
881 .unwrap_or_default(),
882 tree_definitions: previous
883 .map(|metadata| metadata.tree_definitions.clone())
884 .unwrap_or_default(),
885 indexes,
886 graph_projections,
887 analytics_jobs,
888 exports,
889 superblock: crate::physical::SuperblockHeader {
890 format_version: native_state.header.format_version,
891 sequence: native_state.header.sequence,
892 copies: crate::physical::DEFAULT_SUPERBLOCK_COPIES,
893 manifest: crate::physical::ManifestPointers {
894 oldest: crate::physical::BlockReference {
895 index: native_state.header.manifest_oldest_root,
896 checksum: 0,
897 },
898 newest: crate::physical::BlockReference {
899 index: native_state.header.manifest_root,
900 checksum: 0,
901 },
902 },
903 free_set: crate::physical::BlockReference {
904 index: native_state.header.free_set_root,
905 checksum: 0,
906 },
907 collection_roots: native_state.collection_roots.clone(),
908 },
909 snapshots,
910 }
911 }
912
913 pub(crate) fn reconcile_index_states_with_native_artifacts(
914 &self,
915 mut indexes: Vec<PhysicalIndexState>,
916 ) -> Vec<PhysicalIndexState> {
917 let native_artifacts = self
918 .native_physical_state()
919 .and_then(|state| state.registry)
920 .map(|registry| registry.vector_artifacts)
921 .unwrap_or_default();
922 for index in &mut indexes {
923 let Some(collection) = index.collection.as_deref() else {
924 continue;
925 };
926 let Some(artifact_kind) = Self::native_artifact_kind_for_index(index.kind) else {
927 continue;
928 };
929 let Some(artifact) = native_artifacts.iter().find(|artifact| {
930 artifact.collection == collection && artifact.artifact_kind == artifact_kind
931 }) else {
932 index.build_state = if index.backend == "memory-budget-deferred" {
933 "deferred-memory-budget"
934 } else {
935 "metadata-only"
936 }
937 .to_string();
938 continue;
939 };
940 index.entries = artifact.vector_count as usize;
941 index.estimated_memory_bytes = artifact.serialized_bytes;
942 index.backend = format!("{}+native-artifact", index_backend_name(index.kind));
943 index.artifact_kind = Some(artifact.artifact_kind.clone());
944 index.artifact_checksum = Some(artifact.checksum);
945 index.build_state = "artifact-published".to_string();
946 if let Some(pages) = self.native_vector_artifact_pages() {
947 index.artifact_root_page = pages
948 .into_iter()
949 .find(|page| {
950 page.collection == artifact.collection
951 && page.artifact_kind == artifact.artifact_kind
952 })
953 .map(|page| page.root_page);
954 }
955 }
956 indexes
957 }
958
959 pub(crate) fn warmup_native_vector_artifact_for_index(
960 &self,
961 index: &PhysicalIndexState,
962 ) -> Result<(), String> {
963 let Some(collection) = index.collection.as_deref() else {
964 return Ok(());
965 };
966 let Some(artifact_kind) = Self::native_artifact_kind_for_index(index.kind) else {
967 return Ok(());
968 };
969 self.warmup_native_vector_artifact(collection, Some(artifact_kind))?;
970 Ok(())
971 }
972
973 pub(crate) fn apply_runtime_native_artifact_to_index_state(
974 &self,
975 index: &mut PhysicalIndexState,
976 ) -> Result<(), String> {
977 let Some(collection) = index.collection.as_deref() else {
978 return Ok(());
979 };
980 let Some(artifact_kind) = Self::native_artifact_kind_for_index(index.kind) else {
981 return Ok(());
982 };
983 let artifact = self.inspect_native_vector_artifact(collection, Some(artifact_kind))?;
984 index.entries = artifact
985 .graph_edge_count
986 .or(artifact.text_posting_count)
987 .unwrap_or(artifact.node_count) as usize;
988 index.estimated_memory_bytes = artifact.byte_len;
989 index.backend = format!("{}+native-artifact", index_backend_name(index.kind));
990 index.artifact_kind = Some(artifact.artifact_kind.clone());
991 index.artifact_checksum = Some(artifact.checksum);
992 index.build_state = "ready".to_string();
993 index.artifact_root_page = self
994 .native_vector_artifact_pages()
995 .and_then(|pages| {
996 pages.into_iter().find(|page| {
997 page.collection == artifact.collection
998 && page.artifact_kind == artifact.artifact_kind
999 })
1000 })
1001 .map(|page| page.root_page);
1002 Ok(())
1003 }
1004
1005 pub(crate) fn physical_index_state_from_native_state(
1006 &self,
1007 native_state: &NativePhysicalState,
1008 previous: Option<&PhysicalMetadataFile>,
1009 ) -> Vec<PhysicalIndexState> {
1010 let mut fresh = self.physical_index_state();
1011 let Some(registry) = native_state.registry.as_ref() else {
1012 if let Some(previous) = previous {
1013 for index in &previous.indexes {
1014 if !fresh.iter().any(|candidate| candidate.name == index.name) {
1015 fresh.push(index.clone());
1016 }
1017 }
1018 }
1019 return fresh;
1020 };
1021
1022 for index in &mut fresh {
1023 if let Some(native) = registry
1024 .indexes
1025 .iter()
1026 .find(|candidate| candidate.name == index.name)
1027 {
1028 index.enabled = native.enabled;
1029 index.last_refresh_ms = native.last_refresh_ms;
1030 index.backend = native.backend.clone();
1031 index.entries = native.entries as usize;
1032 index.estimated_memory_bytes = native.estimated_memory_bytes;
1033 if index.artifact_kind.is_none() {
1034 index.artifact_kind = Self::native_artifact_kind_for_index(index.kind)
1035 .map(|value| value.to_string());
1036 }
1037 if index.build_state == "catalog-derived" {
1038 index.build_state = "registry-loaded".to_string();
1039 }
1040 }
1041 if index.backend == "memory-budget-deferred"
1042 || previous.is_some_and(|metadata| {
1043 metadata.indexes.iter().any(|previous| {
1044 previous.name == index.name && previous.backend == "memory-budget-deferred"
1045 })
1046 })
1047 {
1048 index.backend = "memory-budget-deferred".to_string();
1049 index.build_state = "deferred-memory-budget".to_string();
1050 index.artifact_root_page = None;
1051 index.artifact_checksum = None;
1052 }
1053 }
1054
1055 for native in ®istry.indexes {
1056 if fresh.iter().any(|index| index.name == native.name) {
1057 continue;
1058 }
1059 let Some(kind) = Self::index_kind_from_str(&native.kind) else {
1060 continue;
1061 };
1062 fresh.push(PhysicalIndexState {
1063 name: native.name.clone(),
1064 kind,
1065 collection: native.collection.clone(),
1066 enabled: native.enabled,
1067 entries: native.entries as usize,
1068 estimated_memory_bytes: native.estimated_memory_bytes,
1069 last_refresh_ms: native.last_refresh_ms,
1070 backend: native.backend.clone(),
1071 artifact_kind: Self::native_artifact_kind_for_index(kind)
1072 .map(|value| value.to_string()),
1073 artifact_root_page: None,
1074 artifact_checksum: None,
1075 build_state: "registry-loaded".to_string(),
1076 });
1077 }
1078
1079 if !registry.indexes_complete {
1080 if let Some(previous) = previous {
1081 for index in &previous.indexes {
1082 if !fresh.iter().any(|candidate| candidate.name == index.name) {
1083 fresh.push(index.clone());
1084 }
1085 }
1086 }
1087 }
1088
1089 fresh
1090 }
1091
1092 pub(crate) fn graph_projections_from_native_state(
1093 &self,
1094 native_state: &NativePhysicalState,
1095 ) -> Vec<PhysicalGraphProjection> {
1096 native_state
1097 .registry
1098 .as_ref()
1099 .map(|registry| {
1100 registry
1101 .graph_projections
1102 .iter()
1103 .map(|projection| PhysicalGraphProjection {
1104 name: projection.name.clone(),
1105 created_at_unix_ms: projection.created_at_unix_ms,
1106 updated_at_unix_ms: projection.updated_at_unix_ms,
1107 state: "materialized".to_string(),
1108 source: projection.source.clone(),
1109 node_labels: projection.node_labels.clone(),
1110 node_types: projection.node_types.clone(),
1111 edge_labels: projection.edge_labels.clone(),
1112 last_materialized_sequence: projection.last_materialized_sequence,
1113 })
1114 .collect()
1115 })
1116 .unwrap_or_default()
1117 }
1118
1119 pub(crate) fn analytics_jobs_from_native_state(
1120 &self,
1121 native_state: &NativePhysicalState,
1122 ) -> Vec<PhysicalAnalyticsJob> {
1123 native_state
1124 .registry
1125 .as_ref()
1126 .map(|registry| {
1127 registry
1128 .analytics_jobs
1129 .iter()
1130 .map(|job| PhysicalAnalyticsJob {
1131 id: job.id.clone(),
1132 kind: job.kind.clone(),
1133 state: job.state.clone(),
1134 projection: job.projection.clone(),
1135 created_at_unix_ms: job.created_at_unix_ms,
1136 updated_at_unix_ms: job.updated_at_unix_ms,
1137 last_run_sequence: job.last_run_sequence,
1138 metadata: job.metadata.clone(),
1139 })
1140 .collect()
1141 })
1142 .unwrap_or_default()
1143 }
1144
1145 pub(crate) fn exports_from_native_state(
1146 &self,
1147 native_state: &NativePhysicalState,
1148 ) -> Vec<ExportDescriptor> {
1149 native_state
1150 .recovery
1151 .as_ref()
1152 .map(|recovery| {
1153 recovery
1154 .exports
1155 .iter()
1156 .map(|export| ExportDescriptor {
1157 name: export.name.clone(),
1158 created_at_unix_ms: export.created_at_unix_ms,
1159 snapshot_id: export.snapshot_id,
1160 superblock_sequence: export.superblock_sequence,
1161 data_path: self
1162 .path()
1163 .map(|path| {
1164 crate::physical::PhysicalMetadataFile::export_data_path_for(
1165 path,
1166 &export.name,
1167 )
1168 .display()
1169 .to_string()
1170 })
1171 .unwrap_or_default(),
1172 metadata_path: self
1173 .path()
1174 .map(|path| {
1175 let export_data_path =
1176 crate::physical::PhysicalMetadataFile::export_data_path_for(
1177 path,
1178 &export.name,
1179 );
1180 crate::physical::PhysicalMetadataFile::metadata_path_for(
1181 &export_data_path,
1182 )
1183 .display()
1184 .to_string()
1185 })
1186 .unwrap_or_default(),
1187 collection_count: export.collection_count as usize,
1188 total_entities: export.total_entities as usize,
1189 })
1190 .collect()
1191 })
1192 .unwrap_or_default()
1193 }
1194
1195 pub(crate) fn snapshots_from_native_state(
1196 &self,
1197 native_state: &NativePhysicalState,
1198 ) -> Vec<crate::physical::SnapshotDescriptor> {
1199 let snapshots: Vec<_> = native_state
1200 .recovery
1201 .as_ref()
1202 .map(|recovery| {
1203 recovery
1204 .snapshots
1205 .iter()
1206 .map(|snapshot| crate::physical::SnapshotDescriptor {
1207 snapshot_id: snapshot.snapshot_id,
1208 created_at_unix_ms: snapshot.created_at_unix_ms,
1209 superblock_sequence: snapshot.superblock_sequence,
1210 collection_count: snapshot.collection_count as usize,
1211 total_entities: snapshot.total_entities as usize,
1212 })
1213 .collect()
1214 })
1215 .unwrap_or_default();
1216 if !snapshots.is_empty() {
1217 return snapshots;
1218 }
1219
1220 let now = SystemTime::now()
1221 .duration_since(UNIX_EPOCH)
1222 .unwrap_or_default()
1223 .as_millis();
1224 let (collection_count, total_entities) = native_state
1225 .catalog
1226 .as_ref()
1227 .map(|catalog| {
1228 (
1229 catalog.collection_count as usize,
1230 catalog.total_entities as usize,
1231 )
1232 })
1233 .unwrap_or_else(|| {
1234 let catalog = self.catalog_snapshot();
1235 (catalog.total_collections, catalog.total_entities)
1236 });
1237
1238 vec![crate::physical::SnapshotDescriptor {
1239 snapshot_id: native_state.header.sequence,
1240 created_at_unix_ms: now,
1241 superblock_sequence: native_state.header.sequence,
1242 collection_count,
1243 total_entities,
1244 }]
1245 }
1246
1247 fn index_kind_from_str(value: &str) -> Option<crate::index::IndexKind> {
1248 match value {
1249 "btree" => Some(crate::index::IndexKind::BTree),
1250 "vector.hnsw" => Some(crate::index::IndexKind::VectorHnsw),
1251 "vector.inverted" => Some(crate::index::IndexKind::VectorInverted),
1252 "vector.turbo" => Some(crate::index::IndexKind::VectorTurbo),
1253 "graph.adjacency" => Some(crate::index::IndexKind::GraphAdjacency),
1254 "text.fulltext" => Some(crate::index::IndexKind::FullText),
1255 "document.pathvalue" => Some(crate::index::IndexKind::DocumentPathValue),
1256 "search.hybrid" => Some(crate::index::IndexKind::HybridSearch),
1257 _ => None,
1258 }
1259 }
1260
1261 pub(crate) fn native_artifact_kind_for_index(kind: IndexKind) -> Option<&'static str> {
1262 match kind {
1263 IndexKind::VectorHnsw => Some("hnsw"),
1264 IndexKind::VectorInverted => Some("ivf"),
1265 IndexKind::VectorTurbo => Some("turboquant"),
1266 IndexKind::GraphAdjacency => Some("graph.adjacency"),
1267 IndexKind::FullText => Some("text.fulltext"),
1268 IndexKind::DocumentPathValue => Some("document.pathvalue"),
1269 _ => None,
1270 }
1271 }
1272
1273 fn index_is_declared(&self, name: &str) -> bool {
1274 self.physical_metadata()
1275 .map(|metadata| metadata.indexes.iter().any(|index| index.name == name))
1276 .unwrap_or(false)
1277 }
1278
1279 pub(crate) fn graph_projection_is_declared(&self, name: &str) -> bool {
1280 self.physical_metadata()
1281 .map(|metadata| {
1282 metadata
1283 .graph_projections
1284 .iter()
1285 .any(|projection| projection.name == name)
1286 })
1287 .unwrap_or(false)
1288 }
1289
1290 pub(crate) fn graph_projection_is_operational(&self, name: &str) -> bool {
1291 self.operational_graph_projections()
1292 .into_iter()
1293 .any(|projection| projection.name == name && projection.state == "materialized")
1294 }
1295
1296 pub(crate) fn analytics_job_id(kind: &str, projection: Option<&str>) -> String {
1297 match projection {
1298 Some(projection) => format!("{kind}::{projection}"),
1299 None => format!("{kind}::global"),
1300 }
1301 }
1302
1303 pub(crate) fn update_physical_metadata<T, F>(
1304 &self,
1305 mutator: F,
1306 ) -> Result<T, Box<dyn std::error::Error>>
1307 where
1308 F: FnOnce(&mut PhysicalMetadataFile) -> T,
1309 {
1310 if self.options.mode != StorageMode::Persistent {
1311 return Err("physical metadata operations require persistent mode".into());
1312 }
1313 if self.options.read_only {
1314 return Err("physical metadata operations are not allowed in read-only mode".into());
1315 }
1316 let Some(path) = self.path() else {
1317 return Err("database path is not available".into());
1318 };
1319
1320 let _publication = self.store.vault_publication_lock.lock();
1327 let mut metadata = self.load_or_bootstrap_physical_metadata(true)?;
1328
1329 if metadata.indexes.is_empty() {
1330 metadata.indexes = self.physical_index_state();
1331 }
1332 metadata.superblock.collection_roots = self.physical_collection_roots();
1333
1334 let result = mutator(&mut metadata);
1335 metadata.save_for_data_path(path)?;
1336 self.persist_native_physical_header(&metadata)?;
1337 Ok(result)
1338 }
1339
1340 pub(crate) fn persist_native_physical_header(
1341 &self,
1342 metadata: &PhysicalMetadataFile,
1343 ) -> Result<(), Box<dyn std::error::Error>> {
1344 if !self.paged_mode {
1345 return Ok(());
1346 }
1347
1348 let _publication = self.store.vault_publication_lock.lock();
1349 let existing_page = self
1350 .store
1351 .physical_file_header()
1352 .map(|header| header.collection_roots_page)
1353 .filter(|page| *page != 0);
1354 let existing_registry_page = self
1355 .store
1356 .physical_file_header()
1357 .map(|header| header.registry_page)
1358 .filter(|page| *page != 0);
1359 let existing_recovery_page = self
1360 .store
1361 .physical_file_header()
1362 .map(|header| header.recovery_page)
1363 .filter(|page| *page != 0);
1364 let existing_catalog_page = self
1365 .store
1366 .physical_file_header()
1367 .map(|header| header.catalog_page)
1368 .filter(|page| *page != 0);
1369 let existing_metadata_state_page = self
1370 .store
1371 .physical_file_header()
1372 .map(|header| header.metadata_state_page)
1373 .filter(|page| *page != 0);
1374 let existing_vector_artifact_page = self
1375 .store
1376 .physical_file_header()
1377 .map(|header| header.vector_artifact_page)
1378 .filter(|page| *page != 0);
1379 let existing_manifest_page = self
1380 .store
1381 .physical_file_header()
1382 .map(|header| header.manifest_page)
1383 .filter(|page| *page != 0);
1384 let (manifest_page, manifest_checksum) = self.store.write_native_manifest_summary(
1385 metadata.superblock.sequence,
1386 &metadata.manifest_events,
1387 existing_manifest_page,
1388 )?;
1389 let (collection_roots_page, collection_roots_checksum) = self
1390 .store
1391 .write_native_collection_roots(&metadata.superblock.collection_roots, existing_page)?;
1392 let build = self.native_vector_artifact_records()?;
1395 let vector_artifact_records = build.records;
1396 let mut metadata = metadata.clone();
1397 for index in &mut metadata.indexes {
1398 let Some(kind) = Self::native_artifact_kind_for_index(index.kind) else {
1399 continue;
1400 };
1401 let Some(collection) = &index.collection else {
1402 continue;
1403 };
1404 if build.deferred_collections.contains(collection) {
1405 index.backend = "memory-budget-deferred".to_string();
1406 index.build_state = "deferred-memory-budget".to_string();
1407 index.artifact_kind = Some(kind.to_string());
1408 index.artifact_root_page = None;
1409 index.artifact_checksum = None;
1410 } else if let Some((summary, _)) =
1411 vector_artifact_records.iter().find(|(summary, _)| {
1412 &summary.collection == collection && summary.artifact_kind == kind
1413 })
1414 {
1415 index.backend = format!("{}+native-artifact", index_backend_name(index.kind));
1416 index.build_state = "artifact-published".to_string();
1417 index.artifact_kind = Some(kind.to_string());
1418 index.artifact_checksum = Some(summary.checksum);
1419 }
1420 }
1421 let metadata = &metadata;
1422 let artifact_summaries = vector_artifact_records
1423 .iter()
1424 .map(|(summary, _)| summary.clone())
1425 .collect::<Vec<_>>();
1426 let registry_summary = self.native_registry_summary_from_metadata(
1427 metadata,
1428 &artifact_summaries,
1429 artifact_summaries.len(),
1430 );
1431 let (registry_page, registry_checksum) = self
1432 .store
1433 .write_native_registry_summary(®istry_summary, existing_registry_page)?;
1434 let recovery_summary = Self::native_recovery_summary_from_metadata(metadata);
1435 let (recovery_page, recovery_checksum) = self
1436 .store
1437 .write_native_recovery_summary(&recovery_summary, existing_recovery_page)?;
1438 let catalog_summary = Self::native_catalog_summary_from_metadata(metadata);
1439 let (catalog_page, catalog_checksum) = self
1440 .store
1441 .write_native_catalog_summary(&catalog_summary, existing_catalog_page)?;
1442 let metadata_state_summary = Self::native_metadata_state_summary_from_metadata(metadata);
1443 let (metadata_state_page, metadata_state_checksum) =
1444 self.store.write_native_metadata_state_summary(
1445 &metadata_state_summary,
1446 existing_metadata_state_page,
1447 )?;
1448 let vector_artifact_payloads = vector_artifact_records
1449 .iter()
1450 .map(|(summary, bytes)| {
1451 (
1452 summary.collection.clone(),
1453 summary.artifact_kind.clone(),
1454 bytes.clone(),
1455 )
1456 })
1457 .collect::<Vec<_>>();
1458 let (vector_artifact_page, vector_artifact_checksum, _vector_artifact_pages) =
1459 self.store.write_native_vector_artifact_store(
1460 &vector_artifact_payloads,
1461 existing_vector_artifact_page,
1462 )?;
1463 let mut header = Self::native_header_from_metadata(metadata);
1464 header.manifest_page = manifest_page;
1465 header.manifest_checksum = manifest_checksum;
1466 header.collection_roots_page = collection_roots_page;
1467 header.collection_roots_checksum = collection_roots_checksum;
1468 header.registry_page = registry_page;
1469 header.registry_checksum = registry_checksum;
1470 header.recovery_page = recovery_page;
1471 header.recovery_checksum = recovery_checksum;
1472 header.catalog_page = catalog_page;
1473 header.catalog_checksum = catalog_checksum;
1474 header.metadata_state_page = metadata_state_page;
1475 header.metadata_state_checksum = metadata_state_checksum;
1476 header.vector_artifact_page = vector_artifact_page;
1477 header.vector_artifact_checksum = vector_artifact_checksum;
1478 if let Some(path) = self.path() {
1479 metadata.save_for_data_path(path)?;
1480 }
1481 self.store.update_physical_file_header(header)?;
1482 self.store.persist()?;
1483 self.store.publish_detached_native_blob_pages()?;
1484 Ok(())
1485 }
1486
1487 pub(crate) fn native_header_from_metadata(
1488 metadata: &PhysicalMetadataFile,
1489 ) -> PhysicalFileHeader {
1490 PhysicalFileHeader {
1491 format_version: metadata.superblock.format_version,
1492 sequence: metadata.superblock.sequence,
1493 manifest_oldest_root: metadata.superblock.manifest.oldest.index,
1494 manifest_root: metadata.superblock.manifest.newest.index,
1495 free_set_root: metadata.superblock.free_set.index,
1496 manifest_page: 0,
1497 manifest_checksum: 0,
1498 collection_roots_page: 0,
1499 collection_roots_checksum: 0,
1500 collection_root_count: metadata.superblock.collection_roots.len() as u32,
1501 snapshot_count: metadata.snapshots.len() as u32,
1502 index_count: metadata.indexes.len() as u32,
1503 catalog_collection_count: metadata.catalog.total_collections as u32,
1504 catalog_total_entities: metadata.catalog.total_entities as u64,
1505 export_count: metadata.exports.len() as u32,
1506 graph_projection_count: metadata.graph_projections.len() as u32,
1507 analytics_job_count: metadata.analytics_jobs.len() as u32,
1508 manifest_event_count: metadata.manifest_events.len() as u32,
1509 registry_page: 0,
1510 registry_checksum: 0,
1511 recovery_page: 0,
1512 recovery_checksum: 0,
1513 catalog_page: 0,
1514 catalog_checksum: 0,
1515 metadata_state_page: 0,
1516 metadata_state_checksum: 0,
1517 vector_artifact_page: 0,
1518 vector_artifact_checksum: 0,
1519 }
1520 }
1521}
1522
1523fn current_unix_ms() -> u64 {
1524 SystemTime::now()
1525 .duration_since(UNIX_EPOCH)
1526 .unwrap_or_default()
1527 .as_millis()
1528 .min(u128::from(u64::MAX)) as u64
1529}