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 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 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 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 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 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 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 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 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 pub fn metadata_path(&self) -> Option<PathBuf> {
555 self.path
556 .as_ref()
557 .map(|path| PhysicalMetadataFile::metadata_path_for(path))
558 }
559
560 pub fn physical_metadata(&self) -> Option<PhysicalMetadataFile> {
562 self.load_or_bootstrap_physical_metadata(false).ok()
563 }
564
565 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 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 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 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 pub fn declared_graph_projections(&self) -> Vec<PhysicalGraphProjection> {
624 self.catalog_model_snapshot().declared_graph_projections
625 }
626
627 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 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 pub fn declared_analytics_jobs(&self) -> Vec<PhysicalAnalyticsJob> {
651 self.catalog_model_snapshot().declared_analytics_jobs
652 }
653
654 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 pub fn declared_indexes(&self) -> Vec<PhysicalIndexState> {
670 self.catalog_model_snapshot().declared_indexes
671 }
672
673 pub fn operational_indexes(&self) -> Vec<PhysicalIndexState> {
675 self.catalog_model_snapshot().operational_indexes
676 }
677
678 pub fn index_statuses(&self) -> Vec<CatalogIndexStatus> {
680 self.catalog_model_snapshot().index_statuses
681 }
682
683 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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}