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`, or Turso — and [`NodeStore`]
7//! dispatches the [`BlobStore`]/[`RootStore`] operations to it. The log stays
8//! local (in-memory for `mem`, filesystem otherwise) because the commit
9//! pipeline appends to it synchronously; see
10//! `docs/design/log-and-transactor.md`.
11
12use std::fmt;
13use std::path::PathBuf;
14use std::sync::Arc;
15use std::time::SystemTime;
16
17use async_trait::async_trait;
18use corium_log::{LogError, MemLogRegistry, TransactionLog, VersionedLog};
19use corium_store::{BlobId, BlobIdStream, BlobStore, FsStore, MemoryStore, RootStore, StoreError};
20
21#[cfg(feature = "postgres")]
22use corium_store::PostgresBlobStore;
23#[cfg(feature = "turso")]
24use corium_store::TursoBlobStore;
25
26/// Selects the transactor's storage-service backend (blobs + roots).
27#[derive(Clone, Default)]
28pub enum StoreSpec {
29    /// In-memory blobs and roots; fully ephemeral and confined to one
30    /// process. The transaction log is in memory too, so the whole database
31    /// vanishes when the process exits — ideal for demos and tests.
32    Memory,
33    /// Blobs and roots under `{data_dir}/store`, log under `{data_dir}/logs`.
34    #[default]
35    Fs,
36    /// Blobs and roots in `PostgreSQL`; the transaction log stays on the local
37    /// filesystem under the data directory.
38    #[cfg(feature = "postgres")]
39    Postgres {
40        /// `PostgreSQL` URL or keyword/value connection string.
41        connection_string: String,
42    },
43    /// Blobs and roots in a Turso (embeddable `SQLite`) database at `path`;
44    /// the transaction log stays on the local filesystem under the data
45    /// directory. `path` is a local database file.
46    #[cfg(feature = "turso")]
47    Turso {
48        /// Filesystem path of the Turso database.
49        path: String,
50    },
51}
52
53impl fmt::Debug for StoreSpec {
54    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
55        match self {
56            Self::Memory => formatter.write_str("Memory"),
57            Self::Fs => formatter.write_str("Fs"),
58            #[cfg(feature = "postgres")]
59            Self::Postgres { .. } => formatter
60                .debug_struct("Postgres")
61                .field("connection_string", &"[REDACTED]")
62                .finish(),
63            #[cfg(feature = "turso")]
64            Self::Turso { path } => formatter.debug_struct("Turso").field("path", path).finish(),
65        }
66    }
67}
68
69/// The blob + root storage service a [`crate::node::TransactorNode`] runs
70/// over, chosen by [`StoreSpec`]. Dispatch is an enum rather than a trait
71/// object so every existing `impl BlobStore + RootStore` / `&dyn RootStore`
72/// call site keeps working unchanged.
73pub enum NodeStore {
74    /// In-memory backend.
75    Mem(MemoryStore),
76    /// Filesystem backend.
77    Fs(FsStore),
78    /// `PostgreSQL` backend.
79    #[cfg(feature = "postgres")]
80    Postgres(PostgresBlobStore),
81    /// Turso backend.
82    #[cfg(feature = "turso")]
83    Turso(TursoBlobStore),
84}
85
86impl NodeStore {
87    /// Opens the storage service for `spec`, relative to `data_dir` for the
88    /// filesystem backend.
89    ///
90    /// # Errors
91    /// Returns an error when the backing store cannot be opened.
92    // Only optional database-backed arms await; mem/fs are synchronous.
93    #[allow(clippy::unused_async)]
94    pub async fn open(spec: &StoreSpec, data_dir: &std::path::Path) -> Result<Self, StoreError> {
95        match spec {
96            StoreSpec::Memory => Ok(Self::Mem(MemoryStore::default())),
97            StoreSpec::Fs => Ok(Self::Fs(FsStore::open(data_dir.join("store"))?)),
98            #[cfg(feature = "postgres")]
99            StoreSpec::Postgres { connection_string } => Ok(Self::Postgres(
100                PostgresBlobStore::connect(connection_string).await?,
101            )),
102            #[cfg(feature = "turso")]
103            StoreSpec::Turso { path } => Ok(Self::Turso(TursoBlobStore::open(path).await?)),
104        }
105    }
106}
107
108#[async_trait]
109impl BlobStore for NodeStore {
110    async fn put(&self, bytes: &[u8]) -> Result<BlobId, StoreError> {
111        match self {
112            Self::Mem(store) => store.put(bytes).await,
113            Self::Fs(store) => store.put(bytes).await,
114            #[cfg(feature = "postgres")]
115            Self::Postgres(store) => store.put(bytes).await,
116            #[cfg(feature = "turso")]
117            Self::Turso(store) => store.put(bytes).await,
118        }
119    }
120
121    async fn get(&self, id: &BlobId) -> Result<Option<Vec<u8>>, StoreError> {
122        match self {
123            Self::Mem(store) => store.get(id).await,
124            Self::Fs(store) => store.get(id).await,
125            #[cfg(feature = "postgres")]
126            Self::Postgres(store) => store.get(id).await,
127            #[cfg(feature = "turso")]
128            Self::Turso(store) => store.get(id).await,
129        }
130    }
131
132    async fn contains(&self, id: &BlobId) -> Result<bool, StoreError> {
133        match self {
134            Self::Mem(store) => store.contains(id).await,
135            Self::Fs(store) => store.contains(id).await,
136            #[cfg(feature = "postgres")]
137            Self::Postgres(store) => store.contains(id).await,
138            #[cfg(feature = "turso")]
139            Self::Turso(store) => store.contains(id).await,
140        }
141    }
142
143    async fn delete(&self, id: &BlobId) -> Result<(), StoreError> {
144        match self {
145            Self::Mem(store) => store.delete(id).await,
146            Self::Fs(store) => store.delete(id).await,
147            #[cfg(feature = "postgres")]
148            Self::Postgres(store) => store.delete(id).await,
149            #[cfg(feature = "turso")]
150            Self::Turso(store) => store.delete(id).await,
151        }
152    }
153
154    async fn list(&self) -> Result<BlobIdStream, StoreError> {
155        match self {
156            Self::Mem(store) => store.list().await,
157            Self::Fs(store) => store.list().await,
158            #[cfg(feature = "postgres")]
159            Self::Postgres(store) => store.list().await,
160            #[cfg(feature = "turso")]
161            Self::Turso(store) => store.list().await,
162        }
163    }
164
165    async fn modified_at(&self, id: &BlobId) -> Result<Option<SystemTime>, StoreError> {
166        match self {
167            Self::Mem(store) => store.modified_at(id).await,
168            Self::Fs(store) => store.modified_at(id).await,
169            #[cfg(feature = "postgres")]
170            Self::Postgres(store) => store.modified_at(id).await,
171            #[cfg(feature = "turso")]
172            Self::Turso(store) => store.modified_at(id).await,
173        }
174    }
175}
176
177#[async_trait]
178impl RootStore for NodeStore {
179    async fn get_root(&self, name: &str) -> Result<Option<Vec<u8>>, StoreError> {
180        match self {
181            Self::Mem(store) => store.get_root(name).await,
182            Self::Fs(store) => store.get_root(name).await,
183            #[cfg(feature = "postgres")]
184            Self::Postgres(store) => store.get_root(name).await,
185            #[cfg(feature = "turso")]
186            Self::Turso(store) => store.get_root(name).await,
187        }
188    }
189
190    async fn cas_root(
191        &self,
192        name: &str,
193        expected: Option<&[u8]>,
194        new: &[u8],
195    ) -> Result<(), StoreError> {
196        match self {
197            Self::Mem(store) => store.cas_root(name, expected, new).await,
198            Self::Fs(store) => store.cas_root(name, expected, new).await,
199            #[cfg(feature = "postgres")]
200            Self::Postgres(store) => store.cas_root(name, expected, new).await,
201            #[cfg(feature = "turso")]
202            Self::Turso(store) => store.cas_root(name, expected, new).await,
203        }
204    }
205
206    async fn delete_root(&self, name: &str) -> Result<(), StoreError> {
207        match self {
208            Self::Mem(store) => store.delete_root(name).await,
209            Self::Fs(store) => store.delete_root(name).await,
210            #[cfg(feature = "postgres")]
211            Self::Postgres(store) => store.delete_root(name).await,
212            #[cfg(feature = "turso")]
213            Self::Turso(store) => store.delete_root(name).await,
214        }
215    }
216
217    async fn list_roots(&self, prefix: &str) -> Result<Vec<String>, StoreError> {
218        match self {
219            Self::Mem(store) => store.list_roots(prefix).await,
220            Self::Fs(store) => store.list_roots(prefix).await,
221            #[cfg(feature = "postgres")]
222            Self::Postgres(store) => store.list_roots(prefix).await,
223            #[cfg(feature = "turso")]
224            Self::Turso(store) => store.list_roots(prefix).await,
225        }
226    }
227}
228
229/// Where a node's per-database transaction logs live. The mem backend keeps
230/// them in a process-shared registry; every other backend uses versioned
231/// files under a directory, exactly as before store selection existed.
232pub enum LogBackend {
233    /// Versioned log files under this directory.
234    Fs(PathBuf),
235    /// In-memory versioned logs shared across a process.
236    Mem(MemLogRegistry),
237}
238
239impl LogBackend {
240    /// The log backend that pairs with `spec`.
241    #[must_use]
242    pub fn for_spec(spec: &StoreSpec, data_dir: &std::path::Path) -> Self {
243        match spec {
244            StoreSpec::Memory => Self::Mem(MemLogRegistry::new()),
245            StoreSpec::Fs => Self::Fs(data_dir.join("logs")),
246            #[cfg(feature = "postgres")]
247            StoreSpec::Postgres { .. } => Self::Fs(data_dir.join("logs")),
248            #[cfg(feature = "turso")]
249            StoreSpec::Turso { .. } => Self::Fs(data_dir.join("logs")),
250        }
251    }
252
253    /// Opens the named log for writing under `write_version`.
254    ///
255    /// # Errors
256    /// Returns an error when a filesystem log cannot be opened.
257    pub fn open(
258        &self,
259        name: &str,
260        write_version: u64,
261    ) -> Result<Arc<dyn TransactionLog>, LogError> {
262        match self {
263            Self::Fs(dir) => Ok(Arc::new(VersionedLog::open(dir, name, write_version)?)),
264            Self::Mem(registry) => Ok(Arc::new(registry.open(name, write_version))),
265        }
266    }
267
268    /// Deletes every log for `name`.
269    ///
270    /// # Errors
271    /// Returns an error when a filesystem log cannot be removed.
272    pub fn delete_all(&self, name: &str) -> Result<(), LogError> {
273        match self {
274            Self::Fs(dir) => VersionedLog::delete_all(dir, name),
275            Self::Mem(registry) => {
276                registry.delete_all(name);
277                Ok(())
278            }
279        }
280    }
281}
corium_transactor/backend.rs

corium_transactor/
backend.rs
