corium_transactor/
backend.rs

1//! Pluggable transactor storage backends.
2//!
3//! A transactor keeps two kinds of durable state: the content-addressed
4//! blob store plus fenced root pointers (the "storage service"), and the
5//! per-database transaction log. [`StoreSpec`] selects the storage service
6//! backend — in-memory, filesystem, `PostgreSQL`, Turso, or S3 — and [`NodeStore`]
7//! dispatches the [`BlobStore`]/[`RootStore`] operations to it. Native
8//! service backends keep transaction logs in the same storage system as blobs
9//! and roots; memory and filesystem retain their existing log stores.
10
11use std::fmt;
12use std::path::PathBuf;
13use std::sync::Arc;
14use std::time::SystemTime;
15
16use async_trait::async_trait;
17use corium_log::{LogError, MemLogRegistry, TransactionLog, VersionedLog};
18#[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
19use corium_log::{NativeLogStorage, NativeVersionedLog};
20use corium_store::{BlobId, BlobIdStream, BlobStore, FsStore, MemoryStore, RootStore, StoreError};
21
22#[cfg(feature = "postgres")]
23use corium_store::PostgresBlobStore;
24#[cfg(feature = "s3")]
25use corium_store::S3BlobStore;
26#[cfg(feature = "turso")]
27use corium_store::TursoBlobStore;
28
29/// Selects the transactor's storage-service backend (blobs + roots).
30#[derive(Clone, Default)]
31pub enum StoreSpec {
32    /// In-memory blobs and roots; fully ephemeral and confined to one
33    /// process. The transaction log is in memory too, so the whole database
34    /// vanishes when the process exits — ideal for demos and tests.
35    Memory,
36    /// Blobs and roots under `{data_dir}/store`, log under `{data_dir}/logs`.
37    #[default]
38    Fs,
39    /// Blobs, roots, and transaction logs in `PostgreSQL`.
40    #[cfg(feature = "postgres")]
41    Postgres {
42        /// `PostgreSQL` URL or keyword/value connection string.
43        connection_string: String,
44    },
45    /// Blobs, roots, and transaction logs in a Turso (embeddable `SQLite`)
46    /// database at `path`. `path` is a local database file.
47    #[cfg(feature = "turso")]
48    Turso {
49        /// Filesystem path of the Turso database.
50        path: String,
51    },
52    /// Blobs, roots, and transaction logs in an S3 (or S3-compatible) bucket.
53    #[cfg(feature = "s3")]
54    S3 {
55        /// Target bucket name.
56        bucket: String,
57        /// Key prefix namespacing every object this store touches.
58        prefix: String,
59    },
60}
61
62impl fmt::Debug for StoreSpec {
63    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
64        match self {
65            Self::Memory => formatter.write_str("Memory"),
66            Self::Fs => formatter.write_str("Fs"),
67            #[cfg(feature = "postgres")]
68            Self::Postgres { .. } => formatter
69                .debug_struct("Postgres")
70                .field("connection_string", &"[REDACTED]")
71                .finish(),
72            #[cfg(feature = "turso")]
73            Self::Turso { path } => formatter.debug_struct("Turso").field("path", path).finish(),
74            #[cfg(feature = "s3")]
75            Self::S3 { bucket, prefix } => formatter
76                .debug_struct("S3")
77                .field("bucket", bucket)
78                .field("prefix", prefix)
79                .finish(),
80        }
81    }
82}
83
84/// The blob + root storage service a [`crate::node::TransactorNode`] runs
85/// over, chosen by [`StoreSpec`]. Dispatch is an enum rather than a trait
86/// object so every existing `impl BlobStore + RootStore` / `&dyn RootStore`
87/// call site keeps working unchanged.
88pub enum NodeStore {
89    /// In-memory backend.
90    Mem(MemoryStore),
91    /// Filesystem backend.
92    Fs(FsStore),
93    /// `PostgreSQL` backend.
94    #[cfg(feature = "postgres")]
95    Postgres(PostgresBlobStore),
96    /// Turso backend.
97    #[cfg(feature = "turso")]
98    Turso(TursoBlobStore),
99    /// S3 backend.
100    #[cfg(feature = "s3")]
101    S3(S3BlobStore),
102}
103
104impl NodeStore {
105    /// Opens the storage service for `spec`, relative to `data_dir` for the
106    /// filesystem backend.
107    ///
108    /// # Errors
109    /// Returns an error when the backing store cannot be opened.
110    // Only optional database-backed arms await; mem/fs are synchronous.
111    #[allow(clippy::unused_async)]
112    pub async fn open(spec: &StoreSpec, data_dir: &std::path::Path) -> Result<Self, StoreError> {
113        match spec {
114            StoreSpec::Memory => Ok(Self::Mem(MemoryStore::default())),
115            StoreSpec::Fs => Ok(Self::Fs(FsStore::open(data_dir.join("store"))?)),
116            #[cfg(feature = "postgres")]
117            StoreSpec::Postgres { connection_string } => Ok(Self::Postgres(
118                PostgresBlobStore::connect(connection_string).await?,
119            )),
120            #[cfg(feature = "turso")]
121            StoreSpec::Turso { path } => Ok(Self::Turso(TursoBlobStore::open(path).await?)),
122            #[cfg(feature = "s3")]
123            StoreSpec::S3 { bucket, prefix } => {
124                Ok(Self::S3(S3BlobStore::connect(bucket, prefix).await?))
125            }
126        }
127    }
128
129    /// Opens an existing storage service for peer reads without running
130    /// backend schema initialization.
131    ///
132    /// # Errors
133    /// Returns an error when the backing store cannot be opened.
134    #[allow(clippy::unused_async)]
135    pub async fn open_existing(
136        spec: &StoreSpec,
137        data_dir: &std::path::Path,
138    ) -> Result<Self, StoreError> {
139        match spec {
140            StoreSpec::Memory => Ok(Self::Mem(MemoryStore::default())),
141            StoreSpec::Fs => Ok(Self::Fs(FsStore::open(data_dir.join("store"))?)),
142            #[cfg(feature = "postgres")]
143            StoreSpec::Postgres { connection_string } => Ok(Self::Postgres(
144                PostgresBlobStore::connect_existing(connection_string).await?,
145            )),
146            #[cfg(feature = "turso")]
147            StoreSpec::Turso { path } => {
148                Ok(Self::Turso(TursoBlobStore::open_existing(path).await?))
149            }
150            #[cfg(feature = "s3")]
151            StoreSpec::S3 { bucket, prefix } => {
152                Ok(Self::S3(S3BlobStore::connect(bucket, prefix).await?))
153            }
154        }
155    }
156}
157
158#[async_trait]
159impl BlobStore for NodeStore {
160    async fn put(&self, bytes: &[u8]) -> Result<BlobId, StoreError> {
161        match self {
162            Self::Mem(store) => store.put(bytes).await,
163            Self::Fs(store) => store.put(bytes).await,
164            #[cfg(feature = "postgres")]
165            Self::Postgres(store) => store.put(bytes).await,
166            #[cfg(feature = "turso")]
167            Self::Turso(store) => store.put(bytes).await,
168            #[cfg(feature = "s3")]
169            Self::S3(store) => store.put(bytes).await,
170        }
171    }
172
173    async fn get(&self, id: &BlobId) -> Result<Option<Vec<u8>>, StoreError> {
174        match self {
175            Self::Mem(store) => store.get(id).await,
176            Self::Fs(store) => store.get(id).await,
177            #[cfg(feature = "postgres")]
178            Self::Postgres(store) => store.get(id).await,
179            #[cfg(feature = "turso")]
180            Self::Turso(store) => store.get(id).await,
181            #[cfg(feature = "s3")]
182            Self::S3(store) => store.get(id).await,
183        }
184    }
185
186    async fn contains(&self, id: &BlobId) -> Result<bool, StoreError> {
187        match self {
188            Self::Mem(store) => store.contains(id).await,
189            Self::Fs(store) => store.contains(id).await,
190            #[cfg(feature = "postgres")]
191            Self::Postgres(store) => store.contains(id).await,
192            #[cfg(feature = "turso")]
193            Self::Turso(store) => store.contains(id).await,
194            #[cfg(feature = "s3")]
195            Self::S3(store) => store.contains(id).await,
196        }
197    }
198
199    async fn delete(&self, id: &BlobId) -> Result<(), StoreError> {
200        match self {
201            Self::Mem(store) => store.delete(id).await,
202            Self::Fs(store) => store.delete(id).await,
203            #[cfg(feature = "postgres")]
204            Self::Postgres(store) => store.delete(id).await,
205            #[cfg(feature = "turso")]
206            Self::Turso(store) => store.delete(id).await,
207            #[cfg(feature = "s3")]
208            Self::S3(store) => store.delete(id).await,
209        }
210    }
211
212    async fn list(&self) -> Result<BlobIdStream, StoreError> {
213        match self {
214            Self::Mem(store) => store.list().await,
215            Self::Fs(store) => store.list().await,
216            #[cfg(feature = "postgres")]
217            Self::Postgres(store) => store.list().await,
218            #[cfg(feature = "turso")]
219            Self::Turso(store) => store.list().await,
220            #[cfg(feature = "s3")]
221            Self::S3(store) => store.list().await,
222        }
223    }
224
225    async fn modified_at(&self, id: &BlobId) -> Result<Option<SystemTime>, StoreError> {
226        match self {
227            Self::Mem(store) => store.modified_at(id).await,
228            Self::Fs(store) => store.modified_at(id).await,
229            #[cfg(feature = "postgres")]
230            Self::Postgres(store) => store.modified_at(id).await,
231            #[cfg(feature = "turso")]
232            Self::Turso(store) => store.modified_at(id).await,
233            #[cfg(feature = "s3")]
234            Self::S3(store) => store.modified_at(id).await,
235        }
236    }
237}
238
239#[async_trait]
240impl RootStore for NodeStore {
241    async fn get_root(&self, name: &str) -> Result<Option<Vec<u8>>, StoreError> {
242        match self {
243            Self::Mem(store) => store.get_root(name).await,
244            Self::Fs(store) => store.get_root(name).await,
245            #[cfg(feature = "postgres")]
246            Self::Postgres(store) => store.get_root(name).await,
247            #[cfg(feature = "turso")]
248            Self::Turso(store) => store.get_root(name).await,
249            #[cfg(feature = "s3")]
250            Self::S3(store) => store.get_root(name).await,
251        }
252    }
253
254    async fn cas_root(
255        &self,
256        name: &str,
257        expected: Option<&[u8]>,
258        new: &[u8],
259    ) -> Result<(), StoreError> {
260        match self {
261            Self::Mem(store) => store.cas_root(name, expected, new).await,
262            Self::Fs(store) => store.cas_root(name, expected, new).await,
263            #[cfg(feature = "postgres")]
264            Self::Postgres(store) => store.cas_root(name, expected, new).await,
265            #[cfg(feature = "turso")]
266            Self::Turso(store) => store.cas_root(name, expected, new).await,
267            #[cfg(feature = "s3")]
268            Self::S3(store) => store.cas_root(name, expected, new).await,
269        }
270    }
271
272    async fn delete_root(&self, name: &str) -> Result<(), StoreError> {
273        match self {
274            Self::Mem(store) => store.delete_root(name).await,
275            Self::Fs(store) => store.delete_root(name).await,
276            #[cfg(feature = "postgres")]
277            Self::Postgres(store) => store.delete_root(name).await,
278            #[cfg(feature = "turso")]
279            Self::Turso(store) => store.delete_root(name).await,
280            #[cfg(feature = "s3")]
281            Self::S3(store) => store.delete_root(name).await,
282        }
283    }
284
285    async fn list_roots(&self, prefix: &str) -> Result<Vec<String>, StoreError> {
286        match self {
287            Self::Mem(store) => store.list_roots(prefix).await,
288            Self::Fs(store) => store.list_roots(prefix).await,
289            #[cfg(feature = "postgres")]
290            Self::Postgres(store) => store.list_roots(prefix).await,
291            #[cfg(feature = "turso")]
292            Self::Turso(store) => store.list_roots(prefix).await,
293            #[cfg(feature = "s3")]
294            Self::S3(store) => store.list_roots(prefix).await,
295        }
296    }
297}
298
299/// Where a node's per-database transaction logs live.
300pub enum LogBackend {
301    /// Versioned log files under this directory.
302    Fs(PathBuf),
303    /// In-memory versioned logs shared across a process.
304    Mem(MemLogRegistry),
305    /// Versioned logs stored through the native root store.
306    #[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
307    Native(Arc<dyn NativeLogStorage>),
308}
309
310impl LogBackend {
311    /// The log backend that pairs with `spec`.
312    #[must_use]
313    #[allow(clippy::needless_pass_by_value)]
314    pub fn for_spec(
315        spec: &StoreSpec,
316        data_dir: &std::path::Path,
317        #[cfg_attr(
318            not(any(feature = "postgres", feature = "turso", feature = "s3")),
319            allow(unused_variables)
320        )]
321        store: Arc<NodeStore>,
322    ) -> Self {
323        match spec {
324            StoreSpec::Memory => Self::Mem(MemLogRegistry::new()),
325            StoreSpec::Fs => Self::Fs(data_dir.join("logs")),
326            #[cfg(feature = "postgres")]
327            StoreSpec::Postgres { .. } => Self::Native(Arc::new(NativeRootLogStore::new(store))),
328            #[cfg(feature = "turso")]
329            StoreSpec::Turso { .. } => Self::Native(Arc::new(NativeRootLogStore::new(store))),
330            #[cfg(feature = "s3")]
331            StoreSpec::S3 { .. } => Self::Native(Arc::new(NativeRootLogStore::new(store))),
332        }
333    }
334
335    /// Opens the named log for writing under `write_version`.
336    ///
337    /// # Errors
338    /// Returns an error when a filesystem log cannot be opened.
339    pub fn open(
340        &self,
341        name: &str,
342        write_version: u64,
343    ) -> Result<Arc<dyn TransactionLog>, LogError> {
344        match self {
345            Self::Fs(dir) => Ok(Arc::new(VersionedLog::open(dir, name, write_version)?)),
346            Self::Mem(registry) => Ok(Arc::new(registry.open(name, write_version))),
347            #[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
348            Self::Native(storage) => Ok(Arc::new(NativeVersionedLog::open(
349                Arc::clone(storage),
350                name,
351                write_version,
352            )?)),
353        }
354    }
355
356    /// Reports whether any log exists for `name`.
357    #[must_use]
358    pub fn exists(&self, name: &str) -> bool {
359        match self {
360            Self::Fs(dir) => VersionedLog::exists(dir, name),
361            Self::Mem(registry) => registry.exists(name),
362            #[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
363            Self::Native(storage) => storage
364                .versions(name)
365                .is_ok_and(|versions| !versions.is_empty()),
366        }
367    }
368
369    /// Deletes every log for `name`.
370    ///
371    /// # Errors
372    /// Returns an error when a filesystem log cannot be removed.
373    pub fn delete_all(&self, name: &str) -> Result<(), LogError> {
374        match self {
375            Self::Fs(dir) => VersionedLog::delete_all(dir, name),
376            Self::Mem(registry) => {
377                registry.delete_all(name);
378                Ok(())
379            }
380            #[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
381            Self::Native(storage) => storage.delete_versions(name),
382        }
383    }
384}
385
386#[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
387struct NativeRootLogStore {
388    store: Arc<NodeStore>,
389}
390
391#[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
392impl NativeRootLogStore {
393    fn new(store: Arc<NodeStore>) -> Self {
394        Self { store }
395    }
396}
397
398#[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
399impl NativeRootLogStore {
400    fn key(name: &str, version: u64) -> String {
401        format!("log:{name}:v{version:020}")
402    }
403
404    fn prefix(name: &str) -> String {
405        format!("log:{name}:v")
406    }
407
408    fn block_on<T>(
409        &self,
410        future: impl std::future::Future<Output = Result<T, StoreError>>,
411    ) -> Result<T, LogError> {
412        tokio::task::block_in_place(|| {
413            tokio::runtime::Handle::current()
414                .block_on(future)
415                .map_err(|error| LogError::Native(error.to_string()))
416        })
417    }
418}
419
420#[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
421impl NativeLogStorage for NativeRootLogStore {
422    fn read_version(&self, name: &str, version: u64) -> Result<Option<Vec<u8>>, LogError> {
423        self.block_on(self.store.get_root(&Self::key(name, version)))
424    }
425
426    fn cas_version(
427        &self,
428        name: &str,
429        version: u64,
430        expected: Option<&[u8]>,
431        new: &[u8],
432    ) -> Result<(), LogError> {
433        self.block_on(
434            self.store
435                .cas_root(&Self::key(name, version), expected, new),
436        )
437    }
438
439    fn versions(&self, name: &str) -> Result<Vec<u64>, LogError> {
440        let prefix = Self::prefix(name);
441        let names = self.block_on(self.store.list_roots(&prefix))?;
442        names
443            .into_iter()
444            .map(|key| {
445                key.strip_prefix(&prefix)
446                    .and_then(|version| version.parse::<u64>().ok())
447                    .ok_or(LogError::Corrupt)
448            })
449            .collect()
450    }
451
452    fn delete_versions(&self, name: &str) -> Result<(), LogError> {
453        for version in self.versions(name)? {
454            self.block_on(self.store.delete_root(&Self::key(name, version)))?;
455        }
456        Ok(())
457    }
458}
corium_transactor/backend.rs

corium_transactor/
backend.rs
