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