Pin<Box<Storage>> to ensure that Storage never moves, removed changes to sqlite-vfs, new just test command

Author: carlsverre

CONTRIBUTING.md

@@ -52,6 +52,13 @@ These tests are useful for learning more about how SQLSync works.
 just unit-test
 just test-end-to-end-local
 just test-end-to-end-local-net
+just test-sqlsync-reducer
+```
+
+You can run all the tests in one command like so:
+
+```bash
+just test
 ```
 
 ### Submitting a pull request

Cargo.lock

@@ -51,6 +51,7 @@ worker = "0.0.18"
 event-listener = "3.0"
 sha2 = "0.10.8"
 serde-wasm-bindgen = "0.6"
+pin-project = "1.1"
 
 # specific revision of tsify needed for serde updates
 tsify = { git = "https://github.com/siefkenj/tsify", rev = "145ed4c8ef6417003e182fad41d1c0f26ed645e5", default-features = false }

Cargo.toml

@@ -14,6 +14,10 @@ unit-test:
 build: build-wasm
     cargo build -p sqlsync
 
+test:
+    just unit-test
+    just run-with-prefix 'test-'
+
 build-wasm:
     just run-with-prefix 'wasm-'
 

justfile

@@ -1,5 +1,6 @@
 # sqlite-vfs
 
-Copied from https://github.com/rklaehn/sqlite-vfs and modified to include more hooks. Plan to upstream.
+Copy of https://github.com/rklaehn/sqlite-vfs.
+Only changed rusqlite dependency.
 
-LICENSE: Apache 2.0
+LICENSE: Apache 2.0

lib/sqlite-vfs/README.md

@@ -65,6 +65,7 @@ fn main() -> anyhow::Result<()> {
                 }
                 Request::Exec { sql, params } => {
                     log::info!("received exec request: {} {:?}", sql, params);
+                    log::info!("the following error is EXPECTED");
                     if sql == "FAIL" {
                         let ptr = ffi.encode(
                             &mut store,

lib/sqlsync-reducer/examples/host.rs

@@ -25,6 +25,7 @@ bs58.workspace = true
 hex.workspace = true
 libsqlite3-sys.workspace = true
 rusqlite.workspace = true
+pin-project.workspace = true
 
 [dependencies.sqlsync-reducer]
 path = "../sqlsync-reducer"

lib/sqlsync/Cargo.toml

@@ -3,6 +3,7 @@ use std::collections::{HashMap, VecDeque};
 use std::convert::From;
 use std::fmt::Debug;
 use std::io;
+use std::pin::Pin;
 
 use rusqlite::Transaction;
 
@@ -25,7 +26,7 @@ struct ReceiveQueueEntry {
 
 pub struct CoordinatorDocument<J: Journal, R> {
     reducer: R,
-    storage: Box<Storage<J>>,
+    storage: Pin<Box<Storage<J>>>,
     sqlite: ConnectionPair,
     timeline_factory: J::Factory,
     timelines: HashMap<JournalId, J>,

lib/sqlsync/src/coordinator.rs

@@ -1,12 +1,16 @@
-use std::convert::From;
+use std::{convert::From, pin::Pin};
 
 use rusqlite::{
     hooks::{AuthAction, AuthContext, Authorization},
     Connection, OpenFlags, Transaction,
 };
-use sqlite_vfs::FilePtr;
 
-use crate::{journal::Journal, page::PAGESIZE, storage::Storage, vfs::StorageVfs};
+use crate::{
+    journal::Journal,
+    page::PAGESIZE,
+    storage::Storage,
+    vfs::{FilePtr, StorageVfs},
+};
 
 pub struct ConnectionPair {
     pub readwrite: Connection,
@@ -15,8 +19,8 @@ pub struct ConnectionPair {
 
 pub fn open_with_vfs<J: Journal>(
     journal: J,
-) -> rusqlite::Result<(ConnectionPair, Box<Storage<J>>)> {
-    let mut storage = Box::new(Storage::new(journal));
+) -> rusqlite::Result<(ConnectionPair, Pin<Box<Storage<J>>>)> {
+    let mut storage = Box::pin(Storage::new(journal));
     let storage_ptr = FilePtr::new(&mut storage);
 
     // generate random vfs name

lib/sqlsync/src/db.rs

@@ -1,4 +1,4 @@
-use std::{fmt::Debug, io};
+use std::{fmt::Debug, io, pin::Pin};
 
 use rusqlite::Connection;
 
@@ -26,7 +26,7 @@ impl Signal for NoopSignal {
 pub struct LocalDocument<J, S> {
     reducer: WasmReducer,
     timeline: J,
-    storage: Box<Storage<J>>,
+    storage: Pin<Box<Storage<J>>>,
     sqlite: ConnectionPair,
 
     // signals

lib/sqlsync/src/local.rs

@@ -1,5 +1,6 @@
 use std::{collections::HashSet, fmt::Debug, io};
 
+use pin_project::pin_project;
 use serde::{Deserialize, Serialize};
 use sqlite_vfs::SQLITE_IOERR;
 
@@ -34,6 +35,7 @@ pub enum StorageChange {
     Tables { root_pages_sorted: Vec<PageIdx> },
 }
 
+#[pin_project]
 pub struct Storage<J> {
     journal: J,
     visible_lsn_range: LsnRange,

lib/sqlsync/src/storage.rs

@@ -1,6 +1,8 @@
+use std::pin::Pin;
+
 use libsqlite3_sys::SQLITE_IOERR;
 use log::{debug, trace};
-use sqlite_vfs::{File, FilePtr, OpenKind, Vfs, VfsResult};
+use sqlite_vfs::{File, OpenKind, Vfs, VfsResult};
 
 use crate::{journal::Journal, storage::Storage, unixtime::unix_timestamp_milliseconds};
 
@@ -59,3 +61,51 @@ impl<J: Journal> Vfs for StorageVfs<J> {
         ((2440587.5 + now / 864.0e5) * 864.0e5) as i64
     }
 }
+
+/// Allow File to be an unsafe pointer
+pub struct FilePtr<T: File>(*mut T);
+
+impl<T: File> FilePtr<T> {
+    pub fn new(f: &mut Pin<Box<T>>) -> Self {
+        // SAFETY: we are creating a raw pointer from a reference to a Box
+        // and we are not moving the Box, so the Box will not be dropped
+        // while the raw pointer is still in use.
+        Self(unsafe { f.as_mut().get_unchecked_mut() })
+    }
+}
+
+impl<T: File> Clone for FilePtr<T> {
+    fn clone(&self) -> Self {
+        Self(self.0)
+    }
+}
+
+impl<T: File> File for FilePtr<T> {
+    fn sector_size(&self) -> usize {
+        unsafe { (*self.0).sector_size() }
+    }
+
+    fn device_characteristics(&self) -> i32 {
+        unsafe { (*self.0).device_characteristics() }
+    }
+
+    fn file_size(&self) -> VfsResult<u64> {
+        unsafe { (*self.0).file_size() }
+    }
+
+    fn truncate(&mut self, size: u64) -> VfsResult<()> {
+        unsafe { (*self.0).truncate(size) }
+    }
+
+    fn write(&mut self, pos: u64, buf: &[u8]) -> VfsResult<usize> {
+        unsafe { (*self.0).write(pos, buf) }
+    }
+
+    fn read(&mut self, pos: u64, buf: &mut [u8]) -> VfsResult<usize> {
+        unsafe { (*self.0).read(pos, buf) }
+    }
+
+    fn sync(&mut self) -> VfsResult<()> {
+        unsafe { (*self.0).sync() }
+    }
+}