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

corium_transactor/
lib.rs
