1use std::collections::BTreeMap;
14use std::path::{Path, PathBuf};
15use std::sync::atomic::{AtomicU64, Ordering};
16use std::sync::Arc;
17
18use uqa_analysis::Analyzer;
19use uqa_core::{DocId, Value};
20
21use crate::document_store::DocumentStore;
22use crate::inverted_index::InvertedIndex;
23use crate::vector_index::{VectorIndex, VectorIndexOpenMode, VectorIndexSpec};
24use crate::CatalogFacade;
25
26#[derive(Debug, thiserror::Error)]
27pub enum StorageBackendError {
28 #[error(transparent)]
29 Memory(#[from] uqa_core::memory::MemoryError),
30 #[error(transparent)]
31 Cancelled(#[from] uqa_core::QueryCancelled),
32 #[error("text analysis failed: {0}")]
33 Analysis(#[from] uqa_analysis::AnalysisError),
34 #[error("payload serialization failed: {0}")]
35 Serde(#[from] serde_json::Error),
36 #[error("{backend} storage failed: {source}")]
37 Backend {
38 backend: &'static str,
39 #[source]
40 source: Box<dyn std::error::Error + Send + Sync>,
41 },
42 #[error("{0}")]
43 Other(String),
44}
45
46impl StorageBackendError {
47 pub fn backend(
48 backend: &'static str,
49 source: impl std::error::Error + Send + Sync + 'static,
50 ) -> Self {
51 Self::Backend {
52 backend,
53 source: Box::new(source),
54 }
55 }
56}
57
58pub type StorageBackendResult<T> = std::result::Result<T, StorageBackendError>;
59
60#[derive(Clone, Debug)]
62pub struct StorageSessionAffinity(Arc<()>);
63
64impl StorageSessionAffinity {
65 #[must_use]
66 pub fn new() -> Self {
67 Self(Arc::new(()))
68 }
69}
70
71impl Default for StorageSessionAffinity {
72 fn default() -> Self {
73 Self::new()
74 }
75}
76
77impl PartialEq for StorageSessionAffinity {
78 fn eq(&self, other: &Self) -> bool {
79 Arc::ptr_eq(&self.0, &other.0)
80 }
81}
82
83impl Eq for StorageSessionAffinity {}
84
85#[derive(Debug, thiserror::Error)]
86#[error("catalog and data backend must share one storage transaction context")]
87pub struct StorageSessionMismatch;
88
89#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
91pub struct StorageSavepointId(u64);
92
93impl StorageSavepointId {
94 #[must_use]
95 pub fn allocate() -> Self {
96 static NEXT_ID: AtomicU64 = AtomicU64::new(1);
97 let id = NEXT_ID
98 .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
99 current.checked_add(1)
100 })
101 .expect("storage savepoint identity space exhausted");
102 Self(id)
103 }
104
105 pub fn backend_name(self) -> String {
106 self.0.to_string()
107 }
108}
109
110pub struct PersistentStorageSession {
116 pub catalog: Arc<dyn CatalogFacade>,
117 pub backend: Arc<dyn PersistentStorageBackend>,
118}
119
120#[derive(Clone, Debug, PartialEq, Eq, Hash)]
122pub enum PersistentStorageIdentity {
123 File(PathBuf),
124 Opaque(String),
125}
126
127impl PersistentStorageIdentity {
128 pub fn for_database_path(path: &Path) -> StorageBackendResult<Self> {
130 let path = resolve_final_symlinks(path)?;
131 match std::fs::canonicalize(&path) {
132 Ok(canonical) => Ok(Self::File(canonical)),
133 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
134 let file_name = path.file_name().ok_or_else(|| {
135 StorageBackendError::Other(format!(
136 "database path `{}` has no file name",
137 path.display()
138 ))
139 })?;
140 let parent = path
141 .parent()
142 .filter(|parent| !parent.as_os_str().is_empty())
143 .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
144 let parent = std::fs::canonicalize(&parent).map_err(|error| {
145 StorageBackendError::Other(format!(
146 "canonicalize database directory `{}`: {error}",
147 parent.display()
148 ))
149 })?;
150 Ok(Self::File(parent.join(file_name)))
151 }
152 Err(error) => Err(StorageBackendError::Other(format!(
153 "canonicalize database `{}`: {error}",
154 path.display()
155 ))),
156 }
157 }
158}
159
160fn resolve_final_symlinks(path: &Path) -> StorageBackendResult<PathBuf> {
161 let mut current = path.to_path_buf();
162 let mut followed = 0usize;
163 loop {
164 match std::fs::symlink_metadata(¤t) {
165 Ok(metadata) if metadata.file_type().is_symlink() => {
166 followed += 1;
167 if followed > 40 {
168 return Err(StorageBackendError::Other(format!(
169 "database path `{}` has too many symbolic-link levels",
170 path.display()
171 )));
172 }
173 let target = std::fs::read_link(¤t).map_err(|error| {
174 StorageBackendError::Other(format!(
175 "read database symbolic link `{}`: {error}",
176 current.display()
177 ))
178 })?;
179 current = if target.is_absolute() {
180 target
181 } else {
182 current
183 .parent()
184 .filter(|parent| !parent.as_os_str().is_empty())
185 .unwrap_or_else(|| Path::new("."))
186 .join(target)
187 };
188 }
189 Ok(_) => return Ok(current),
190 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(current),
191 Err(error) => {
192 return Err(StorageBackendError::Other(format!(
193 "inspect database path `{}`: {error}",
194 current.display()
195 )))
196 }
197 }
198 }
199}
200
201#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
203pub enum StorageTransactionModel {
204 #[default]
205 ProviderSerialized,
206 VersionedConcurrent { database: crate::mvcc::DatabaseId },
208}
209
210impl StorageTransactionModel {
211 pub fn is_versioned(self) -> bool {
212 matches!(self, Self::VersionedConcurrent { .. })
213 }
214}
215
216impl PersistentStorageSession {
217 pub fn new(
218 catalog: Arc<dyn CatalogFacade>,
219 backend: Arc<dyn PersistentStorageBackend>,
220 ) -> Self {
221 Self { catalog, backend }
222 }
223
224 pub fn validate_transaction_affinity(&self) -> StorageBackendResult<()> {
226 let affinity = self.backend.transaction_affinity();
227 let model = self.backend.transaction_model();
228 if self.catalog.transaction_affinity() != affinity
229 || self.catalog.transaction_model() != model
230 || (model.is_versioned() && affinity.is_none())
231 {
232 return Err(StorageBackendError::backend(
233 "session",
234 StorageSessionMismatch,
235 ));
236 }
237 if model.is_versioned() && self.backend.write_cancellation().is_none() {
238 return Err(StorageBackendError::Other(
239 "versioned storage backend does not expose write cancellation".into(),
240 ));
241 }
242 Ok(())
243 }
244}
245
246pub trait PersistentStorageProvider: Send + Sync {
252 fn open_session(&self) -> StorageBackendResult<PersistentStorageSession>;
253
254 fn open_session_with_cancellation(
256 &self,
257 cancellation: &uqa_core::CancellationToken,
258 ) -> StorageBackendResult<PersistentStorageSession> {
259 cancellation.check()?;
260 let session = self.open_session()?;
261 if session.backend.transaction_model().is_versioned() {
262 return Err(StorageBackendError::Other(
263 "cancellable independent storage sessions are not implemented by this provider"
264 .into(),
265 ));
266 }
267 Ok(session)
268 }
269
270 fn open_initial_session(&self) -> StorageBackendResult<PersistentStorageSession> {
272 self.open_session()
273 }
274
275 fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
277 Ok(None)
278 }
279
280 fn auxiliary_encryption_key(&self) -> Option<crate::StorageEncryptionKey> {
284 None
285 }
286}
287
288pub trait PersistentStorageBackend: Send + Sync {
290 fn retention_control(&self) -> Option<crate::read_control::StorageReadControl> {
292 None
293 }
294
295 fn notification_publications(
297 &self,
298 ) -> Option<&dyn crate::notifications::NotificationPublicationStore> {
299 None
300 }
301
302 fn serializable_session(&self) -> Option<&dyn crate::mvcc::SerializableSession> {
304 None
305 }
306
307 fn write_cancellation(&self) -> Option<uqa_core::CancellationToken> {
309 None
310 }
311
312 fn open_session_with_cancellation(
314 &self,
315 cancellation: &uqa_core::CancellationToken,
316 ) -> StorageBackendResult<PersistentStorageSession> {
317 cancellation.check()?;
318 if self.transaction_model().is_versioned() {
319 return Err(StorageBackendError::Other(
320 "cancellable independent storage sessions are not implemented by this backend"
321 .into(),
322 ));
323 }
324 self.open_session()
325 }
326
327 fn open_controlled_session(
329 &self,
330 _control: &crate::read_control::StorageReadControl,
331 ) -> StorageBackendResult<PersistentStorageSession> {
332 Err(StorageBackendError::Other(
333 "controlled independent sessions are not implemented by this backend".into(),
334 ))
335 }
336
337 fn open_retained_read_session(
339 &self,
340 cancellation: &uqa_core::CancellationToken,
341 ) -> StorageBackendResult<PersistentStorageSession> {
342 cancellation.check()?;
343 Err(StorageBackendError::Other(
344 "retained read sessions are not implemented by this storage backend".into(),
345 ))
346 }
347
348 fn transaction_model(&self) -> StorageTransactionModel {
350 StorageTransactionModel::ProviderSerialized
351 }
352
353 fn identifier_allocator(&self) -> Option<&dyn crate::mvcc::IdentifierAllocator> {
355 None
356 }
357
358 fn transaction_affinity(&self) -> Option<StorageSessionAffinity> {
360 None
361 }
362
363 fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
365 Ok(None)
366 }
367
368 fn auxiliary_encryption_key(&self) -> Option<crate::StorageEncryptionKey> {
371 None
372 }
373
374 fn open_session(&self) -> StorageBackendResult<PersistentStorageSession> {
376 Err(StorageBackendError::Other(
377 "independent sessions are not implemented for this persistent backend".into(),
378 ))
379 }
380
381 fn supports_concurrent_pinned_read_and_write(&self) -> bool {
383 false
384 }
385
386 fn document_store(&self, table: &str) -> Box<dyn DocumentStore>;
387
388 fn migrate_document_storage(&self) -> StorageBackendResult<()> {
390 Ok(())
391 }
392
393 fn inverted_index(&self, table: &str, analyzer: Analyzer) -> Box<dyn InvertedIndex>;
394
395 fn migrate_inverted_index_storage(&self) -> StorageBackendResult<()> {
398 Ok(())
399 }
400
401 fn vector_index(
402 &self,
403 table: &str,
404 field: &str,
405 dimensions: u32,
406 spec: VectorIndexSpec,
407 mode: VectorIndexOpenMode,
408 ) -> StorageBackendResult<Box<dyn VectorIndex>>;
409
410 fn diskann_index(
412 &self,
413 _binding: crate::diskann_index::DiskANNIndexBinding<'_>,
414 _options: crate::diskann_index::DiskANNIndexOptions,
415 _temporary: &crate::diskann_index::build::DiskANNTemporaryBudget,
416 _mode: VectorIndexOpenMode,
417 ) -> StorageBackendResult<Box<dyn VectorIndex>> {
418 Err(StorageBackendError::Other(
419 "this backend does not support catalog-bound DiskANN indexes".into(),
420 ))
421 }
422
423 fn diskann_journal_pruner(
425 &self,
426 _binding: crate::diskann_index::DiskANNIndexBinding<'_>,
427 _max_record_bytes: usize,
428 ) -> StorageBackendResult<Box<dyn crate::diskann_index::changes::DiskANNJournalPruner>> {
429 Err(StorageBackendError::Other(
430 "this backend does not support DiskANN journal maintenance".into(),
431 ))
432 }
433
434 fn diskann_maintenance_source(
436 &self,
437 _binding: crate::diskann_index::DiskANNIndexBinding<'_>,
438 _max_record_bytes: usize,
439 ) -> StorageBackendResult<Box<dyn crate::diskann_index::maintenance::DiskANNMaintenanceSource>>
440 {
441 Err(StorageBackendError::Other(
442 "this backend does not support DiskANN rebuild maintenance".into(),
443 ))
444 }
445
446 fn retire_diskann_index(
447 &self,
448 _binding: crate::diskann_index::DiskANNIndexBinding<'_>,
449 ) -> StorageBackendResult<()> {
450 Err(StorageBackendError::Other(
451 "this backend does not support DiskANN retirement".into(),
452 ))
453 }
454
455 fn drop_vector_index_metadata(&self, _table: &str, _field: &str) -> StorageBackendResult<()> {
456 Ok(())
457 }
458
459 fn persists_btree_indexes(&self) -> bool {
461 false
462 }
463
464 fn load_btree_index(
467 &self,
468 _table: &str,
469 _field: &crate::ValueIndexKey,
470 ) -> StorageBackendResult<Option<Vec<(DocId, Value)>>> {
471 Ok(None)
472 }
473
474 fn btree_index_fields(&self, _table: &str) -> StorageBackendResult<Vec<crate::ValueIndexKey>> {
475 Ok(Vec::new())
476 }
477
478 fn read_btree_index_entry(
480 &self,
481 _table: &str,
482 _field: &crate::ValueIndexKey,
483 _doc_id: DocId,
484 ) -> StorageBackendResult<crate::ValueIndexEntry> {
485 if self.persists_btree_indexes() {
486 return Err(crate::StorageBackendError::Other(
487 "point reads of stored B-tree entries are not implemented for this backend".into(),
488 ));
489 }
490 Ok(crate::ValueIndexEntry::Unbuilt)
491 }
492
493 fn btree_index_repairs(&self) -> StorageBackendResult<Vec<(String, crate::ValueIndexKey)>> {
497 Ok(Vec::new())
498 }
499
500 fn clear_btree_index_repair(
501 &self,
502 _table: &str,
503 _field: &crate::ValueIndexKey,
504 ) -> StorageBackendResult<()> {
505 Ok(())
506 }
507
508 fn replace_btree_index(
509 &self,
510 _table: &str,
511 _field: &crate::ValueIndexKey,
512 _values: &[(DocId, Value)],
513 ) -> StorageBackendResult<()> {
514 Ok(())
515 }
516
517 fn repair_btree_index(
521 &self,
522 table: &str,
523 field: &crate::ValueIndexKey,
524 complete: &[(DocId, Value)],
525 _stale_doc_ids: &[DocId],
526 _missing: &[(DocId, Value)],
527 ) -> StorageBackendResult<()> {
528 self.replace_btree_index(table, field, complete)
529 }
530
531 fn replace_btree_indexes(
534 &self,
535 table: &str,
536 indexes: &[(&crate::ValueIndexKey, &[(DocId, Value)])],
537 ) -> StorageBackendResult<()> {
538 for (field, values) in indexes {
539 self.replace_btree_index(table, field, values)?;
540 }
541 Ok(())
542 }
543
544 fn apply_btree_index_write(
545 &self,
546 _table: &str,
547 _doc_id: DocId,
548 _values: Option<&BTreeMap<crate::ValueIndexKey, Value>>,
549 ) -> StorageBackendResult<()> {
550 Ok(())
551 }
552
553 fn drop_btree_index(
554 &self,
555 _table: &str,
556 _field: &crate::ValueIndexKey,
557 ) -> StorageBackendResult<()> {
558 Ok(())
559 }
560
561 fn clear_btree_indexes(&self, _table: &str) -> StorageBackendResult<()> {
562 Ok(())
563 }
564
565 fn reclaim_obsolete(&self) -> StorageBackendResult<()> {
567 if self.transaction_model().is_versioned() {
568 return Err(StorageBackendError::Other(
569 "versioned storage reclamation is not implemented by this backend".into(),
570 ));
571 }
572 Ok(())
573 }
574
575 fn vacuum(&self) -> StorageBackendResult<()> {
577 Ok(())
578 }
579
580 fn begin_transaction(&self) -> StorageBackendResult<()>;
581
582 fn begin_read_transaction(&self) -> StorageBackendResult<()> {
586 self.begin_transaction()
587 }
588
589 fn begin_upgradeable_transaction(&self) -> StorageBackendResult<()> {
591 self.begin_transaction()
592 }
593
594 fn refresh_transaction_snapshot(
596 &self,
597 _cancellation: &uqa_core::CancellationToken,
598 ) -> StorageBackendResult<()> {
599 Err(StorageBackendError::Other(
600 "transaction snapshot refresh is not supported by this backend".into(),
601 ))
602 }
603
604 fn in_transaction(&self) -> bool;
606
607 fn transaction_has_written(&self) -> StorageBackendResult<bool>;
609
610 fn change_version(&self) -> StorageBackendResult<Option<u64>> {
613 Ok(None)
614 }
615
616 fn change_version_monitor_is_nonblocking(&self) -> StorageBackendResult<bool> {
618 Ok(true)
619 }
620
621 fn pin_transaction_snapshot(&self) -> StorageBackendResult<()> {
623 Ok(())
624 }
625
626 fn commit_transaction(&self) -> StorageBackendResult<()>;
627
628 fn rollback_transaction(&self) -> StorageBackendResult<()>;
629
630 fn savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
631
632 fn release_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
633
634 fn rollback_to_savepoint(&self, id: StorageSavepointId) -> StorageBackendResult<()>;
635}
636
637#[cfg(all(test, unix))]
638mod identity_tests {
639 use super::*;
640
641 #[test]
642 fn dangling_database_symlink_keeps_the_target_identity_after_creation() {
643 use std::os::unix::fs::symlink;
644
645 let directory = tempfile::tempdir().unwrap();
646 let target = directory.path().join("target.db");
647 let link = directory.path().join("database.db");
648 symlink("target.db", &link).unwrap();
649
650 let before = PersistentStorageIdentity::for_database_path(&link).unwrap();
651 std::fs::File::create(&target).unwrap();
652 let after = PersistentStorageIdentity::for_database_path(&link).unwrap();
653
654 assert_eq!(before, after);
655 assert_eq!(
656 after,
657 PersistentStorageIdentity::File(target.canonicalize().unwrap())
658 );
659 }
660}