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, TxRecord, 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
310/// Runs filesystem log operations on Tokio's blocking pool while exposing the
311/// same async interface used by native storage logs.
312struct BlockingTransactionLog(Arc<dyn TransactionLog>);
313
314#[async_trait]
315impl TransactionLog for BlockingTransactionLog {
316    fn append(&self, record: &TxRecord) -> Result<(), LogError> {
317        self.0.append(record)
318    }
319
320    async fn append_async(&self, record: &TxRecord) -> Result<(), LogError> {
321        let log = Arc::clone(&self.0);
322        let record = record.clone();
323        tokio::task::spawn_blocking(move || log.append(&record))
324            .await
325            .map_err(|error| LogError::Native(format!("log task failed: {error}")))?
326    }
327
328    fn tx_range(&self, start: u64, end: Option<u64>) -> Result<Vec<TxRecord>, LogError> {
329        self.0.tx_range(start, end)
330    }
331
332    async fn tx_range_async(
333        &self,
334        start: u64,
335        end: Option<u64>,
336    ) -> Result<Vec<TxRecord>, LogError> {
337        let log = Arc::clone(&self.0);
338        tokio::task::spawn_blocking(move || log.tx_range(start, end))
339            .await
340            .map_err(|error| LogError::Native(format!("log task failed: {error}")))?
341    }
342}
343
344impl LogBackend {
345    /// The log backend that pairs with `spec`.
346    #[must_use]
347    #[allow(clippy::needless_pass_by_value)]
348    pub fn for_spec(
349        spec: &StoreSpec,
350        data_dir: &std::path::Path,
351        #[cfg_attr(
352            not(any(feature = "postgres", feature = "turso", feature = "s3")),
353            allow(unused_variables)
354        )]
355        store: Arc<NodeStore>,
356    ) -> Self {
357        match spec {
358            StoreSpec::Memory => Self::Mem(MemLogRegistry::new()),
359            StoreSpec::Fs => Self::Fs(data_dir.join("logs")),
360            #[cfg(feature = "postgres")]
361            StoreSpec::Postgres { .. } => Self::Native(Arc::new(NativeRootLogStore::new(store))),
362            #[cfg(feature = "turso")]
363            StoreSpec::Turso { .. } => Self::Native(Arc::new(NativeRootLogStore::new(store))),
364            #[cfg(feature = "s3")]
365            StoreSpec::S3 { .. } => Self::Native(Arc::new(NativeRootLogStore::new(store))),
366        }
367    }
368
369    /// Opens the named log for writing under `write_version`.
370    ///
371    /// # Errors
372    /// Returns an error when a transaction log cannot be opened.
373    pub async fn open(
374        &self,
375        name: &str,
376        write_version: u64,
377    ) -> Result<Arc<dyn TransactionLog>, LogError> {
378        match self {
379            Self::Fs(dir) => {
380                let dir = dir.clone();
381                let name = name.to_owned();
382                let log = tokio::task::spawn_blocking(move || {
383                    VersionedLog::open(dir, &name, write_version)
384                })
385                .await
386                .map_err(|error| LogError::Native(format!("log task failed: {error}")))??;
387                Ok(Arc::new(BlockingTransactionLog(Arc::new(log))))
388            }
389            Self::Mem(registry) => Ok(Arc::new(registry.open(name, write_version))),
390            #[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
391            Self::Native(storage) => Ok(Arc::new(
392                NativeVersionedLog::open(Arc::clone(storage), name, write_version).await?,
393            )),
394        }
395    }
396
397    /// Reports whether any log exists for `name`.
398    #[must_use]
399    pub async fn exists(&self, name: &str) -> bool {
400        match self {
401            Self::Fs(dir) => {
402                let dir = dir.clone();
403                let name = name.to_owned();
404                tokio::task::spawn_blocking(move || VersionedLog::exists(dir, &name))
405                    .await
406                    .unwrap_or(false)
407            }
408            Self::Mem(registry) => registry.exists(name),
409            #[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
410            Self::Native(storage) => storage
411                .list_chunks(name)
412                .await
413                .is_ok_and(|chunks| !chunks.is_empty()),
414        }
415    }
416
417    /// Deletes every log for `name`.
418    ///
419    /// # Errors
420    /// Returns an error when a transaction log cannot be removed.
421    pub async fn delete_all(&self, name: &str) -> Result<(), LogError> {
422        match self {
423            Self::Fs(dir) => {
424                let dir = dir.clone();
425                let name = name.to_owned();
426                tokio::task::spawn_blocking(move || VersionedLog::delete_all(dir, &name))
427                    .await
428                    .map_err(|error| LogError::Native(format!("log task failed: {error}")))?
429            }
430            Self::Mem(registry) => {
431                registry.delete_all(name);
432                Ok(())
433            }
434            #[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
435            Self::Native(storage) => storage.delete_all(name).await,
436        }
437    }
438}
439
440#[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
441struct NativeRootLogStore {
442    store: Arc<NodeStore>,
443}
444
445#[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
446impl NativeRootLogStore {
447    fn new(store: Arc<NodeStore>) -> Self {
448        Self { store }
449    }
450}
451
452#[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
453impl NativeRootLogStore {
454    /// Object key for one log chunk. Chunk `0` keeps the historical
455    /// unsuffixed key, so logs written by earlier releases read back as their
456    /// chunk `0` without migration.
457    fn key(name: &str, version: u64, chunk: u64) -> String {
458        if chunk == 0 {
459            format!("log:{name}:v{version:020}")
460        } else {
461            format!("log:{name}:v{version:020}:c{chunk:020}")
462        }
463    }
464
465    fn prefix(name: &str) -> String {
466        format!("log:{name}:v")
467    }
468
469    /// Parses a `(version, chunk)` pair out of a listed log key. Chunk `0`
470    /// keys carry no `:c` suffix (see [`Self::key`]).
471    fn parse_key(prefix: &str, key: &str) -> Option<(u64, u64)> {
472        let rest = key.strip_prefix(prefix)?;
473        match rest.split_once(":c") {
474            Some((version, chunk)) => Some((version.parse().ok()?, chunk.parse().ok()?)),
475            None => Some((rest.parse().ok()?, 0)),
476        }
477    }
478}
479
480#[cfg(any(feature = "postgres", feature = "turso", feature = "s3"))]
481#[async_trait]
482impl NativeLogStorage for NativeRootLogStore {
483    async fn read_chunk(
484        &self,
485        name: &str,
486        version: u64,
487        chunk: u64,
488    ) -> Result<Option<Vec<u8>>, LogError> {
489        self.store
490            .get_root(&Self::key(name, version, chunk))
491            .await
492            .map_err(|error| LogError::Native(error.to_string()))
493    }
494
495    async fn cas_chunk(
496        &self,
497        name: &str,
498        version: u64,
499        chunk: u64,
500        expected: Option<&[u8]>,
501        new: &[u8],
502    ) -> Result<(), LogError> {
503        self.store
504            .cas_root(&Self::key(name, version, chunk), expected, new)
505            .await
506            .map_err(|error| LogError::Native(error.to_string()))
507    }
508
509    async fn list_chunks(&self, name: &str) -> Result<Vec<(u64, u64)>, LogError> {
510        let prefix = Self::prefix(name);
511        let names = self
512            .store
513            .list_roots(&prefix)
514            .await
515            .map_err(|error| LogError::Native(error.to_string()))?;
516        names
517            .into_iter()
518            .map(|key| Self::parse_key(&prefix, &key).ok_or(LogError::Corrupt))
519            .collect()
520    }
521
522    async fn delete_all(&self, name: &str) -> Result<(), LogError> {
523        for (version, chunk) in self.list_chunks(name).await? {
524            self.store
525                .delete_root(&Self::key(name, version, chunk))
526                .await
527                .map_err(|error| LogError::Native(error.to_string()))?;
528        }
529        Ok(())
530    }
531}
532
533#[cfg(all(test, any(feature = "postgres", feature = "turso", feature = "s3")))]
534mod tests {
535    use super::*;
536    use corium_core::{Datom, EntityId, Value};
537
538    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
539    async fn native_root_log_crosses_chunk_boundaries_on_the_runtime() {
540        let run = async {
541            let dir = tempfile::tempdir().expect("tempdir");
542            let store = Arc::new(
543                NodeStore::open(&StoreSpec::Memory, dir.path())
544                    .await
545                    .expect("memory store"),
546            );
547            let storage = Arc::new(NativeRootLogStore::new(store));
548            let log = NativeVersionedLog::open(Arc::clone(&storage), "db", 1)
549                .await
550                .expect("open native log");
551            for t in 1..=3 {
552                log.append_async(&TxRecord {
553                    t,
554                    tx_instant: i64::try_from(t).expect("small t"),
555                    datoms: vec![Datom {
556                        e: EntityId::from_raw(t),
557                        a: EntityId::from_raw(1),
558                        v: Value::Bytes(vec![0; 300 * 1024].into()),
559                        tx: EntityId::from_raw(t),
560                        added: true,
561                    }],
562                })
563                .await
564                .expect("append");
565            }
566            let chunks = storage.list_chunks("db").await.expect("list chunks");
567            assert_eq!(chunks.len(), 3);
568            assert_eq!(log.replay_async().await.expect("replay").len(), 3);
569        };
570        tokio::time::timeout(std::time::Duration::from_secs(5), run)
571            .await
572            .expect("native log operation stalled on its runtime");
573    }
574}
corium_transactor/backend.rs

corium_transactor/
backend.rs
