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