corium_transactor/
lib.rs

1//! Embedded single-writer transaction pipeline and index publisher, plus the
2//! networked transactor process (lease, gRPC services, indexing job).
3
4pub mod backend;
5pub mod backup;
6pub mod lease;
7pub mod metrics;
8pub mod node;
9pub mod server;
10
11pub use backend::{LogBackend, NodeStore, StoreSpec};
12
13use corium_core::{EntityId, IndexOrder, KeywordInterner, Partition, Schema};
14use corium_db::{Db, FIRST_USER_ID, Idents};
15use corium_index::Segment;
16use corium_log::{LogError, TransactionLog, TxRecord};
17use corium_store::{BlobStore, RootStore, StoreError};
18use corium_tx::{PreparedTx, TxError, TxItem, prepare};
19use std::{
20    sync::{Arc, Mutex, mpsc},
21    time::{SystemTime, UNIX_EPOCH},
22};
23use thiserror::Error;
24
25/// Result delivered after a transaction is durable and visible.
26#[derive(Clone, Debug)]
27pub struct TxReport {
28    /// Database before the transaction.
29    pub db_before: Db,
30    /// Database including the transaction.
31    pub db_after: Db,
32    /// Prepared transaction and tempid map.
33    pub tx: PreparedTx,
34    /// Commit timestamp.
35    pub tx_instant: i64,
36}
37
38/// Pipeline errors.
39#[derive(Debug, Error)]
40pub enum TransactError {
41    /// Transaction rejected before durability.
42    #[error(transparent)]
43    Tx(#[from] TxError),
44    /// Durable log failed.
45    #[error(transparent)]
46    Log(#[from] LogError),
47    /// Index/root store failed.
48    #[error(transparent)]
49    Store(#[from] StoreError),
50    /// System clock predates the Unix epoch.
51    #[error("system clock is before Unix epoch")]
52    Clock,
53    /// An index-building worker failed before returning its result.
54    #[error("index task failed: {0}")]
55    IndexTask(String),
56    /// A newer lease version owns the database root; this writer is deposed.
57    #[error("deposed: database root is owned by lease version {published}")]
58    Deposed {
59        /// Lease version found on the published root.
60        published: u64,
61    },
62}
63
64struct State {
65    db: Db,
66    next_user: u64,
67    last_instant: i64,
68    subscribers: Vec<mpsc::Sender<TxReport>>,
69}
70
71/// A serialized, in-process transactor. The log append is the commit point.
72pub struct EmbeddedTransactor {
73    log: Arc<dyn TransactionLog>,
74    state: Mutex<State>,
75}
76impl EmbeddedTransactor {
77    /// Recovers a transactor by replaying the durable log exactly once.
78    ///
79    /// # Errors
80    /// Returns an error when the durable log cannot be replayed.
81    pub fn recover(schema: Schema, log: Arc<dyn TransactionLog>) -> Result<Self, TransactError> {
82        Self::recover_from(Db::new(schema), log)
83    }
84
85    /// Recovers from an empty base database value (schema plus naming) by
86    /// replaying the durable log exactly once.
87    ///
88    /// # Errors
89    /// Returns an error when the durable log cannot be replayed.
90    pub fn recover_from(base: Db, log: Arc<dyn TransactionLog>) -> Result<Self, TransactError> {
91        let mut db = base;
92        let mut last_instant = i64::MIN;
93        for record in log.replay()? {
94            db = db.with_transaction(record.t, &record.datoms);
95            last_instant = last_instant.max(record.tx_instant);
96        }
97        // Allocation must resume past every id that ever appeared in the log,
98        // not just ids with current datoms; otherwise a fully retracted
99        // entity's id would be reused after a restart.
100        let next_user = db
101            .recorded_datoms()
102            .iter()
103            .filter(|d| d.e.partition() == Partition::User as u32)
104            .map(|d| d.e.sequence() + 1)
105            .max()
106            .unwrap_or(FIRST_USER_ID);
107        Ok(Self {
108            log,
109            state: Mutex::new(State {
110                db,
111                next_user,
112                last_instant,
113                subscribers: Vec::new(),
114            }),
115        })
116    }
117    /// Returns the current immutable database value.
118    #[must_use]
119    pub fn db(&self) -> Db {
120        self.state
121            .lock()
122            .unwrap_or_else(std::sync::PoisonError::into_inner)
123            .db
124            .clone()
125    }
126    /// Subscribes to reports for transactions committed after this call.
127    pub fn subscribe(&self) -> mpsc::Receiver<TxReport> {
128        let (tx, rx) = mpsc::channel();
129        self.state
130            .lock()
131            .unwrap_or_else(std::sync::PoisonError::into_inner)
132            .subscribers
133            .push(tx);
134        rx
135    }
136    /// Validates, durably appends, applies, and reports a transaction.
137    ///
138    /// # Errors
139    /// Returns an error for rejected transaction data, clock failure, or when
140    /// the durable append fails. No report is sent on error.
141    pub fn transact(
142        &self,
143        items: impl IntoIterator<Item = TxItem>,
144    ) -> Result<TxReport, TransactError> {
145        let mut state = self
146            .state
147            .lock()
148            .unwrap_or_else(std::sync::PoisonError::into_inner);
149        let before = state.db.clone();
150        let t = before.basis_t() + 1;
151        let tx_id = EntityId::new(Partition::Tx as u32, t);
152        let prepared = prepare(&before, items, tx_id, state.next_user)?;
153        let millis = i64::try_from(
154            SystemTime::now()
155                .duration_since(UNIX_EPOCH)
156                .map_err(|_| TransactError::Clock)?
157                .as_millis(),
158        )
159        .unwrap_or(i64::MAX);
160        let tx_instant = millis.max(state.last_instant.saturating_add(1));
161        self.log.append(&TxRecord {
162            t,
163            tx_instant,
164            datoms: prepared.datoms.clone(),
165        })?;
166        state.db = before.with_transaction(t, &prepared.datoms);
167        state.last_instant = tx_instant;
168        state.next_user = prepared
169            .tempids
170            .values()
171            .filter(|e| e.partition() == Partition::User as u32)
172            .map(|e| e.sequence() + 1)
173            .max()
174            .unwrap_or(state.next_user)
175            .max(state.next_user);
176        let report = TxReport {
177            db_before: before,
178            db_after: state.db.clone(),
179            tx: prepared,
180            tx_instant,
181        };
182        state
183            .subscribers
184            .retain(|subscriber| subscriber.send(report.clone()).is_ok());
185        Ok(report)
186    }
187    /// Replaces the ident/keyword naming attached to the current database
188    /// value (used when the boundary interns new keywords).
189    pub fn update_naming(&self, idents: Idents, interner: KeywordInterner) {
190        let mut state = self
191            .state
192            .lock()
193            .unwrap_or_else(std::sync::PoisonError::into_inner);
194        state.db = state.db.clone().with_naming(idents, interner);
195    }
196
197    /// Builds a consistent snapshot of all four indexes and publishes their blob ids.
198    ///
199    /// Blobs are uploaded before the root CAS. Transactions may continue while the
200    /// immutable snapshot is encoded; a later run indexes any remaining log tail.
201    ///
202    /// Publication is fenced by `lease_version` and monotone in
203    /// `index_basis_t`: a root already published under a newer lease version
204    /// deposes this writer ([`TransactError::Deposed`]); a root at an equal
205    /// or newer basis (or one that wins a concurrent CAS race) leaves this
206    /// snapshot's blobs for garbage collection. The freshly built root is
207    /// returned when it, or a newer basis, is installed.
208    ///
209    /// # Errors
210    /// Returns an error if a blob upload, root read, or fenced publication fails.
211    pub async fn publish_indexes(
212        &self,
213        store: &(impl BlobStore + RootStore),
214        root_name: &str,
215        lease_version: u64,
216    ) -> Result<DbRoot, TransactError> {
217        let snapshot = self.db();
218        let datoms = snapshot.datoms();
219        let segments = tokio::task::spawn_blocking(move || {
220            [
221                IndexOrder::Eavt,
222                IndexOrder::Aevt,
223                IndexOrder::Avet,
224                IndexOrder::Vaet,
225            ]
226            .into_iter()
227            .map(|order| {
228                let segment = Segment::build(order, datoms.clone());
229                let mut bytes = Vec::new();
230                for (key, _) in segment.entries() {
231                    bytes.extend_from_slice(&(key.len() as u64).to_be_bytes());
232                    bytes.extend_from_slice(key);
233                }
234                bytes
235            })
236            .collect::<Vec<_>>()
237        })
238        .await
239        .map_err(|error| TransactError::IndexTask(error.to_string()))?;
240        let mut ids = Vec::new();
241        for bytes in segments {
242            ids.push(store.put(&bytes).await?);
243        }
244        let root = DbRoot {
245            format_version: corium_store::FORMAT_VERSION,
246            lease_version,
247            owner: String::new(),
248            lease_expires_unix_ms: 0,
249            owner_endpoint: String::new(),
250            index_basis_t: snapshot.basis_t(),
251            roots: Some([
252                ids[0].clone(),
253                ids[1].clone(),
254                ids[2].clone(),
255                ids[3].clone(),
256            ]),
257        };
258        publish_root(store, root_name, &root).await?;
259        Ok(root)
260    }
261}
262
263/// Publishes `root` under the fencing rules described on
264/// [`EmbeddedTransactor::publish_indexes`].
265///
266/// # Errors
267/// Returns [`TransactError::Deposed`] when a newer lease version owns the
268/// root, or a store error when the CAS cannot be completed.
269pub async fn publish_root(
270    store: &dyn RootStore,
271    root_name: &str,
272    root: &DbRoot,
273) -> Result<(), TransactError> {
274    loop {
275        let previous = store.get_root(root_name).await?;
276        let stored = previous.as_deref().and_then(DbRoot::decode);
277        let mut next = root.clone();
278        if let Some(stored) = stored {
279            if stored.lease_version > root.lease_version {
280                return Err(TransactError::Deposed {
281                    published: stored.lease_version,
282                });
283            }
284            if stored.lease_version == root.lease_version
285                && stored.index_basis_t >= root.index_basis_t
286            {
287                return Ok(());
288            }
289            // The stored record carries the live lease fields (renewals CAS
290            // the same key); publication must not clobber them.
291            if stored.lease_version == root.lease_version {
292                next.owner = stored.owner;
293                next.lease_expires_unix_ms = stored.lease_expires_unix_ms;
294                next.owner_endpoint = stored.owner_endpoint;
295            }
296        }
297        match store
298            .cas_root(root_name, previous.as_deref(), &next.encode())
299            .await
300        {
301            Ok(()) => return Ok(()),
302            Err(StoreError::CasFailed { .. }) => {}
303            Err(error) => return Err(error.into()),
304        }
305    }
306}
307
308pub use corium_store::{DbRoot, db_root_name};
corium_transactor/lib.rs

corium_transactor/
lib.rs
