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