Skip to main content

reddb_server/storage/unified/devx/reddb/
impl_metadata.rs

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        // Export pruning is only meaningful for persistent mode where we
18        // have a metadata sidecar that tracks file-backed export artifacts.
19        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    // ========================================================================
232    // Builder Methods - Create Entities
233    // ========================================================================
234
235    /// Start building a graph node
236    ///
237    /// # Example
238    /// ```ignore
239    /// let host = db.node("hosts", "Host")
240    ///     .property("ip", "192.168.1.1")
241    ///     .save()?;
242    /// ```
243    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    /// Start building a graph edge
253    ///
254    /// # Example
255    /// ```ignore
256    /// let edge = db.edge("connections", "CONNECTS_TO")
257    ///     .from(host_a)
258    ///     .to(host_b)
259    ///     .weight(0.95)
260    ///     .property("protocol", "TCP")
261    ///     .save()?;
262    /// ```
263    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    /// Start building a vector entry
273    ///
274    /// # Example
275    /// ```ignore
276    /// let vec = db.vector("embeddings")
277    ///     .dense(embedding)
278    ///     .content("Original text content")
279    ///     .metadata("source", "document.pdf")
280    ///     .save()?;
281    /// ```
282    pub fn vector(&self, collection: impl Into<String>) -> VectorBuilder {
283        VectorBuilder::new(self.store.clone(), self.preprocessors.clone(), collection)
284    }
285
286    /// Start building a table row
287    ///
288    /// # Example
289    /// ```ignore
290    /// let row = db.row("scans", vec![
291    ///     ("timestamp", Value::Timestamp(now)),
292    ///     ("target", Value::text("192.168.1.0/24")),
293    ///     ("findings", Value::Integer(42)),
294    /// ]).save()?;
295    /// ```
296    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    /// Start building a document
306    ///
307    /// Documents are stored as enriched table rows with a full JSON body
308    /// field and flattened top-level keys for filtering.
309    ///
310    /// # Example
311    /// ```ignore
312    /// let doc = db.doc("articles")
313    ///     .field("title", "Hello World")
314    ///     .field("views", 42)
315    ///     .metadata("source", "web")
316    ///     .save()?;
317    /// ```
318    pub fn doc(&self, collection: impl Into<String>) -> DocumentBuilder {
319        DocumentBuilder::new(self.store.clone(), self.preprocessors.clone(), collection)
320    }
321
322    /// Start building a key-value pair
323    ///
324    /// KV pairs are stored as table rows with named fields `key` and `value`.
325    ///
326    /// # Example
327    /// ```ignore
328    /// let id = db.kv("config", "theme", Value::text("dark"))
329    ///     .metadata("updated_by", "admin")
330    ///     .save()?;
331    /// ```
332    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    /// Get a key-value pair by key, returning the value and entity id
348    ///
349    /// Clone only the selected value. Legacy positional rows without an
350    /// indexed key use their collection schema in the borrowed fallback scan.
351    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    /// Delete a key-value pair by key, returning whether it was found and removed
403    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            // Load metadata without persisting (avoids blocking catalog snapshot on boot)
420            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            // Skip repair on boot — deferred to first explicit persist_metadata() call.
435            // This avoids the recursive catalog_model_snapshot → physical_metadata loop
436            // that caused stack overflow / 12-second hang on startup.
437        }
438        self.load_collection_ttl_defaults_from_metadata();
439        // Single-file artifact: collection contracts ride the store's binary
440        // dump, not the metadata sidecar (which is never written in this
441        // profile). Seed the contract cache from that blob so each
442        // collection's declared model survives a restart — otherwise recovery
443        // re-infers the model from the stored entities and a KV collection
444        // comes back as a table.
445        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        // Issue #866 — rehydrate the hypertable chunk spine before the
453        // API opens so chunk routing / pruning / TTL work immediately
454        // after a restart.
455        self.load_hypertables_from_metadata();
456        // Pending deliveries retain their persisted lock deadlines on recovery
457        // (ADR 0020). The lifecycle lazily reclaims expired locks on delivery;
458        // deleting every pending row here would redeliver still-owned work.
459        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        // Default embedded options can reopen an existing paged file. Its
467        // native headers and index artifacts still require publication.
468        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        // Issue #866 — persist the hypertable chunk spine (specs +
496        // chunk bounds / routing / TTL) onto the same metadata path as
497        // collection contracts so a restart recovers it identically.
498        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    /// Rebuild the external physical metadata view from the native state published in the
527    /// paged database file.
528    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                // Accept the bootstrap when the native state is either
603                // (a) fully populated and consistent (the original
604                // contract), or (b) trivially empty — a freshly created
605                // database with no collections written yet. Without (b)
606                // a brand-new data file can never reach
607                // `readiness_for_query = true`, because the bootstrap
608                // refuses to run until the registry/catalog/recovery
609                // structures are "complete", which they never become
610                // until the bootstrap has already run once.
611                //
612                // The emptiness check is conservative: header.sequence
613                // must still be at its initial value AND all three
614                // physical state summaries must be absent. Anything
615                // else falls through to the original error so we never
616                // paper over partially corrupted files.
617                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 &registry.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        // #2376: the metadata sidecar and the native header pages are
1321        // published read-modify-write. Two unserialised publishers lost each
1322        // other's collection contracts and double-freed the header pages
1323        // (a freed page came back as a B-tree page). Every publisher takes
1324        // the store's reentrant publication lock, which `persist()` already
1325        // takes, so this adds no lock-order edge.
1326        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        // Build payload artifacts once, on the publication path. Observers use
1393        // the persisted summaries instead of rebuilding search indexes.
1394        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(&registry_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}