pulsify_sqlite/
store.rs

1use crate::{Error, Result};
2use pulsify_protocol::{
3    validate_vector, PatchRow, VersionVector, MAX_MESSAGE_ROWS, MAX_PATCH_BYTES, MAX_SAFE_INTEGER,
4    MAX_VECTOR_ENTRIES,
5};
6use rusqlite::{params, Connection, OptionalExtension, Transaction, TransactionBehavior};
7use std::collections::{BTreeMap, BTreeSet};
8use std::path::Path;
9use std::time::Duration;
10
11/// Byte-for-byte `SQLITE_PATCH_SCHEMA` from `@pulsify/core/storage`.
12pub const SCHEMA: &str = "
13CREATE TABLE IF NOT EXISTS pulsify_op_rows (
14 doc_id TEXT NOT NULL, sid INTEGER NOT NULL, time_start INTEGER NOT NULL,
15 span INTEGER NOT NULL CHECK(span > 0), time_end INTEGER GENERATED ALWAYS AS (time_start + span) STORED,
16 dependencies TEXT NOT NULL, operations BLOB NOT NULL, ready INTEGER NOT NULL DEFAULT 0,
17 PRIMARY KEY(doc_id, sid, time_start)
18) WITHOUT ROWID;
19CREATE INDEX IF NOT EXISTS pulsify_rows_frontier ON pulsify_op_rows(doc_id, sid, time_end);
20CREATE INDEX IF NOT EXISTS pulsify_rows_pending ON pulsify_op_rows(doc_id) WHERE ready = 0;
21CREATE TABLE IF NOT EXISTS pulsify_doc_clocks (
22 doc_id TEXT NOT NULL, sid INTEGER NOT NULL, clock INTEGER NOT NULL,
23 PRIMARY KEY(doc_id, sid)
24) WITHOUT ROWID;
25CREATE VIEW IF NOT EXISTS pulsify_dangling_op_rows AS SELECT * FROM pulsify_op_rows WHERE ready = 0;
26";
27
28#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
29pub struct AppendResult {
30    pub inserted: usize,
31    pub ready: usize,
32    pub pending: usize,
33}
34
35/// Immutable rows keyed by `(doc_id, sid, time_start)`. Rows stay pending until
36/// their actor prefix is contiguous and their dependencies are stored.
37pub struct SqliteStore {
38    db: Connection,
39}
40
41impl SqliteStore {
42    /// Opens or creates a durable database file (WAL, synchronous FULL).
43    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
44        let db = Connection::open(path)?;
45        db.busy_timeout(Duration::from_secs(5))?;
46        db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?;
47        Self::new(db)
48    }
49    pub fn open_in_memory() -> Result<Self> {
50        Self::new(Connection::open_in_memory()?)
51    }
52    /// Installs the schema on an existing connection.
53    pub fn new(db: Connection) -> Result<Self> {
54        db.execute_batch(SCHEMA)?;
55        Ok(Self { db })
56    }
57    pub fn connection(&self) -> &Connection {
58        &self.db
59    }
60
61    /// Validates and stores rows of one document in one IMMEDIATE transaction.
62    /// Exact retransmissions are ignored; a conflict rolls back the whole batch.
63    pub fn append(&mut self, rows: &[PatchRow]) -> Result<AppendResult> {
64        let Some(doc_id) = validate_rows(rows)? else {
65            return Ok(AppendResult::default());
66        };
67        let tx = self
68            .db
69            .transaction_with_behavior(TransactionBehavior::Immediate)?;
70        let result = append_rows(&tx, doc_id, rows)?;
71        tx.commit()?;
72        Ok(result)
73    }
74
75    pub fn vector(&self, doc_id: &str) -> Result<VersionVector> {
76        validate_doc_id(doc_id)?;
77        read_vector(&self.db, doc_id)
78    }
79
80    /// Ready rows the peer lacks, ordered by actor and clock.
81    pub fn missing(&self, doc_id: &str, remote: &VersionVector) -> Result<Vec<PatchRow>> {
82        validate_doc_id(doc_id)?;
83        validate_vector(remote)?;
84        read_missing(&self.db, doc_id, remote)
85    }
86
87    /// `missing` and `vector` from one consistent snapshot.
88    pub fn sync_state(
89        &mut self,
90        doc_id: &str,
91        remote: &VersionVector,
92    ) -> Result<(Vec<PatchRow>, VersionVector)> {
93        validate_doc_id(doc_id)?;
94        validate_vector(remote)?;
95        let tx = self.db.transaction()?;
96        let rows = read_missing(&tx, doc_id, remote)?;
97        let vector = read_vector(&tx, doc_id)?;
98        tx.commit()?;
99        Ok((rows, vector))
100    }
101}
102
103pub fn validate_doc_id(doc_id: &str) -> Result<()> {
104    if doc_id.is_empty() || doc_id.contains('\0') || doc_id.encode_utf16().count() > 1024 {
105        return Err(Error::new("INVALID_MESSAGE", "Invalid document ID"));
106    }
107    Ok(())
108}
109
110fn validate_rows(rows: &[PatchRow]) -> Result<Option<&str>> {
111    if rows.len() > MAX_MESSAGE_ROWS {
112        return Err(Error::new("LIMIT_EXCEEDED", "Too many patch rows"));
113    }
114    let Some(first) = rows.first() else {
115        return Ok(None);
116    };
117    let doc_id = first.doc_id.as_str();
118    validate_doc_id(doc_id)?;
119    for row in rows {
120        validate_vector(&row.dependencies)?;
121        if row.doc_id != doc_id
122            || row.sid < 65536
123            || row.sid > MAX_SAFE_INTEGER
124            || row.clock < 1
125            || row.span < 1
126            || row.clock.saturating_add(row.span) > MAX_SAFE_INTEGER
127            || row.operations.is_empty()
128            || row.operations.len() > MAX_PATCH_BYTES
129            || own_clock(&row.dependencies, row.sid) > row.clock
130        {
131            return Err(Error::new("INVALID_MESSAGE", "Invalid storage segment"));
132        }
133    }
134    Ok(Some(doc_id))
135}
136
137fn own_clock(vv: &VersionVector, sid: u64) -> u64 {
138    vv.iter()
139        .find(|(id, _)| *id == sid)
140        .map_or(0, |(_, clock)| *clock)
141}
142
143fn conflict(message: &str) -> Error {
144    Error::new("CONFLICT", message)
145}
146
147/// The stored envelope of a neighbouring segment of the same actor.
148struct Segment {
149    clock: u64,
150    span: u64,
151    dependencies: VersionVector,
152}
153
154fn assert_compatible(row: &PatchRow, other: &Segment) -> Result<()> {
155    if row.clock < other.clock + other.span && other.clock < row.clock + row.span {
156        return Err(conflict("Overlapping immutable storage segments"));
157    }
158    let (previous_end, next_dependency) = if row.clock < other.clock {
159        (
160            row.clock + row.span,
161            own_clock(&other.dependencies, row.sid),
162        )
163    } else {
164        (
165            other.clock + other.span,
166            own_clock(&row.dependencies, row.sid),
167        )
168    };
169    if next_dependency < previous_end {
170        return Err(conflict("Actor history fork"));
171    }
172    Ok(())
173}
174
175fn append_rows(tx: &Transaction<'_>, doc_id: &str, rows: &[PatchRow]) -> Result<AppendResult> {
176    let mut get = tx.prepare_cached(
177        "SELECT span,dependencies,operations FROM pulsify_op_rows WHERE doc_id=?1 AND sid=?2 AND time_start=?3",
178    )?;
179    let mut neighbours = tx.prepare_cached(
180        "SELECT time_start,span,dependencies FROM pulsify_op_rows WHERE doc_id=?1 AND sid=?2 AND
181 (time_start=(SELECT MAX(time_start) FROM pulsify_op_rows WHERE doc_id=?1 AND sid=?2 AND time_start<?3)
182 OR time_start=(SELECT MIN(time_start) FROM pulsify_op_rows WHERE doc_id=?1 AND sid=?2 AND time_start>?3))",
183    )?;
184    let mut insert = tx.prepare_cached(
185        "INSERT INTO pulsify_op_rows(doc_id,sid,time_start,span,dependencies,operations) VALUES(?,?,?,?,?,?)",
186    )?;
187    let mut inserted = 0;
188    for row in rows {
189        let (sid, clock) = (row.sid as i64, row.clock as i64);
190        let existing = get
191            .query_row(params![doc_id, sid, clock], |r| {
192                Ok((
193                    r.get::<_, i64>(0)?,
194                    r.get::<_, String>(1)?,
195                    r.get::<_, Vec<u8>>(2)?,
196                ))
197            })
198            .optional()?;
199        if let Some((span, dependencies, operations)) = existing {
200            let dependencies: VersionVector = serde_json::from_str(&dependencies)?;
201            if span as u64 != row.span
202                || dependencies != row.dependencies
203                || operations != row.operations
204            {
205                return Err(conflict("Conflicting storage segment"));
206            }
207            continue;
208        }
209        let found = neighbours
210            .query_map(params![doc_id, sid, clock], |r| {
211                Ok((
212                    r.get::<_, i64>(0)?,
213                    r.get::<_, i64>(1)?,
214                    r.get::<_, String>(2)?,
215                ))
216            })?
217            .collect::<rusqlite::Result<Vec<_>>>()?;
218        for (clock, span, dependencies) in found {
219            let segment = Segment {
220                clock: clock as u64,
221                span: span as u64,
222                dependencies: serde_json::from_str(&dependencies)?,
223            };
224            assert_compatible(row, &segment)?;
225        }
226        insert.execute(params![
227            doc_id,
228            sid,
229            clock,
230            row.span as i64,
231            serde_json::to_string(&row.dependencies)?,
232            row.operations,
233        ])?;
234        inserted += 1;
235    }
236    let mut frontier: BTreeMap<u64, u64> = read_vector(tx, doc_id)?.into_iter().collect();
237    let mut pending = tx
238        .prepare_cached(
239            "SELECT sid,time_start,time_end,dependencies FROM pulsify_op_rows WHERE doc_id=? AND ready=0 ORDER BY sid,time_start",
240        )?
241        .query_map([doc_id], |r| {
242            Ok((
243                r.get::<_, i64>(0)? as u64,
244                r.get::<_, i64>(1)? as u64,
245                r.get::<_, i64>(2)? as u64,
246                r.get::<_, String>(3)?,
247            ))
248        })?
249        .collect::<rusqlite::Result<Vec<_>>>()?
250        .into_iter()
251        .map(|(sid, start, end, dependencies)| {
252            Ok((sid, start, end, serde_json::from_str(&dependencies)?))
253        })
254        .collect::<Result<Vec<(u64, u64, u64, VersionVector)>>>()?;
255    let actors: BTreeSet<u64> = frontier
256        .keys()
257        .copied()
258        .chain(pending.iter().map(|row| row.0))
259        .collect();
260    if actors.len() > MAX_VECTOR_ENTRIES {
261        return Err(Error::new(
262            "LIMIT_EXCEEDED",
263            "Document actor limit exceeded",
264        ));
265    }
266    let mut promote = tx.prepare_cached(
267        "UPDATE pulsify_op_rows SET ready=1 WHERE doc_id=? AND sid=? AND time_start=?",
268    )?;
269    let mut ready = 0;
270    let mut changed_actors = BTreeSet::new();
271    loop {
272        let mut changed = false;
273        let mut waiting = Vec::new();
274        for row in pending {
275            let (sid, start, end, ref dependencies) = row;
276            let contiguous = start == frontier.get(&sid).copied().unwrap_or(1);
277            let covered = dependencies
278                .iter()
279                .all(|(actor, clock)| frontier.get(actor).copied().unwrap_or(0) >= *clock);
280            if !contiguous || !covered {
281                waiting.push(row);
282                continue;
283            }
284            promote.execute(params![doc_id, sid as i64, start as i64])?;
285            let clock = frontier.entry(sid).or_insert(0);
286            *clock = (*clock).max(end);
287            changed_actors.insert(sid);
288            ready += 1;
289            changed = true;
290        }
291        pending = waiting;
292        if !changed {
293            break;
294        }
295    }
296    let mut advance = tx.prepare_cached(
297        "INSERT INTO pulsify_doc_clocks(doc_id,sid,clock) VALUES(?,?,?) ON CONFLICT(doc_id,sid) DO UPDATE SET clock=MAX(clock,excluded.clock)",
298    )?;
299    for sid in changed_actors {
300        advance.execute(params![doc_id, sid as i64, frontier[&sid] as i64])?;
301    }
302    Ok(AppendResult {
303        inserted,
304        ready,
305        pending: pending.len(),
306    })
307}
308
309fn read_vector(db: &Connection, doc_id: &str) -> Result<VersionVector> {
310    Ok(db
311        .prepare_cached("SELECT sid,clock FROM pulsify_doc_clocks WHERE doc_id=? ORDER BY sid")?
312        .query_map([doc_id], |r| {
313            Ok((r.get::<_, i64>(0)? as u64, r.get::<_, i64>(1)? as u64))
314        })?
315        .collect::<rusqlite::Result<_>>()?)
316}
317
318fn read_missing(db: &Connection, doc_id: &str, remote: &VersionVector) -> Result<Vec<PatchRow>> {
319    let map = |r: &rusqlite::Row<'_>| {
320        Ok((
321            r.get::<_, i64>(0)?,
322            r.get::<_, i64>(1)?,
323            r.get::<_, i64>(2)?,
324            r.get::<_, String>(3)?,
325            r.get::<_, Vec<u8>>(4)?,
326        ))
327    };
328    let found = if remote.is_empty() {
329        db.prepare_cached(
330            "SELECT sid,time_start,span,dependencies,operations FROM pulsify_op_rows WHERE doc_id=? AND ready=1 ORDER BY sid,time_start",
331        )?
332        .query_map([doc_id], map)?
333        .collect::<rusqlite::Result<Vec<_>>>()?
334    } else {
335        db.prepare_cached(
336            "WITH remote(sid,clock) AS (SELECT json_extract(value,'$[0]'),json_extract(value,'$[1]') FROM json_each(?1))
337 SELECT o.sid,o.time_start,o.span,o.dependencies,o.operations FROM pulsify_op_rows o LEFT JOIN remote r ON r.sid=o.sid
338 WHERE o.doc_id=?2 AND o.ready=1 AND o.time_end>COALESCE(r.clock,0) ORDER BY o.sid,o.time_start",
339        )?
340        .query_map(params![serde_json::to_string(remote)?, doc_id], map)?
341        .collect::<rusqlite::Result<Vec<_>>>()?
342    };
343    found
344        .into_iter()
345        .map(|(sid, clock, span, dependencies, operations)| {
346            Ok(PatchRow {
347                doc_id: doc_id.to_owned(),
348                sid: sid as u64,
349                clock: clock as u64,
350                span: span as u64,
351                dependencies: serde_json::from_str(&dependencies)?,
352                operations,
353            })
354        })
355        .collect()
356}
357
358#[cfg(test)]
359pub(crate) mod tests {
360    use super::*;
361
362    pub const A: u64 = 65536;
363    pub const B: u64 = 65537;
364
365    pub fn row(sid: u64, clock: u64, span: u64, dependencies: &[(u64, u64)]) -> PatchRow {
366        PatchRow {
367            doc_id: "doc".into(),
368            sid,
369            clock,
370            span,
371            dependencies: dependencies.to_vec(),
372            operations: vec![sid as u8, clock as u8, span as u8],
373        }
374    }
375    /// A contiguous segment of one actor, depending on its previous segment.
376    pub fn next(sid: u64, clock: u64, span: u64) -> PatchRow {
377        if clock == 1 {
378            row(sid, 1, span, &[])
379        } else {
380            row(sid, clock, span, &[(sid, clock)])
381        }
382    }
383    fn count(store: &SqliteStore, sql: &str) -> i64 {
384        store.connection().query_row(sql, [], |r| r.get(0)).unwrap()
385    }
386
387    #[test]
388    fn stores_and_reads_contiguous_rows() {
389        let mut store = SqliteStore::open_in_memory().unwrap();
390        let result = store
391            .append(&[next(A, 1, 2), next(A, 3, 1), next(B, 1, 4)])
392            .unwrap();
393        assert_eq!(
394            result,
395            AppendResult {
396                inserted: 3,
397                ready: 3,
398                pending: 0
399            }
400        );
401        assert_eq!(store.vector("doc").unwrap(), vec![(A, 4), (B, 5)]);
402        assert_eq!(store.missing("doc", &vec![]).unwrap().len(), 3);
403        let missing = store.missing("doc", &vec![(A, 3)]).unwrap();
404        assert_eq!(missing, vec![next(A, 3, 1), next(B, 1, 4)]);
405        assert!(store
406            .missing("doc", &vec![(A, 4), (B, 5)])
407            .unwrap()
408            .is_empty());
409        assert!(store.missing("other", &vec![]).unwrap().is_empty());
410        assert!(store.vector("other").unwrap().is_empty());
411        let stored: String = store
412            .connection()
413            .query_row(
414                "SELECT dependencies FROM pulsify_op_rows WHERE sid=? AND time_start=3",
415                [A as i64],
416                |r| r.get(0),
417            )
418            .unwrap();
419        assert_eq!(stored, "[[65536,3]]");
420    }
421
422    #[test]
423    fn exact_duplicates_are_idempotent() {
424        let mut store = SqliteStore::open_in_memory().unwrap();
425        store.append(&[next(A, 1, 2)]).unwrap();
426        let again = store.append(&[next(A, 1, 2), next(A, 1, 2)]).unwrap();
427        assert_eq!(again.inserted, 0);
428        assert_eq!(count(&store, "SELECT COUNT(*) FROM pulsify_op_rows"), 1);
429        assert_eq!(store.append(&[]).unwrap(), AppendResult::default());
430    }
431
432    #[test]
433    fn conflicts_roll_back_the_whole_batch() {
434        let mut store = SqliteStore::open_in_memory().unwrap();
435        store.append(&[next(A, 1, 2)]).unwrap();
436        let mut changed = next(A, 1, 2);
437        changed.operations = vec![9];
438        let cases = [
439            (changed, "Conflicting storage segment"),
440            (next(A, 2, 2), "Overlapping immutable storage segments"),
441            (row(A, 3, 1, &[(A, 2)]), "Actor history fork"),
442        ];
443        for (bad, message) in cases {
444            let error = store.append(&[next(B, 1, 1), bad]).unwrap_err();
445            assert_eq!((error.code, error.message.as_str()), ("CONFLICT", message));
446            assert_eq!(count(&store, "SELECT COUNT(*) FROM pulsify_op_rows"), 1);
447            assert_eq!(store.vector("doc").unwrap(), vec![(A, 3)]);
448        }
449        // Forks are detected against later neighbours and rows of the same batch.
450        store.append(&[row(A, 5, 1, &[(A, 3)])]).unwrap();
451        let error = store.append(&[next(A, 3, 2)]).unwrap_err();
452        assert_eq!(error.message, "Actor history fork");
453        let error = store
454            .append(&[next(A, 6, 1), row(A, 7, 1, &[(A, 4)])])
455            .unwrap_err();
456        assert_eq!(error.message, "Actor history fork");
457        assert_eq!(count(&store, "SELECT COUNT(*) FROM pulsify_op_rows"), 2);
458    }
459
460    #[test]
461    fn gaps_and_missing_dependencies_stay_pending_until_promoted() {
462        let mut store = SqliteStore::open_in_memory().unwrap();
463        let gap = store.append(&[next(A, 3, 2)]).unwrap();
464        assert_eq!((gap.inserted, gap.ready, gap.pending), (1, 0, 1));
465        let dependent = store.append(&[row(B, 1, 1, &[(A, 5)])]).unwrap();
466        assert_eq!((dependent.ready, dependent.pending), (0, 2));
467        assert!(store.vector("doc").unwrap().is_empty());
468        assert!(store.missing("doc", &vec![]).unwrap().is_empty());
469        assert_eq!(
470            count(&store, "SELECT COUNT(*) FROM pulsify_dangling_op_rows"),
471            2
472        );
473        let filled = store.append(&[next(A, 1, 2)]).unwrap();
474        assert_eq!((filled.inserted, filled.ready, filled.pending), (1, 3, 0));
475        assert_eq!(store.vector("doc").unwrap(), vec![(A, 5), (B, 2)]);
476        assert_eq!(
477            store.missing("doc", &vec![]).unwrap(),
478            vec![next(A, 1, 2), next(A, 3, 2), row(B, 1, 1, &[(A, 5)])]
479        );
480        assert_eq!(
481            count(&store, "SELECT COUNT(*) FROM pulsify_op_rows WHERE ready=0"),
482            0
483        );
484    }
485
486    #[test]
487    fn rejects_invalid_rows_and_documents() {
488        let mut store = SqliteStore::open_in_memory().unwrap();
489        let mut other_doc = next(B, 1, 1);
490        other_doc.doc_id = "other".into();
491        let mut empty = next(A, 1, 1);
492        empty.operations.clear();
493        for rows in [
494            vec![next(A, 1, 1), other_doc],
495            vec![row(1, 1, 1, &[])],
496            vec![row(A, 0, 1, &[])],
497            vec![row(A, 1, 0, &[])],
498            vec![row(A, 2, 1, &[(A, 3)])],
499            vec![row(A, 1, 1, &[(B, 1), (A, 1)])],
500            vec![empty],
501        ] {
502            assert_eq!(store.append(&rows).unwrap_err().code, "INVALID_MESSAGE");
503        }
504        let too_many: Vec<_> = (1..=MAX_MESSAGE_ROWS as u64 + 1)
505            .map(|clock| next(A, clock, 1))
506            .collect();
507        assert_eq!(store.append(&too_many).unwrap_err().code, "LIMIT_EXCEEDED");
508        assert_eq!(store.vector("").unwrap_err().code, "INVALID_MESSAGE");
509        assert_eq!(store.vector("a\0b").unwrap_err().code, "INVALID_MESSAGE");
510        assert_eq!(
511            store.vector(&"x".repeat(1025)).unwrap_err().code,
512            "INVALID_MESSAGE"
513        );
514        assert_eq!(
515            store
516                .missing("doc", &vec![(B, 1), (A, 1)])
517                .unwrap_err()
518                .code,
519            "INVALID_MESSAGE"
520        );
521        assert_eq!(count(&store, "SELECT COUNT(*) FROM pulsify_op_rows"), 0);
522    }
523
524    #[test]
525    fn limits_the_number_of_actors() {
526        let mut store = SqliteStore::open_in_memory().unwrap();
527        let rows: Vec<_> = (0..MAX_VECTOR_ENTRIES as u64)
528            .map(|i| next(A + i, 1, 1))
529            .collect();
530        store.append(&rows).unwrap();
531        let error = store
532            .append(&[next(A + MAX_VECTOR_ENTRIES as u64, 2, 1)])
533            .unwrap_err();
534        assert_eq!(error.code, "LIMIT_EXCEEDED");
535        assert_eq!(store.vector("doc").unwrap().len(), MAX_VECTOR_ENTRIES);
536        assert_eq!(
537            count(&store, "SELECT COUNT(*) FROM pulsify_op_rows"),
538            MAX_VECTOR_ENTRIES as i64
539        );
540    }
541
542    #[test]
543    fn persists_across_reopen() {
544        let path = std::env::temp_dir().join(format!(
545            "pulsify-sqlite-{}-{:?}.sqlite",
546            std::process::id(),
547            std::thread::current().id()
548        ));
549        let _ = std::fs::remove_file(&path);
550        {
551            let mut store = SqliteStore::open(&path).unwrap();
552            store.append(&[next(A, 1, 3), next(A, 5, 1)]).unwrap();
553        }
554        let mut store = SqliteStore::open(&path).unwrap();
555        assert_eq!(store.vector("doc").unwrap(), vec![(A, 4)]);
556        assert_eq!(store.append(&[next(A, 4, 1)]).unwrap().ready, 2);
557        let (rows, vector) = store.sync_state("doc", &vec![(A, 4)]).unwrap();
558        assert_eq!(vector, vec![(A, 6)]);
559        assert_eq!(rows, vec![next(A, 4, 1), next(A, 5, 1)]);
560        drop(store);
561        for suffix in ["", "-wal", "-shm"] {
562            let _ = std::fs::remove_file(format!("{}{suffix}", path.display()));
563        }
564    }
565}
pulsify_sqlite/store.rs

pulsify_sqlite/
store.rs
