Skip to main content

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

1use super::*;
2
3impl RedDB {
4    pub(crate) fn write_column_block_page(
5        &self,
6        bytes: &[u8],
7    ) -> Result<crate::storage::engine::PageLocation, Box<dyn std::error::Error>> {
8        use crate::storage::engine::PageType;
9
10        let store = self.store();
11        let Some(pager) = store.pager() else {
12            return Ok(crate::storage::engine::PageLocation::new(
13                0,
14                0,
15                bytes.len() as u32,
16            ));
17        };
18        let content_size = crate::storage::engine::page::CONTENT_SIZE;
19        let page_count = bytes.len().max(1).div_ceil(content_size);
20        let mut first_page_id = None;
21        for page_index in 0..page_count {
22            let mut page = pager.allocate_page(PageType::ColumnBlock)?;
23            let page_id = page.page_id();
24            if let Some(first) = first_page_id {
25                let expected = first + page_index as u32;
26                if page_id != expected {
27                    return Err(format!(
28                        "column block page allocation was not contiguous: got {page_id}, expected {expected}"
29                    )
30                    .into());
31                }
32            } else {
33                first_page_id = Some(page_id);
34            }
35            let start = page_index * content_size;
36            let end = (start + content_size).min(bytes.len());
37            if start < end {
38                page.content_mut()[..end - start].copy_from_slice(&bytes[start..end]);
39            }
40            pager.write_page_encrypted(page_id, page)?;
41        }
42        Ok(crate::storage::engine::PageLocation::new(
43            first_page_id.expect("page_count is at least one"),
44            0,
45            bytes.len() as u32,
46        ))
47    }
48
49    pub(crate) fn read_column_block_page(
50        &self,
51        loc: crate::storage::engine::PageLocation,
52    ) -> Result<Option<Vec<u8>>, Box<dyn std::error::Error>> {
53        let store = self.store();
54        let Some(pager) = store.pager() else {
55            return Ok(None);
56        };
57        let content_size = crate::storage::engine::page::CONTENT_SIZE;
58        let mut remaining = loc.length as usize;
59        let mut page_id = loc.page_id;
60        let mut offset = loc.offset as usize;
61        let mut out = Vec::with_capacity(remaining);
62        while remaining > 0 {
63            let page = pager.read_page_decrypted(page_id)?;
64            if page.page_type()? != crate::storage::engine::PageType::ColumnBlock {
65                return Err(format!("page {page_id} is not a ColumnBlock page").into());
66            }
67            if offset > content_size {
68                return Err(format!(
69                    "column block page location offset out of bounds: {offset} > {content_size}"
70                )
71                .into());
72            }
73            let available = content_size - offset;
74            let take = remaining.min(available);
75            out.extend_from_slice(&page.content()[offset..offset + take]);
76            remaining -= take;
77            page_id = page_id.saturating_add(1);
78            offset = 0;
79        }
80        Ok(Some(out))
81    }
82
83    pub fn tree_definitions(&self) -> Vec<crate::physical::PhysicalTreeDefinition> {
84        self.physical_metadata()
85            .map(|metadata| metadata.tree_definitions)
86            .unwrap_or_default()
87    }
88
89    pub fn tree_definition(
90        &self,
91        collection: &str,
92        name: &str,
93    ) -> Option<crate::physical::PhysicalTreeDefinition> {
94        self.tree_definitions()
95            .into_iter()
96            .find(|definition| definition.collection == collection && definition.name == name)
97    }
98
99    pub fn save_tree_definition(
100        &self,
101        definition: crate::physical::PhysicalTreeDefinition,
102    ) -> Result<crate::physical::PhysicalTreeDefinition, Box<dyn std::error::Error>> {
103        self.update_physical_metadata(|metadata| {
104            if let Some(existing) = metadata.tree_definitions.iter_mut().find(|existing| {
105                existing.collection == definition.collection && existing.name == definition.name
106            }) {
107                *existing = definition.clone();
108            } else {
109                metadata.tree_definitions.push(definition.clone());
110            }
111            metadata.tree_definitions.sort_by(|left, right| {
112                left.collection
113                    .cmp(&right.collection)
114                    .then_with(|| left.name.cmp(&right.name))
115            });
116            definition.clone()
117        })
118    }
119
120    pub fn remove_tree_definition(
121        &self,
122        collection: &str,
123        name: &str,
124    ) -> Result<Option<crate::physical::PhysicalTreeDefinition>, Box<dyn std::error::Error>> {
125        self.update_physical_metadata(|metadata| {
126            metadata
127                .tree_definitions
128                .iter()
129                .position(|definition| {
130                    definition.collection == collection && definition.name == name
131                })
132                .map(|index| metadata.tree_definitions.remove(index))
133        })
134    }
135
136    pub fn collection_default_ttl_ms(&self, collection: &str) -> Option<u64> {
137        self.collection_ttl_defaults_ms
138            .read()
139            .ok()
140            .and_then(|defaults| defaults.get(collection).copied())
141    }
142
143    pub fn set_collection_default_ttl_ms(&self, collection: impl Into<String>, ttl_ms: u64) {
144        if let Ok(mut defaults) = self.collection_ttl_defaults_ms.write() {
145            defaults.insert(collection.into(), ttl_ms);
146        }
147    }
148
149    pub fn clear_collection_default_ttl_ms(&self, collection: &str) {
150        if let Ok(mut defaults) = self.collection_ttl_defaults_ms.write() {
151            defaults.remove(collection);
152        }
153    }
154
155    pub fn collection_contracts(&self) -> Vec<crate::physical::CollectionContract> {
156        self.contract_cache_map()
157            .values()
158            .map(|arc| (**arc).clone())
159            .collect()
160    }
161
162    pub fn collection_contract(
163        &self,
164        collection: &str,
165    ) -> Option<crate::physical::CollectionContract> {
166        self.contract_cache_map()
167            .get(collection)
168            .map(|arc| (**arc).clone())
169    }
170
171    /// Arc-bumping variant — callers that only need read access skip
172    /// the per-call full `CollectionContract` clone. Deep clones cost
173    /// ~200-400 ns each (vec of declared columns + string allocs);
174    /// the UPDATE fast path hits this 2-3× per mutation.
175    pub fn collection_contract_arc(
176        &self,
177        collection: &str,
178    ) -> Option<std::sync::Arc<crate::physical::CollectionContract>> {
179        self.contract_cache_map()
180            .get(collection)
181            .map(std::sync::Arc::clone)
182    }
183
184    /// Return (or lazily build) the in-memory contract map.
185    /// Reparses PhysicalMetadataFile JSON only once per invalidation.
186    /// Called 3× per insert (ensure_model, enforce_uniqueness, normalize_fields)
187    /// so the cache is load-bearing for insert throughput.
188    fn contract_cache_map(
189        &self,
190    ) -> std::sync::Arc<
191        std::collections::HashMap<String, std::sync::Arc<crate::physical::CollectionContract>>,
192    > {
193        self.collection_contract_cache.get_or_load(|| {
194            self.physical_metadata()
195                .map(|metadata| metadata.collection_contracts)
196                .unwrap_or_default()
197        })
198    }
199
200    pub(crate) fn invalidate_collection_contract_cache(&self) {
201        self.collection_contract_cache.invalidate();
202    }
203
204    /// Single-file artifact only: mirror the current contract set into the
205    /// store's auxiliary-metadata blob so it rides the binary dump. Without
206    /// this the contracts live only in the in-memory cache and a collection's
207    /// declared model (e.g. `kv`) is lost on reopen — recovery would re-infer
208    /// it from the stored entities and a KV collection would come back as a
209    /// table. No-op for other profiles, which persist contracts via the
210    /// metadata sidecar.
211    pub(crate) fn sync_single_file_contract_blob(&self) -> Result<(), Box<dyn std::error::Error>> {
212        let contracts = self.collection_contracts();
213        let bytes = crate::physical::serialize_collection_contracts(&contracts);
214        self.store().persist_aux_metadata(bytes)?;
215        Ok(())
216    }
217
218    /// Seed the in-memory contract cache from the store's auxiliary-metadata
219    /// blob (single-file artifact). Called on open so a collection's declared
220    /// model survives a restart. Does nothing when the blob is absent (older
221    /// V9 dumps) or empty.
222    pub(crate) fn seed_contract_cache_from_store_aux(&self) -> std::io::Result<()> {
223        let bytes = self.store().aux_metadata();
224        if bytes.is_empty() {
225            return Ok(());
226        }
227        let contracts = crate::physical::deserialize_collection_contracts(&bytes)?;
228        let map: std::collections::HashMap<_, _> = contracts
229            .into_iter()
230            .map(|contract| (contract.name.clone(), std::sync::Arc::new(contract)))
231            .collect();
232        // Restore the per-collection state derived from contracts (context
233        // index enablement + default TTLs) so the full collection definition —
234        // not just the declared model — survives a restart.
235        let store = self.store();
236        let index = store.context_index();
237        if let Ok(mut defaults) = self.collection_ttl_defaults_ms.write() {
238            for contract in map.values() {
239                index.set_collection_enabled(&contract.name, contract.context_index_enabled);
240                if let Some(ttl_ms) = contract.default_ttl_ms {
241                    defaults.insert(contract.name.clone(), ttl_ms);
242                }
243            }
244        }
245        self.collection_contract_cache.replace(map);
246        Ok(())
247    }
248
249    fn update_collection_contract_metadata<T, F>(
250        &self,
251        mutator: F,
252    ) -> Result<T, Box<dyn std::error::Error>>
253    where
254        F: FnOnce(&mut PhysicalMetadataFile) -> T,
255    {
256        if !self.paged_mode
257            || self.options.storage_profile.deploy_profile
258                != crate::storage::DeployProfile::Embedded
259            || self.options.storage_profile.packaging
260                != crate::storage::StoragePackaging::SingleFile
261        {
262            return self.update_physical_metadata(mutator);
263        }
264        if self.options.read_only {
265            return Err("physical metadata operations are not allowed in read-only mode".into());
266        }
267        let Some(_) = self.path() else {
268            return Err("database path is not available".into());
269        };
270
271        // Paged contracts live in an internal record protected by the WAL.
272        // Native artifact summaries are not a complete contract representation.
273        // A definition edit must not rebuild every collection's search indexes.
274        // Retain publication serialization and the canonical atomic writer.
275        let _publication = self.store.vault_publication_lock.lock();
276        let mut metadata = self.load_or_bootstrap_physical_metadata(false)?;
277        let result = mutator(&mut metadata);
278        self.persist_physical_metadata_record(&metadata)?;
279        Ok(result)
280    }
281
282    /// Supply missing system contracts during read-only boot without creating
283    /// a metadata sidecar. Runtime schema rehydration consumes this private view.
284    pub(crate) fn cache_read_only_bootstrap_contract(
285        &self,
286        contract: crate::physical::CollectionContract,
287    ) {
288        assert!(self.options.read_only);
289        self.contract_cache_map();
290        self.collection_contract_cache.insert(contract);
291    }
292
293    pub fn save_collection_contract(
294        &self,
295        contract: crate::physical::CollectionContract,
296    ) -> Result<crate::physical::CollectionContract, Box<dyn std::error::Error>> {
297        let _publication = self.store.vault_publication_lock.lock();
298        if let Ok(mut defaults) = self.collection_ttl_defaults_ms.write() {
299            if let Some(ttl_ms) = contract.default_ttl_ms {
300                defaults.insert(contract.name.clone(), ttl_ms);
301            } else {
302                defaults.remove(&contract.name);
303            }
304        }
305
306        self.store()
307            .context_index()
308            .set_collection_enabled(&contract.name, contract.context_index_enabled);
309        if let Some(manager) = self.store().get_collection(&contract.name) {
310            let columns = contract
311                .declared_columns
312                .iter()
313                .map(|column| column.name.clone())
314                .collect();
315            manager.set_column_schema_if_empty(columns);
316        }
317
318        // Existing paged files can be opened with embedded options. Their
319        // checkpoints do not encode the binary dump's auxiliary contract blob.
320        if self.options.storage_profile.deploy_profile == crate::storage::DeployProfile::Embedded
321            && self.options.storage_profile.packaging
322                == crate::storage::StoragePackaging::SingleFile
323            && !self.paged_mode
324        {
325            self.collection_contract_cache.insert(contract.clone());
326            self.sync_single_file_contract_blob()?;
327            return Ok(contract);
328        }
329
330        self.invalidate_collection_contract_cache();
331
332        let saved = self.update_collection_contract_metadata(|metadata| {
333            if let Some(existing) = metadata
334                .collection_contracts
335                .iter_mut()
336                .find(|existing| existing.name == contract.name)
337            {
338                *existing = contract.clone();
339            } else {
340                metadata.collection_contracts.push(contract.clone());
341            }
342            metadata
343                .collection_contracts
344                .sort_by(|left, right| left.name.cmp(&right.name));
345
346            if let Some(ttl_ms) = contract.default_ttl_ms {
347                metadata
348                    .collection_ttl_defaults_ms
349                    .insert(contract.name.clone(), ttl_ms);
350            } else {
351                metadata.collection_ttl_defaults_ms.remove(&contract.name);
352            }
353
354            contract.clone()
355        })?;
356        self.invalidate_collection_contract_cache();
357        Ok(saved)
358    }
359
360    pub fn remove_collection_contract(
361        &self,
362        collection: &str,
363    ) -> Result<Option<crate::physical::CollectionContract>, Box<dyn std::error::Error>> {
364        if self.options.read_only {
365            return Err("physical metadata operations are not allowed in read-only mode".into());
366        }
367        let _publication = self.store.vault_publication_lock.lock();
368        if let Ok(mut defaults) = self.collection_ttl_defaults_ms.write() {
369            defaults.remove(collection);
370        }
371
372        self.store()
373            .context_index()
374            .set_collection_enabled(collection, false);
375
376        if self.options.storage_profile.deploy_profile == crate::storage::DeployProfile::Embedded
377            && self.options.storage_profile.packaging
378                == crate::storage::StoragePackaging::SingleFile
379            && !self.paged_mode
380        {
381            let removed = self.collection_contract_cache.remove(collection);
382            self.sync_single_file_contract_blob()?;
383            return Ok(removed);
384        }
385
386        self.invalidate_collection_contract_cache();
387
388        if self.paged_mode {
389            let Some(_) = self.path() else {
390                return Err("database path is not available".into());
391            };
392            // DROP already published the native collection registry before it
393            // reaches this contract edit.  Rebuild only the small physical
394            // metadata catalogue from its last durable roots, removing the
395            // dropped root and descriptors explicitly.  Calling the generic
396            // metadata updater here used to fingerprint every row in changed
397            // internal collections (notably append-heavy red_config) while
398            // holding the publication lock, turning a tiny DROP after cold
399            // boot into work proportional to the database's history.
400            //
401            // Other collection fingerprints describe their last checkpoint,
402            // just as they do between ordinary DML and the next checkpoint;
403            // the canonical checkpoint path remains responsible for
404            // recomputing changed fingerprints.
405            let previous = self.load_or_bootstrap_physical_metadata(false)?;
406            let mut roots = previous.superblock.collection_roots.clone();
407            roots.remove(collection);
408            let indexes = previous
409                .indexes
410                .iter()
411                .filter(|index| index.collection.as_deref() != Some(collection))
412                .cloned()
413                .collect();
414            let mut metadata = PhysicalMetadataFile::from_state(
415                self.options.clone(),
416                self.catalog_snapshot(),
417                roots,
418                indexes,
419                Some(&previous),
420            );
421            let removed = metadata
422                .collection_contracts
423                .iter()
424                .position(|contract| contract.name == collection)
425                .map(|index| metadata.collection_contracts.remove(index));
426            metadata.collection_ttl_defaults_ms.remove(collection);
427            self.persist_physical_metadata_record(&metadata)?;
428            self.persist_native_physical_header(&metadata)?;
429            self.invalidate_collection_contract_cache();
430            return Ok(removed);
431        }
432
433        let removed = self.update_collection_contract_metadata(|metadata| {
434            let removed = metadata
435                .collection_contracts
436                .iter()
437                .position(|contract| contract.name == collection)
438                .map(|index| metadata.collection_contracts.remove(index));
439            metadata.collection_ttl_defaults_ms.remove(collection);
440            metadata
441                .indexes
442                .retain(|index| index.collection.as_deref() != Some(collection));
443            removed
444        })?;
445        self.invalidate_collection_contract_cache();
446        Ok(removed)
447    }
448
449    pub(crate) fn collection_ttl_defaults_snapshot(&self) -> BTreeMap<String, u64> {
450        self.collection_ttl_defaults_ms
451            .read()
452            .map(|defaults| {
453                defaults
454                    .iter()
455                    .map(|(collection, ttl_ms)| (collection.clone(), *ttl_ms))
456                    .collect()
457            })
458            .unwrap_or_default()
459    }
460
461    pub(crate) fn load_collection_ttl_defaults_from_metadata(&self) {
462        let metadata = self.physical_metadata();
463        let defaults = metadata
464            .as_ref()
465            .map(|m| m.collection_ttl_defaults_ms.clone())
466            .unwrap_or_default();
467
468        if let Ok(mut current) = self.collection_ttl_defaults_ms.write() {
469            current.clear();
470            current.extend(defaults);
471        }
472
473        if let Some(metadata) = metadata {
474            let store = self.store();
475            let index = store.context_index();
476            for contract in &metadata.collection_contracts {
477                index.set_collection_enabled(&contract.name, contract.context_index_enabled);
478            }
479        }
480    }
481
482    /// Snapshot the live hypertable registry (specs + every chunk)
483    /// into the sidecar-serialisable form. Called from
484    /// [`persist_metadata`] so the chunk routing spine is durable on
485    /// the same metadata path as collection contracts — issue #866.
486    /// Returns an empty vec (and touches nothing) when no hypertable
487    /// has ever been declared, so non-hypertable databases pay no
488    /// persist cost.
489    pub(crate) fn hypertable_registry_snapshot(&self) -> Vec<crate::physical::PhysicalHypertable> {
490        let registry = self.hypertables();
491        if registry.is_empty() {
492            return Vec::new();
493        }
494        use std::collections::BTreeMap;
495        let mut chunks_by_table: BTreeMap<String, Vec<crate::physical::PhysicalHypertableChunk>> =
496            BTreeMap::new();
497        for chunk in registry.snapshot_chunks() {
498            chunks_by_table
499                .entry(chunk.id.hypertable.clone())
500                .or_default()
501                .push(crate::physical::PhysicalHypertableChunk {
502                    start_ns: chunk.id.start_ns,
503                    end_ns_exclusive: chunk.end_ns_exclusive,
504                    row_count: chunk.row_count,
505                    min_ts_ns: chunk.min_ts_ns,
506                    max_ts_ns: chunk.max_ts_ns,
507                    sealed: chunk.sealed,
508                    ttl_override_ns: chunk.ttl_override_ns,
509                    columnar_page: chunk.columnar_page,
510                    columnar_derived: chunk.columnar_page.is_some(),
511                });
512        }
513        registry
514            .list()
515            .into_iter()
516            .map(|spec| crate::physical::PhysicalHypertable {
517                chunks: chunks_by_table.remove(&spec.name).unwrap_or_default(),
518                name: spec.name,
519                time_column: spec.time_column,
520                chunk_interval_ns: spec.chunk_interval_ns,
521                default_ttl_ns: spec.default_ttl_ns,
522            })
523            .collect()
524    }
525
526    /// Rehydrate the hypertable registry from the physical metadata
527    /// sidecar at boot. Mirror of
528    /// [`load_collection_ttl_defaults_from_metadata`] — specs and
529    /// chunks are restored verbatim so bounds / routing / TTL are
530    /// identical to the pre-restart state (issue #866).
531    pub(crate) fn load_hypertables_from_metadata(&self) {
532        let Some(metadata) = self.physical_metadata() else {
533            return;
534        };
535        if metadata.hypertables.is_empty() {
536            return;
537        }
538        let registry = self.hypertables();
539        for hypertable in &metadata.hypertables {
540            let mut spec = crate::storage::timeseries::HypertableSpec::new(
541                hypertable.name.clone(),
542                hypertable.time_column.clone(),
543                hypertable.chunk_interval_ns,
544            );
545            if let Some(ttl) = hypertable.default_ttl_ns {
546                spec = spec.with_ttl_ns(ttl);
547            }
548            registry.register(spec);
549            for chunk in &hypertable.chunks {
550                let meta = crate::storage::timeseries::ChunkMeta {
551                    id: crate::storage::timeseries::ChunkId {
552                        hypertable: hypertable.name.clone(),
553                        start_ns: chunk.start_ns,
554                    },
555                    end_ns_exclusive: chunk.end_ns_exclusive,
556                    row_count: chunk.row_count,
557                    min_ts_ns: chunk.min_ts_ns,
558                    max_ts_ns: chunk.max_ts_ns,
559                    sealed: chunk.sealed,
560                    ttl_override_ns: chunk.ttl_override_ns,
561                    columnar_page: chunk.columnar_page,
562                };
563                let id = meta.id.clone();
564                let columnar_page = meta.columnar_page;
565                registry.restore_chunk(meta);
566                if let Some(loc) = columnar_page {
567                    if let Ok(Some(bytes)) = self.read_column_block_page(loc) {
568                        registry.restore_columnar_block(&id, bytes);
569                    }
570                }
571            }
572        }
573    }
574
575    pub fn run_maintenance(&self) -> Result<(), Box<dyn std::error::Error>> {
576        self.store.run_maintenance()?;
577        self.persist_metadata()?;
578        if self.paged_mode {
579            if let Some(path) = self.path() {
580                self.remove_imported_metadata_sidecars(path)?;
581            }
582        }
583        Ok(())
584    }
585
586    /// Path to the physical metadata sidecar, if persistent.
587    pub fn metadata_path(&self) -> Option<PathBuf> {
588        self.path
589            .as_ref()
590            .map(|path| PhysicalMetadataFile::metadata_path_for(path))
591    }
592
593    /// Inspect the current physical metadata view without publishing repairs.
594    pub fn physical_metadata(&self) -> Option<PhysicalMetadataFile> {
595        self.load_or_bootstrap_physical_metadata(false).ok()
596    }
597
598    /// Physical index registry derived for the current database state.
599    pub fn physical_indexes(&self) -> Vec<PhysicalIndexState> {
600        let indexes = self
601            .physical_metadata()
602            .map(|metadata| metadata.indexes)
603            .unwrap_or_default();
604        self.physical_indexes_from_declarations(indexes)
605    }
606
607    pub(crate) fn physical_indexes_from_declarations(
608        &self,
609        indexes: Vec<PhysicalIndexState>,
610    ) -> Vec<PhysicalIndexState> {
611        let indexes = Some(indexes)
612            .filter(|indexes| !indexes.is_empty())
613            .or_else(|| {
614                self.native_physical_state()
615                    .map(|state| self.physical_index_state_from_native_state(&state, None))
616            })
617            .unwrap_or_else(|| self.physical_index_state());
618        self.reconcile_index_states_with_native_artifacts(indexes)
619    }
620
621    /// List registered named exports from the current physical metadata view.
622    pub fn exports(&self) -> Vec<ExportDescriptor> {
623        self.physical_metadata()
624            .map(|metadata| metadata.exports)
625            .or_else(|| {
626                self.native_physical_state()
627                    .map(|state| self.exports_from_native_state(&state))
628            })
629            .unwrap_or_default()
630    }
631
632    /// List recorded snapshots from the current physical metadata view.
633    pub fn snapshots(&self) -> Vec<crate::physical::SnapshotDescriptor> {
634        if let Some(metadata) = self.physical_metadata() {
635            if !metadata.snapshots.is_empty() {
636                return metadata.snapshots;
637            }
638        }
639        self.native_physical_state()
640            .map(|state| self.snapshots_from_native_state(&state))
641            .unwrap_or_default()
642    }
643
644    /// List persisted named graph projections from the current physical metadata view.
645    pub fn graph_projections(&self) -> Vec<PhysicalGraphProjection> {
646        self.physical_metadata()
647            .map(|metadata| metadata.graph_projections)
648            .or_else(|| {
649                self.native_physical_state()
650                    .map(|state| self.graph_projections_from_native_state(&state))
651            })
652            .unwrap_or_default()
653    }
654
655    /// List graph projections declared in the catalog view.
656    pub fn declared_graph_projections(&self) -> Vec<PhysicalGraphProjection> {
657        self.catalog_model_snapshot().declared_graph_projections
658    }
659
660    /// List graph projections currently observed in the operational view.
661    pub fn operational_graph_projections(&self) -> Vec<PhysicalGraphProjection> {
662        self.graph_projections()
663            .into_iter()
664            .filter(|projection| {
665                projection.last_materialized_sequence.is_some()
666                    || matches!(projection.state.as_str(), "materialized" | "stale")
667            })
668            .collect()
669    }
670
671    /// List persisted analytics job metadata from the current physical metadata view.
672    pub fn analytics_jobs(&self) -> Vec<PhysicalAnalyticsJob> {
673        self.physical_metadata()
674            .map(|metadata| metadata.analytics_jobs)
675            .or_else(|| {
676                self.native_physical_state()
677                    .map(|state| self.analytics_jobs_from_native_state(&state))
678            })
679            .unwrap_or_default()
680    }
681
682    /// List analytics jobs declared in the catalog view.
683    pub fn declared_analytics_jobs(&self) -> Vec<PhysicalAnalyticsJob> {
684        self.catalog_model_snapshot().declared_analytics_jobs
685    }
686
687    /// List analytics jobs currently observed in the operational view.
688    pub fn operational_analytics_jobs(&self) -> Vec<PhysicalAnalyticsJob> {
689        self.analytics_jobs()
690            .into_iter()
691            .filter(|job| {
692                job.last_run_sequence.is_some()
693                    || matches!(
694                        job.state.as_str(),
695                        "running" | "completed" | "failed" | "queued" | "stale"
696                    )
697            })
698            .collect()
699    }
700
701    /// List indexes declared in the catalog view.
702    pub fn declared_indexes(&self) -> Vec<PhysicalIndexState> {
703        self.catalog_model_snapshot().declared_indexes
704    }
705
706    /// List indexes currently observed in the operational view.
707    pub fn operational_indexes(&self) -> Vec<PhysicalIndexState> {
708        self.catalog_model_snapshot().operational_indexes
709    }
710
711    /// List reconciled index status entries from the catalog snapshot.
712    pub fn index_statuses(&self) -> Vec<CatalogIndexStatus> {
713        self.catalog_model_snapshot().index_statuses
714    }
715
716    /// Resolve one index status entry from the catalog snapshot.
717    pub fn index_status(&self, name: &str) -> Option<CatalogIndexStatus> {
718        self.catalog_model_snapshot()
719            .index_statuses
720            .into_iter()
721            .find(|status| status.name == name)
722    }
723
724    /// Upsert a named graph projection in the persisted physical metadata.
725    pub fn save_graph_projection(
726        &self,
727        name: impl Into<String>,
728        node_labels: Vec<String>,
729        node_types: Vec<String>,
730        edge_labels: Vec<String>,
731        source: impl Into<String>,
732    ) -> Result<PhysicalGraphProjection, Box<dyn std::error::Error>> {
733        let name = name.into();
734        let source = source.into();
735        self.update_physical_metadata(|metadata| {
736            let now = SystemTime::now()
737                .duration_since(UNIX_EPOCH)
738                .unwrap_or_default()
739                .as_millis();
740            let projection = if let Some(existing) = metadata
741                .graph_projections
742                .iter_mut()
743                .find(|projection| projection.name == name)
744            {
745                existing.updated_at_unix_ms = now;
746                existing.state = "declared".to_string();
747                existing.source = source.clone();
748                existing.node_labels = node_labels.clone();
749                existing.node_types = node_types.clone();
750                existing.edge_labels = edge_labels.clone();
751                existing.last_materialized_sequence = None;
752                existing.clone()
753            } else {
754                let projection = PhysicalGraphProjection {
755                    name: name.clone(),
756                    created_at_unix_ms: now,
757                    updated_at_unix_ms: now,
758                    state: "declared".to_string(),
759                    source: source.clone(),
760                    node_labels: node_labels.clone(),
761                    node_types: node_types.clone(),
762                    edge_labels: edge_labels.clone(),
763                    last_materialized_sequence: None,
764                };
765                metadata.graph_projections.push(projection.clone());
766                projection
767            };
768
769            Self::mark_projection_dependent_jobs_stale(metadata, &name, now);
770
771            metadata
772                .graph_projections
773                .sort_by(|left, right| left.name.cmp(&right.name));
774            projection
775        })
776    }
777
778    /// Mark a declared graph projection as materialized in the current physical metadata.
779    pub fn materialize_graph_projection(
780        &self,
781        name: &str,
782    ) -> Result<Option<PhysicalGraphProjection>, Box<dyn std::error::Error>> {
783        self.update_physical_metadata(|metadata| {
784            let now = SystemTime::now()
785                .duration_since(UNIX_EPOCH)
786                .unwrap_or_default()
787                .as_millis();
788            let idx = metadata
789                .graph_projections
790                .iter()
791                .position(|projection| projection.name == name);
792            if let Some(idx) = idx {
793                metadata.graph_projections[idx].updated_at_unix_ms = now;
794                metadata.graph_projections[idx].state = "materialized".to_string();
795                metadata.graph_projections[idx].last_materialized_sequence =
796                    Some(metadata.superblock.sequence);
797                let result = metadata.graph_projections[idx].clone();
798                Self::rearm_projection_dependent_jobs_declared(metadata, name, now);
799                return Some(result);
800            }
801            None
802        })
803    }
804
805    /// Mark a declared graph projection as materializing.
806    pub fn mark_graph_projection_materializing(
807        &self,
808        name: &str,
809    ) -> Result<Option<PhysicalGraphProjection>, Box<dyn std::error::Error>> {
810        self.update_physical_metadata(|metadata| {
811            let now = SystemTime::now()
812                .duration_since(UNIX_EPOCH)
813                .unwrap_or_default()
814                .as_millis();
815            let idx = metadata
816                .graph_projections
817                .iter()
818                .position(|projection| projection.name == name);
819            if let Some(idx) = idx {
820                metadata.graph_projections[idx].updated_at_unix_ms = now;
821                metadata.graph_projections[idx].state = "materializing".to_string();
822                metadata.graph_projections[idx].last_materialized_sequence = None;
823                let result = metadata.graph_projections[idx].clone();
824                Self::mark_projection_dependent_jobs_stale(metadata, name, now);
825                return Some(result);
826            }
827            None
828        })
829    }
830
831    /// Mark a graph projection as failed.
832    pub fn fail_graph_projection(
833        &self,
834        name: &str,
835    ) -> Result<Option<PhysicalGraphProjection>, Box<dyn std::error::Error>> {
836        self.update_physical_metadata(|metadata| {
837            let now = SystemTime::now()
838                .duration_since(UNIX_EPOCH)
839                .unwrap_or_default()
840                .as_millis();
841            let idx = metadata
842                .graph_projections
843                .iter()
844                .position(|projection| projection.name == name);
845            if let Some(idx) = idx {
846                metadata.graph_projections[idx].updated_at_unix_ms = now;
847                metadata.graph_projections[idx].state = "failed".to_string();
848                metadata.graph_projections[idx].last_materialized_sequence = None;
849                let result = metadata.graph_projections[idx].clone();
850                Self::mark_projection_dependent_jobs_stale(metadata, name, now);
851                return Some(result);
852            }
853            None
854        })
855    }
856
857    /// Mark a graph projection as stale while preserving any last materialized sequence.
858    pub fn mark_graph_projection_stale(
859        &self,
860        name: &str,
861    ) -> Result<Option<PhysicalGraphProjection>, Box<dyn std::error::Error>> {
862        self.update_physical_metadata(|metadata| {
863            let now = SystemTime::now()
864                .duration_since(UNIX_EPOCH)
865                .unwrap_or_default()
866                .as_millis();
867            let idx = metadata
868                .graph_projections
869                .iter()
870                .position(|projection| projection.name == name);
871            if let Some(idx) = idx {
872                metadata.graph_projections[idx].updated_at_unix_ms = now;
873                metadata.graph_projections[idx].state = "stale".to_string();
874                let result = metadata.graph_projections[idx].clone();
875                Self::mark_projection_dependent_jobs_stale(metadata, name, now);
876                return Some(result);
877            }
878            None
879        })
880    }
881
882    fn mark_projection_dependent_jobs_stale(
883        metadata: &mut PhysicalMetadataFile,
884        projection_name: &str,
885        now: u128,
886    ) {
887        for job in metadata.analytics_jobs.iter_mut() {
888            if job.projection.as_deref() == Some(projection_name) && job.state != "declared" {
889                job.state = "stale".to_string();
890                job.updated_at_unix_ms = now;
891            }
892        }
893    }
894
895    fn rearm_projection_dependent_jobs_declared(
896        metadata: &mut PhysicalMetadataFile,
897        projection_name: &str,
898        now: u128,
899    ) {
900        for job in metadata.analytics_jobs.iter_mut() {
901            if job.projection.as_deref() == Some(projection_name) && job.state == "stale" {
902                job.state = "declared".to_string();
903                job.last_run_sequence = None;
904                job.updated_at_unix_ms = now;
905            }
906        }
907    }
908
909    /// Declare or update analytics job metadata without marking it as executed.
910    pub fn save_analytics_job(
911        &self,
912        kind: impl Into<String>,
913        projection: Option<String>,
914        metadata_entries: BTreeMap<String, String>,
915    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
916        let kind = kind.into();
917        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
918        if let Some(projection_name) = projection.as_deref() {
919            if !self.graph_projection_is_declared(projection_name) {
920                return Err(format!(
921                    "graph projection '{projection_name}' is not declared in physical metadata"
922                )
923                .into());
924            }
925        }
926
927        self.update_physical_metadata(|metadata| {
928            let now = SystemTime::now()
929                .duration_since(UNIX_EPOCH)
930                .unwrap_or_default()
931                .as_millis();
932
933            let job = if let Some(existing) = metadata
934                .analytics_jobs
935                .iter_mut()
936                .find(|job| job.id == job_id)
937            {
938                existing.kind = kind.clone();
939                existing.projection = projection.clone();
940                existing.updated_at_unix_ms = now;
941                existing.metadata = metadata_entries.clone();
942                if existing.last_run_sequence.is_none() {
943                    existing.state = "declared".to_string();
944                }
945                existing.clone()
946            } else {
947                let job = PhysicalAnalyticsJob {
948                    id: job_id.clone(),
949                    kind: kind.clone(),
950                    state: "declared".to_string(),
951                    projection: projection.clone(),
952                    created_at_unix_ms: now,
953                    updated_at_unix_ms: now,
954                    last_run_sequence: None,
955                    metadata: metadata_entries.clone(),
956                };
957                metadata.analytics_jobs.push(job.clone());
958                job
959            };
960
961            metadata
962                .analytics_jobs
963                .sort_by(|left, right| left.id.cmp(&right.id));
964            job
965        })
966    }
967
968    /// Record or update analytics job metadata in the persisted physical metadata.
969    pub fn record_analytics_job(
970        &self,
971        kind: impl Into<String>,
972        projection: Option<String>,
973        metadata_entries: BTreeMap<String, String>,
974    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
975        let kind = kind.into();
976        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
977        if let Some(projection_name) = projection.as_deref() {
978            if !self.graph_projection_is_declared(projection_name) {
979                return Err(format!(
980                    "graph projection '{projection_name}' is not declared in physical metadata"
981                )
982                .into());
983            }
984            if !self.graph_projection_is_operational(projection_name) {
985                return Err(format!(
986                    "graph projection '{projection_name}' is declared but not operationally materialized"
987                )
988                .into());
989            }
990        }
991
992        self.update_physical_metadata(|metadata| {
993            let now = SystemTime::now()
994                .duration_since(UNIX_EPOCH)
995                .unwrap_or_default()
996                .as_millis();
997
998            let existing = metadata
999                .analytics_jobs
1000                .iter_mut()
1001                .find(|job| job.id == job_id)?;
1002            existing.state = "completed".to_string();
1003            existing.kind = kind.clone();
1004            existing.projection = projection.clone();
1005            existing.updated_at_unix_ms = now;
1006            existing.last_run_sequence = Some(metadata.superblock.sequence);
1007            existing.metadata = metadata_entries.clone();
1008            let job = existing.clone();
1009
1010            metadata
1011                .analytics_jobs
1012                .sort_by(|left, right| left.id.cmp(&right.id));
1013            Some(job)
1014        })
1015        .and_then(|job| {
1016            job.ok_or_else(|| {
1017                format!("analytics job '{job_id}' is not declared in physical metadata").into()
1018            })
1019        })
1020    }
1021
1022    /// Mark a declared analytics job as running.
1023    pub fn queue_analytics_job(
1024        &self,
1025        kind: impl Into<String>,
1026        projection: Option<String>,
1027        metadata_entries: BTreeMap<String, String>,
1028    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
1029        let kind = kind.into();
1030        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
1031        if let Some(projection_name) = projection.as_deref() {
1032            if !self.graph_projection_is_declared(projection_name) {
1033                return Err(format!(
1034                    "graph projection '{projection_name}' is not declared in physical metadata"
1035                )
1036                .into());
1037            }
1038            if !self.graph_projection_is_operational(projection_name) {
1039                return Err(format!(
1040                    "graph projection '{projection_name}' is declared but not operationally materialized"
1041                )
1042                .into());
1043            }
1044        }
1045
1046        self.update_physical_metadata(|metadata| {
1047            let now = SystemTime::now()
1048                .duration_since(UNIX_EPOCH)
1049                .unwrap_or_default()
1050                .as_millis();
1051
1052            let existing = metadata
1053                .analytics_jobs
1054                .iter_mut()
1055                .find(|job| job.id == job_id)?;
1056            existing.state = "queued".to_string();
1057            existing.kind = kind.clone();
1058            existing.projection = projection.clone();
1059            existing.updated_at_unix_ms = now;
1060            existing.metadata = metadata_entries.clone();
1061            let job = existing.clone();
1062
1063            metadata
1064                .analytics_jobs
1065                .sort_by(|left, right| left.id.cmp(&right.id));
1066            Some(job)
1067        })
1068        .and_then(|job| {
1069            job.ok_or_else(|| {
1070                format!("analytics job '{job_id}' is not declared in physical metadata").into()
1071            })
1072        })
1073    }
1074
1075    /// Mark a declared analytics job as running.
1076    pub fn start_analytics_job(
1077        &self,
1078        kind: impl Into<String>,
1079        projection: Option<String>,
1080        metadata_entries: BTreeMap<String, String>,
1081    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
1082        let kind = kind.into();
1083        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
1084        if let Some(projection_name) = projection.as_deref() {
1085            if !self.graph_projection_is_declared(projection_name) {
1086                return Err(format!(
1087                    "graph projection '{projection_name}' is not declared in physical metadata"
1088                )
1089                .into());
1090            }
1091            if !self.graph_projection_is_operational(projection_name) {
1092                return Err(format!(
1093                    "graph projection '{projection_name}' is declared but not operationally materialized"
1094                )
1095                .into());
1096            }
1097        }
1098
1099        self.update_physical_metadata(|metadata| {
1100            let now = SystemTime::now()
1101                .duration_since(UNIX_EPOCH)
1102                .unwrap_or_default()
1103                .as_millis();
1104
1105            let existing = metadata
1106                .analytics_jobs
1107                .iter_mut()
1108                .find(|job| job.id == job_id)?;
1109            existing.state = "running".to_string();
1110            existing.kind = kind.clone();
1111            existing.projection = projection.clone();
1112            existing.updated_at_unix_ms = now;
1113            existing.metadata = metadata_entries.clone();
1114            let job = existing.clone();
1115
1116            metadata
1117                .analytics_jobs
1118                .sort_by(|left, right| left.id.cmp(&right.id));
1119            Some(job)
1120        })
1121        .and_then(|job| {
1122            job.ok_or_else(|| {
1123                format!("analytics job '{job_id}' is not declared in physical metadata").into()
1124            })
1125        })
1126    }
1127
1128    /// Mark a declared analytics job as failed.
1129    pub fn fail_analytics_job(
1130        &self,
1131        kind: impl Into<String>,
1132        projection: Option<String>,
1133        metadata_entries: BTreeMap<String, String>,
1134    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
1135        let kind = kind.into();
1136        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
1137
1138        self.update_physical_metadata(|metadata| {
1139            let now = SystemTime::now()
1140                .duration_since(UNIX_EPOCH)
1141                .unwrap_or_default()
1142                .as_millis();
1143
1144            let existing = metadata
1145                .analytics_jobs
1146                .iter_mut()
1147                .find(|job| job.id == job_id)?;
1148            existing.state = "failed".to_string();
1149            existing.kind = kind.clone();
1150            existing.projection = projection.clone();
1151            existing.updated_at_unix_ms = now;
1152            existing.metadata = metadata_entries.clone();
1153            let job = existing.clone();
1154
1155            metadata
1156                .analytics_jobs
1157                .sort_by(|left, right| left.id.cmp(&right.id));
1158            Some(job)
1159        })
1160        .and_then(|job| {
1161            job.ok_or_else(|| {
1162                format!("analytics job '{job_id}' is not declared in physical metadata").into()
1163            })
1164        })
1165    }
1166
1167    /// Mark a declared analytics job as stale.
1168    pub fn mark_analytics_job_stale(
1169        &self,
1170        kind: impl Into<String>,
1171        projection: Option<String>,
1172        metadata_entries: BTreeMap<String, String>,
1173    ) -> Result<PhysicalAnalyticsJob, Box<dyn std::error::Error>> {
1174        let kind = kind.into();
1175        let job_id = Self::analytics_job_id(&kind, projection.as_deref());
1176
1177        self.update_physical_metadata(|metadata| {
1178            let now = SystemTime::now()
1179                .duration_since(UNIX_EPOCH)
1180                .unwrap_or_default()
1181                .as_millis();
1182
1183            let existing = metadata
1184                .analytics_jobs
1185                .iter_mut()
1186                .find(|job| job.id == job_id)?;
1187            existing.state = "stale".to_string();
1188            existing.kind = kind.clone();
1189            existing.projection = projection.clone();
1190            existing.updated_at_unix_ms = now;
1191            existing.metadata = metadata_entries.clone();
1192            let job = existing.clone();
1193
1194            metadata
1195                .analytics_jobs
1196                .sort_by(|left, right| left.id.cmp(&right.id));
1197            Some(job)
1198        })
1199        .and_then(|job| {
1200            job.ok_or_else(|| {
1201                format!("analytics job '{job_id}' is not declared in physical metadata").into()
1202            })
1203        })
1204    }
1205
1206    /// Create a named export by copying the current database file, including its internal metadata.
1207    pub fn create_named_export(
1208        &self,
1209        name: impl Into<String>,
1210    ) -> Result<ExportDescriptor, Box<dyn std::error::Error>> {
1211        let name = name.into();
1212        if self.options.mode != StorageMode::Persistent {
1213            return Err("exports require persistent mode".into());
1214        }
1215        let Some(path) = self.path() else {
1216            return Err("database path is not available".into());
1217        };
1218
1219        self.flush()?;
1220
1221        let _publication = self.store.vault_publication_lock.lock();
1222        let mut metadata = match self.load_or_bootstrap_physical_metadata(true) {
1223            Ok(metadata) => metadata,
1224            Err(_)
1225                if self.options.storage_profile.deploy_profile
1226                    == crate::storage::DeployProfile::Embedded
1227                    && self.options.storage_profile.packaging
1228                        == crate::storage::StoragePackaging::SingleFile =>
1229            {
1230                crate::physical::PhysicalMetadataFile::from_state(
1231                    self.options.clone(),
1232                    self.catalog_snapshot(),
1233                    self.physical_collection_roots(),
1234                    self.physical_index_state(),
1235                    None,
1236                )
1237            }
1238            Err(err) => return Err(err),
1239        };
1240        let export_data_path = reddb_file::copy_physical_export_data_file(path, &name)?;
1241
1242        let descriptor = ExportDescriptor {
1243            name: name.clone(),
1244            created_at_unix_ms: SystemTime::now()
1245                .duration_since(UNIX_EPOCH)
1246                .unwrap_or_default()
1247                .as_millis(),
1248            snapshot_id: metadata
1249                .snapshots
1250                .last()
1251                .map(|snapshot| snapshot.snapshot_id),
1252            superblock_sequence: metadata.superblock.sequence,
1253            data_path: export_data_path.display().to_string(),
1254            metadata_path: export_data_path.display().to_string(),
1255            collection_count: metadata.catalog.total_collections,
1256            total_entities: metadata.catalog.total_entities,
1257        };
1258
1259        metadata
1260            .exports
1261            .retain(|export| export.name != descriptor.name);
1262        metadata.exports.push(descriptor.clone());
1263        self.prune_export_registry(&mut metadata.exports);
1264        self.persist_physical_metadata_record(&metadata)?;
1265
1266        Ok(descriptor)
1267    }
1268
1269    /// Enable or disable a physical index entry in the persisted registry.
1270    pub fn set_index_enabled(
1271        &self,
1272        name: &str,
1273        enabled: bool,
1274    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1275        let Some(status) = self.index_status(name) else {
1276            return Err(format!("index '{name}' is not present in catalog status").into());
1277        };
1278        if !status.declared {
1279            return Err(format!("index '{name}' is not declared in physical metadata").into());
1280        }
1281        self.update_physical_metadata(|metadata| {
1282            let now = SystemTime::now()
1283                .duration_since(UNIX_EPOCH)
1284                .unwrap_or_default()
1285                .as_millis();
1286            if let Some(index) = metadata.indexes.iter_mut().find(|index| index.name == name) {
1287                index.enabled = enabled;
1288                if !enabled {
1289                    index.build_state = "disabled".to_string();
1290                } else if index.build_state == "disabled" {
1291                    index.build_state = if index.artifact_root_page.is_some() {
1292                        "ready".to_string()
1293                    } else {
1294                        "declared-unbuilt".to_string()
1295                    };
1296                }
1297                index.last_refresh_ms = Some(now);
1298                return Some(index.clone());
1299            }
1300            None
1301        })
1302    }
1303
1304    /// Mark a declared physical index as building in the persisted registry.
1305    pub fn mark_index_building(
1306        &self,
1307        name: &str,
1308    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1309        let Some(status) = self.index_status(name) else {
1310            return Err(format!("index '{name}' is not present in catalog status").into());
1311        };
1312        if !status.declared {
1313            return Err(format!("index '{name}' is not declared in physical metadata").into());
1314        }
1315        if status.lifecycle_state == "disabled" {
1316            return Err(format!("index '{name}' is disabled").into());
1317        }
1318        self.update_physical_metadata(|metadata| {
1319            let now = SystemTime::now()
1320                .duration_since(UNIX_EPOCH)
1321                .unwrap_or_default()
1322                .as_millis();
1323            if let Some(index) = metadata.indexes.iter_mut().find(|index| index.name == name) {
1324                index.build_state = "building".to_string();
1325                index.last_refresh_ms = Some(now);
1326                return Some(index.clone());
1327            }
1328            None
1329        })
1330    }
1331
1332    /// Mark a declared physical index as failed in the persisted registry.
1333    pub fn fail_index(
1334        &self,
1335        name: &str,
1336    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1337        let Some(status) = self.index_status(name) else {
1338            return Err(format!("index '{name}' is not present in catalog status").into());
1339        };
1340        if !status.declared {
1341            return Err(format!("index '{name}' is not declared in physical metadata").into());
1342        }
1343        self.update_physical_metadata(|metadata| {
1344            let now = SystemTime::now()
1345                .duration_since(UNIX_EPOCH)
1346                .unwrap_or_default()
1347                .as_millis();
1348            if let Some(index) = metadata.indexes.iter_mut().find(|index| index.name == name) {
1349                index.build_state = "failed".to_string();
1350                index.last_refresh_ms = Some(now);
1351                return Some(index.clone());
1352            }
1353            None
1354        })
1355    }
1356
1357    /// Mark a declared physical index as stale in the persisted registry.
1358    pub fn mark_index_stale(
1359        &self,
1360        name: &str,
1361    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1362        let Some(status) = self.index_status(name) else {
1363            return Err(format!("index '{name}' is not present in catalog status").into());
1364        };
1365        if !status.declared {
1366            return Err(format!("index '{name}' is not declared in physical metadata").into());
1367        }
1368        self.update_physical_metadata(|metadata| {
1369            let now = SystemTime::now()
1370                .duration_since(UNIX_EPOCH)
1371                .unwrap_or_default()
1372                .as_millis();
1373            if let Some(index) = metadata.indexes.iter_mut().find(|index| index.name == name) {
1374                index.build_state = if index.enabled {
1375                    "stale".to_string()
1376                } else {
1377                    "disabled".to_string()
1378                };
1379                index.last_refresh_ms = Some(now);
1380                return Some(index.clone());
1381            }
1382            None
1383        })
1384    }
1385
1386    /// Mark a declared physical index as ready in the persisted registry.
1387    pub fn mark_index_ready(
1388        &self,
1389        name: &str,
1390    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1391        self.warmup_index(name)
1392    }
1393
1394    /// Mark a physical index as warmed up/refreshed in the persisted registry.
1395    pub fn warmup_index(
1396        &self,
1397        name: &str,
1398    ) -> Result<Option<PhysicalIndexState>, Box<dyn std::error::Error>> {
1399        let Some(status) = self.index_status(name) else {
1400            return Err(format!("index '{name}' is not present in catalog status").into());
1401        };
1402        if !status.declared {
1403            return Err(format!("index '{name}' is not declared in physical metadata").into());
1404        }
1405        if status.lifecycle_state == "disabled" {
1406            return Err(format!("index '{name}' is disabled").into());
1407        }
1408        if !status.operational {
1409            return Err(
1410                format!("index '{name}' is declared but not operationally materialized").into(),
1411            );
1412        }
1413        let warmed_artifact = self
1414            .physical_indexes()
1415            .into_iter()
1416            .find(|index| index.name == name)
1417            .map(|mut index| {
1418                self.warmup_native_vector_artifact_for_index(&index)?;
1419                self.apply_runtime_native_artifact_to_index_state(&mut index)?;
1420                Ok::<_, String>(index)
1421            })
1422            .transpose()
1423            .map_err(|err| -> Box<dyn std::error::Error> { err.into() })?;
1424
1425        self.update_physical_metadata(|metadata| {
1426            let now = SystemTime::now()
1427                .duration_since(UNIX_EPOCH)
1428                .unwrap_or_default()
1429                .as_millis();
1430            if let Some(index) = metadata.indexes.iter_mut().find(|index| index.name == name) {
1431                if let Some(warmed) = warmed_artifact.as_ref() {
1432                    index.entries = warmed.entries;
1433                    index.estimated_memory_bytes = warmed.estimated_memory_bytes;
1434                    index.backend = warmed.backend.clone();
1435                    index.build_state = "ready".to_string();
1436                }
1437                index.last_refresh_ms = Some(now);
1438                return Some(index.clone());
1439            }
1440            None
1441        })
1442    }
1443
1444    /// Rebuild physical index metadata from the current catalog, optionally restricted to one collection.
1445    pub fn rebuild_index_registry(
1446        &self,
1447        collection: Option<&str>,
1448    ) -> Result<Vec<PhysicalIndexState>, Box<dyn std::error::Error>> {
1449        let fresh = self.reconcile_index_states_with_native_artifacts(self.physical_index_state());
1450        self.update_physical_metadata(|metadata| {
1451            let now = SystemTime::now()
1452                .duration_since(UNIX_EPOCH)
1453                .unwrap_or_default()
1454                .as_millis();
1455
1456            let mut affected = Vec::new();
1457            let declared = metadata.indexes.clone();
1458            for declared_index in declared {
1459                let matches_collection = collection.is_none_or(|collection_name| {
1460                    declared_index.collection.as_deref() == Some(collection_name)
1461                });
1462                if !matches_collection {
1463                    continue;
1464                }
1465
1466                let mut rebuilt = fresh
1467                    .iter()
1468                    .find(|index| index.name == declared_index.name)
1469                    .cloned()
1470                    .unwrap_or_else(|| {
1471                        let mut index = declared_index.clone();
1472                        index.build_state = "declared-unbuilt".to_string();
1473                        index
1474                    });
1475                rebuilt.enabled = declared_index.enabled;
1476                rebuilt.artifact_kind = rebuilt
1477                    .artifact_kind
1478                    .or_else(|| declared_index.artifact_kind.clone());
1479                rebuilt.artifact_root_page = rebuilt
1480                    .artifact_root_page
1481                    .or(declared_index.artifact_root_page);
1482                rebuilt.artifact_checksum = rebuilt
1483                    .artifact_checksum
1484                    .or(declared_index.artifact_checksum);
1485                rebuilt.build_state =
1486                    Self::finalize_rebuilt_index_build_state(&declared_index, &rebuilt);
1487                rebuilt.last_refresh_ms = Some(now);
1488
1489                if let Some(existing) = metadata
1490                    .indexes
1491                    .iter_mut()
1492                    .find(|index| index.name == rebuilt.name)
1493                {
1494                    *existing = rebuilt.clone();
1495                } else {
1496                    metadata.indexes.push(rebuilt.clone());
1497                }
1498
1499                affected.push(rebuilt);
1500            }
1501
1502            affected
1503        })
1504    }
1505
1506    fn finalize_rebuilt_index_build_state(
1507        declared: &PhysicalIndexState,
1508        rebuilt: &PhysicalIndexState,
1509    ) -> String {
1510        if !rebuilt.enabled {
1511            return "disabled".to_string();
1512        }
1513
1514        if declared.build_state == "failed" || rebuilt.build_state == "failed" {
1515            return "failed".to_string();
1516        }
1517
1518        let native_artifact_family = Self::native_artifact_kind_for_index(rebuilt.kind).is_some();
1519        if native_artifact_family {
1520            if rebuilt.artifact_root_page.is_some() && rebuilt.artifact_checksum.is_some() {
1521                return "ready".to_string();
1522            }
1523            if declared.artifact_root_page.is_some()
1524                || declared.artifact_checksum.is_some()
1525                || declared.artifact_kind.is_some()
1526            {
1527                return "stale".to_string();
1528            }
1529            return "declared-unbuilt".to_string();
1530        }
1531
1532        if rebuilt.entries > 0 {
1533            return "ready".to_string();
1534        }
1535
1536        if matches!(
1537            declared.build_state.as_str(),
1538            "stale" | "artifact-published" | "registry-loaded"
1539        ) {
1540            return "stale".to_string();
1541        }
1542
1543        "declared-unbuilt".to_string()
1544    }
1545}