ubiquisync_sql/
tracker.rs

1//! Recording the log entries a processor ingests.
2//!
3//! [`LogTracker`] is the hook the processor calls for every ingested entry,
4//! enqueued into the same batch that applies the entry. Implementations decide
5//! what to record. [`LogIndexTracker`] is the built-in one — an op-log table
6//! keyed by `(peer_id, entry_idx)` that future readers can dedup, revision
7//! history, and attribution from without replaying the reducer.
8//!
9//! Readers derive from the table without replaying the reducer —
10//! [`all_cursors`](LogTracker::all_cursors) for the version vector,
11//! [`HistoryTracker::read_entries`] to reconstruct entries. The expunged-marker
12//! rows [`track_expunged`](LogTracker::track_expunged) writes (`tag` =
13//! [`TAG_EXPUNGED`], no attribution, no timestamp) occupy a stream index but are
14//! not real entries, so a reader must not count or attribute them.
15
16use std::marker::PhantomData;
17
18use ubiquisync_core::{
19    codec::{CodecError, DecodedEntry, IndexableOp, TAG_EXPUNGED},
20    hlc::Timestamp,
21    log_entry::LogEntry,
22    sync::PeerCursors,
23    uuid::Uuid,
24};
25
26use crate::{
27    db::{Db, DbBatch, DbError, DbType, DbValue, ValueBinder},
28    util::quote_ident,
29};
30
31/// Records each log entry as the processor ingests it.
32///
33/// The processor calls [`init`](LogTracker::init) once at startup, then
34/// [`track_one`](LogTracker::track_one) for every ingested entry (or
35/// [`track_expunged`](LogTracker::track_expunged) for an expunged marker),
36/// inside the batch that applies that entry. What gets recorded, and how, is up
37/// to the implementation — see [`LogIndexTracker`] for the built-in op-log.
38///
39/// # Duplicate rejection
40///
41/// An implementation may or may not reject a repeated `(peer_id, entry_idx)`.
42/// If it does — e.g. by making `(peer_id, entry_idx)` a unique key, so a
43/// repeat fails the batch and rolls the whole apply back — then the reducer it
44/// is paired with need not be idempotent. If it does not, the reducer must be.
45/// Nothing yet enforces that pairing at the type level (a future marker on the
46/// reducer could); today it is a contract the wiring must honor.
47#[async_trait::async_trait]
48pub trait LogTracker<Op>: Sized + Send + Sync {
49    /// Initialize this tracker's backing state, namespaced by `prefix`, and
50    /// return an instance bound to it.
51    async fn init(db: &dyn Db, prefix: &str) -> Result<Self, DbError>;
52
53    /// Record `entry`, identified by `(peer_id, entry_idx)`, by enqueuing its
54    /// writes into `batch` so they commit together with the rest of the entry's
55    /// application. Must not commit on its own.
56    fn track_one(
57        &self,
58        peer_id: &Uuid,
59        entry_idx: u64,
60        timestamp: Timestamp,
61        server_user_id: Option<Uuid>,
62        op: &Op,
63        batch: &mut dyn DbBatch,
64    ) -> Result<(), LogTrackerError>;
65
66    /// Record the expunged marker at `(peer_id, entry_idx)`, naming the `hash`
67    /// of the entry that was expunged, by enqueuing into `batch`. This occupies
68    /// the stream index so the peer's cursor advances past the gap without a
69    /// real entry ever being applied there. Must not commit on its own.
70    fn track_expunged(
71        &self,
72        peer_id: &Uuid,
73        entry_idx: u64,
74        hash: &blake3::Hash,
75        batch: &mut dyn DbBatch,
76    ) -> Result<(), LogTrackerError>;
77
78    /// Every peer's cursor as a version vector; seeds a processor's in-memory
79    /// cursor view at open.
80    async fn all_cursors(&self, db: &dyn Db) -> Result<PeerCursors, DbError>;
81}
82
83/// A [`LogTracker`] that also reconstructs stored entries — the op-log case,
84/// which keeps full history and so can back a replication
85/// [`LogSource`](ubiquisync_core::sync::LogSource).
86#[async_trait::async_trait]
87pub trait HistoryTracker<Op>: LogTracker<Op> {
88    /// Up to `limit` of `peer_id`'s entries at or after `from`, ascending
89    /// (expunged markers included), reconstructed from stored rows.
90    async fn read_entries(
91        &self,
92        db: &dyn Db,
93        peer_id: &Uuid,
94        from: u64,
95        limit: u64,
96    ) -> Result<Vec<(u64, DecodedEntry<Op>)>, LogTrackerError>;
97}
98
99/// Failure from [`LogTracker::track_one`]. Splitting the op into its index
100/// triple is a [`CodecError`]; binding a value that won't fit the backend's
101/// signed-integer column (a `u64` past `i64::MAX`, via [`DbValue::from_u64`])
102/// is a [`DbError`]. Both are surfaced so the tracker can use the same checked
103/// conversion the rest of the crate relies on instead of a lossy `as` cast.
104#[derive(Debug, thiserror::Error)]
105pub enum LogTrackerError {
106    /// Splitting the op into its index triple failed.
107    #[error("codec error: {0}")]
108    Codec(#[from] CodecError),
109    /// A backend operation failed — including a `u64` that won't fit the
110    /// signed-integer column.
111    #[error("db error: {0}")]
112    Db(#[from] DbError),
113}
114
115/// The default [`LogTracker`]: writes each entry into a single
116/// `<prefix>__oplog` table keyed by `(peer_id, entry_idx)`, storing the HLC
117/// timestamp and the op's decoded index key/value.
118///
119/// The `(peer_id, entry_idx)` primary key rejects a repeated index with a
120/// unique violation, rolling the whole apply back — so a reducer driven through
121/// this tracker is not required to be idempotent (see the
122/// [duplicate-rejection contract](LogTracker#duplicate-rejection)).
123pub struct LogIndexTracker<Op> {
124    quoted_table_name: String,
125    _phantom: PhantomData<Op>,
126}
127
128#[async_trait::async_trait]
129impl<Op: IndexableOp + Send + Sync> LogTracker<Op> for LogIndexTracker<Op> {
130    async fn init(db: &dyn Db, prefix: &str) -> Result<Self, DbError> {
131        let quoted_table_name = quote_ident(&format!("{prefix}__oplog"));
132        let dialect = db.dialect();
133        let int_type = DbType::Integer.sql_type(dialect);
134        let blob_type = DbType::Blob.sql_type(dialect);
135        let uuid_type = DbType::Uuid.sql_type(dialect);
136        let without_rowid = dialect.without_rowid();
137        let sql = format!(
138            "CREATE TABLE IF NOT EXISTS {quoted_table_name} (\
139            peer_id {uuid_type} NOT NULL,\
140            entry_idx {int_type} NOT NULL,\
141            server_user_id {uuid_type} NULL,\
142            ts {int_type} NOT NULL,\
143            tag {int_type} NOT NULL,\
144            index_key {blob_type} NULL,\
145            index_value {blob_type} NULL,\
146            PRIMARY KEY(peer_id, entry_idx))\
147            {without_rowid};"
148        );
149        db.exec(&sql, &[]).await?;
150        Ok(Self {
151            quoted_table_name,
152            _phantom: Default::default(),
153        })
154    }
155
156    fn track_one(
157        &self,
158        peer_id: &Uuid,
159        entry_idx: u64,
160        timestamp: Timestamp,
161        server_user_id: Option<Uuid>,
162        op: &Op,
163        batch: &mut dyn DbBatch,
164    ) -> Result<(), LogTrackerError> {
165        let mut value_binder = ValueBinder::new(batch.dialect());
166
167        let peer_id_bind = value_binder.bind_next(DbValue::Uuid(*peer_id));
168        // `entry_idx` is a u64 stream counter going into a signed-integer
169        // column. `from_u64` rejects a value past `i64::MAX` rather than wrap it
170        // negative — the same checked store the rest of the crate uses.
171        let entry_idx_bind = value_binder.bind_next(DbValue::from_u64(entry_idx)?);
172        let server_user_id_bind = if let Some(server_user_id) = server_user_id {
173            value_binder.bind_next(DbValue::Uuid(server_user_id))
174        } else {
175            value_binder.bind_next(DbValue::Null)
176        };
177        // The packed HLC timestamp is a full-width u64; store it through the same
178        // `from_u64` guard `SqlHlcStorage` uses, so a value past `i64::MAX` can't
179        // wrap negative and misorder a `MAX`/`GREATEST` merge on `ts`.
180        let ts_bind = value_binder.bind_next(DbValue::from_u64(timestamp.raw())?);
181
182        let index_entry = op.to_index_entry()?;
183        let tag_bind = value_binder.bind_next(DbValue::Integer(index_entry.tag as i64));
184        let index_key_bind = value_binder.bind_next(DbValue::Blob(index_entry.key));
185        let value_bind = value_binder.bind_next(DbValue::Blob(index_entry.value));
186
187        let sql = format!(
188            "INSERT INTO {} (\"peer_id\", \"entry_idx\", \"server_user_id\", \"ts\", \"tag\", \"index_key\", \"index_value\") \
189             VALUES({peer_id_bind}, {entry_idx_bind}, {server_user_id_bind}, {ts_bind}, {tag_bind}, {index_key_bind}, {value_bind})",
190            self.quoted_table_name
191        );
192
193        batch.add_statement(&sql, &value_binder.values());
194        Ok(())
195    }
196
197    async fn all_cursors(&self, db: &dyn Db) -> Result<PeerCursors, DbError> {
198        let sql = format!(
199            "SELECT \"peer_id\", MAX(\"entry_idx\") FROM {} GROUP BY \"peer_id\"",
200            self.quoted_table_name
201        );
202        let rows = db.query(&sql, &[]).await?;
203        let mut cursors = PeerCursors::new();
204        for row in &rows {
205            // MAX over a non-empty group is never NULL: the cursor is one past
206            // the highest recorded index for that peer.
207            cursors.insert(row.get_uuid(0)?, row.get_u64(1)? + 1);
208        }
209        Ok(cursors)
210    }
211
212    fn track_expunged(
213        &self,
214        peer_id: &Uuid,
215        entry_idx: u64,
216        hash: &blake3::Hash,
217        batch: &mut dyn DbBatch,
218    ) -> Result<(), LogTrackerError> {
219        let mut value_binder = ValueBinder::new(batch.dialect());
220
221        let peer_id_bind = value_binder.bind_next(DbValue::Uuid(*peer_id));
222        let entry_idx_bind = value_binder.bind_next(DbValue::from_u64(entry_idx)?);
223        // An expunged marker carries no attribution and no timestamp (see
224        // `TAG_EXPUNGED`); the row exists only to occupy the stream index. Store
225        // NULL attribution and ts 0, and stash the expunged entry's hash in
226        // `index_value` under the reserved tag so `index_key` stays NULL.
227        let tag_bind = value_binder.bind_next(DbValue::Integer(TAG_EXPUNGED as i64));
228        let hash_bind = value_binder.bind_next(DbValue::Blob(hash.as_bytes().to_vec()));
229
230        let sql = format!(
231            "INSERT INTO {} (\"peer_id\", \"entry_idx\", \"server_user_id\", \"ts\", \"tag\", \"index_key\", \"index_value\") \
232             VALUES({peer_id_bind}, {entry_idx_bind}, NULL, 0, {tag_bind}, NULL, {hash_bind})",
233            self.quoted_table_name
234        );
235
236        batch.add_statement(&sql, &value_binder.values());
237        Ok(())
238    }
239}
240
241#[async_trait::async_trait]
242impl<Op: IndexableOp + Send + Sync> HistoryTracker<Op> for LogIndexTracker<Op> {
243    async fn read_entries(
244        &self,
245        db: &dyn Db,
246        peer_id: &Uuid,
247        from: u64,
248        limit: u64,
249    ) -> Result<Vec<(u64, DecodedEntry<Op>)>, LogTrackerError> {
250        let mut binder = ValueBinder::new(db.dialect());
251        let peer_bind = binder.bind_next(DbValue::Uuid(*peer_id));
252        let from_bind = binder.bind_next(DbValue::from_u64(from)?);
253        let limit_bind = binder.bind_next(DbValue::from_u64(limit)?);
254        let sql = format!(
255            "SELECT \"entry_idx\", \"server_user_id\", \"ts\", \"tag\", \"index_key\", \"index_value\" \
256             FROM {} WHERE \"peer_id\" = {peer_bind} AND \"entry_idx\" >= {from_bind} \
257             ORDER BY \"entry_idx\" ASC LIMIT {limit_bind}",
258            self.quoted_table_name
259        );
260        let rows = db.query(&sql, &binder.values()).await?;
261
262        let mut out = Vec::with_capacity(rows.len());
263        for row in &rows {
264            let idx = row.get_u64(0)?;
265            let tag = u8::try_from(row.get_i64(3)?).map_err(|_| DbError::TypeMismatch {
266                col: 3,
267                expected: "u8 tag",
268            })?;
269            let decoded = if tag == TAG_EXPUNGED {
270                // The marker stashed the expunged entry's 32-byte hash in
271                // index_value (see `track_expunged`).
272                let bytes: [u8; 32] =
273                    row.get_blob(5)?
274                        .try_into()
275                        .map_err(|_| DbError::TypeMismatch {
276                            col: 5,
277                            expected: "32-byte hash",
278                        })?;
279                DecodedEntry::Expunged(blake3::Hash::from_bytes(bytes))
280            } else {
281                let op = Op::from_index_parts(
282                    tag,
283                    row.get_optional_blob(4)?.unwrap_or_default(),
284                    row.get_optional_blob(5)?.unwrap_or_default(),
285                )?;
286                DecodedEntry::LogEntry(LogEntry {
287                    server_user_id: row.get_optional_uuid(1)?,
288                    timestamp: Timestamp::from_raw(row.get_u64(2)?),
289                    op,
290                })
291            };
292            out.push((idx, decoded));
293        }
294        Ok(out)
295    }
296}
ubiquisync_sql/tracker.rs

ubiquisync_sql/
tracker.rs
