Merge pull request #29 from orbitinghail:carlsverre/issue27

Make it possible for reducers to handle errors

Author: carlsverre

CHANGELOG.md

@@ -0,0 +1,9 @@
+This changelog documents changes across multiple projects contained in this monorepo. Each project is released for every SQLSync version, even if the project has not changed. The reason for this decision is to simplify testing and debugging. Lockstep versioning will be relaxed as SQLSync matures.
+
+# 0.2.0 - Dec 1 2023
+
+- Reducer can now handle query errors (#29)
+
+# 0.1.0 - Oct 23 2023
+
+- Initial release

Cargo.lock

@@ -17,8 +17,9 @@ members = [
 authors = ["Carl Sverre", "orbitinghail"]
 edition = "2021"
 homepage = "https://sqlsync.dev"
-license-file = "LICENSE"
+license = "Apache-2.0"
 repository = "https://github.com/orbitinghail/sqlsync"
+version = "0.2.0"
 
 [profile.release]
 lto = true

Cargo.toml

@@ -79,7 +79,7 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
           msg TEXT NOT NULL,
           created_at TEXT NOT NULL
         )"
-      ).await;
+      ).await?;
     }
 
     Mutation::AddMessage { id, msg } => {
@@ -88,7 +88,7 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
         "insert into messages (id, msg, created_at)
           values (?, ?, datetime('now'))",
         id, msg
-      ).await;
+      ).await?;
     }
   }
 

GUIDE.md

@@ -29,7 +29,7 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
                     created_at TEXT NOT NULL
                 )"
             )
-            .await;
+            .await?;
         }
 
         Mutation::CreateTask { id, description } => {
@@ -40,19 +40,19 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
                 id,
                 description
             )
-            .await;
+            .await?;
         }
 
         Mutation::DeleteTask { id } => {
-            execute!("delete from tasks where id = ?", id).await;
+            execute!("delete from tasks where id = ?", id).await?;
         }
 
         Mutation::ToggleCompleted { id } => {
             execute!(
                 "update tasks set completed = not completed where id = ?",
                 id
             )
-            .await;
+            .await?;
         }
     }
 

demo/demo-reducer/src/lib.rs

@@ -5,6 +5,11 @@
   <meta charset="UTF-8" />
   <meta name="viewport" content="width=device-width, initial-scale=1.0" />
   <title>SQLSync Demo</title>
+
+  <link rel="icon" type="image/png" href="/favicon-dark-32x32.png" media="(prefers-color-scheme: light)" />
+  <link rel="icon" type="image/png" href="/favicon-light-32x32.png" media="(prefers-color-scheme: dark)" />
+  <link rel="icon" type="image/svg+xml" href="/favicon.svg" />
+
   <script src="https://cdn.usefathom.com/script.js" data-site="LWITZJZR" async></script>
 </head>
 

demo/frontend/index.html

@@ -0,0 +1,2 @@
+User-agent: *
+Disallow: /

demo/frontend/public/favicon-dark-32x32.png

@@ -1,25 +1,15 @@
 {
   "name": "@orbitinghail/sqlsync-react",
-  "version": "0.1.0",
+  "version": "0.2.0",
   "description": "SQLSync is a collaborative offline-first wrapper around SQLite. It is designed to synchronize web application state between users, devices, and the edge.",
   "homepage": "https://sqlsync.dev",
   "license": "Apache-2.0",
-  "keywords": [
-    "sqlsync",
-    "sql",
-    "database",
-    "sqlite",
-    "offline-first",
-    "local-first"
-  ],
+  "keywords": ["sqlsync", "sql", "database", "sqlite", "offline-first", "local-first"],
   "repository": {
     "type": "git",
     "url": "https://github.com/orbitinghail/sqlsync"
   },
-  "files": [
-    "dist",
-    "src"
-  ],
+  "files": ["dist", "src"],
   "type": "module",
   "main": "./dist/sqlsync-react.js",
   "types": "./src/index.ts",
@@ -55,4 +45,4 @@
     "fast-equals": "^5.0.1",
     "fast-sha256": "^1.3.0"
   }
-}
+}

demo/frontend/public/favicon-light-32x32.png

@@ -16,15 +16,18 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
 
     match mutation {
         Mutation::InitSchema => {
-            futures::join!(
-                execute!(
-                    "CREATE TABLE IF NOT EXISTS counter (
+            let create_table = execute!(
+                "CREATE TABLE IF NOT EXISTS counter (
                     id INTEGER PRIMARY KEY,
                     value INTEGER
                 )"
-                ),
-                execute!("INSERT OR IGNORE INTO counter (id, value) VALUES (0, 0)")
             );
+            let init_counter = execute!(
+                "INSERT OR IGNORE INTO counter (id, value) VALUES (0, 0)"
+            );
+
+            create_table.await?;
+            init_counter.await?;
         }
 
         Mutation::Incr { value } => {
@@ -33,7 +36,7 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
                 ON CONFLICT (id) DO UPDATE SET value = value + ?",
                 value
             )
-            .await;
+            .await?;
         }
 
         Mutation::Decr { value } => {
@@ -42,7 +45,7 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
                 ON CONFLICT (id) DO UPDATE SET value = value - ?",
                 value
             )
-            .await;
+            .await?;
         }
     }
 

demo/frontend/public/favicon.svg

@@ -1,27 +1,27 @@
 [package]
 name = "sqlsync-reducer"
 resolver = "2"
-version = "0.1.0"
 description = "SQLSync is a collaborative offline-first wrapper around SQLite. A SQLSync Reducer is responsible for executing mutations on the database both locally and on the server."
 
+version.workspace = true
 authors.workspace = true
 edition.workspace = true
 homepage.workspace = true
-license-file.workspace = true
+license.workspace = true
 repository.workspace = true
 
 [dependencies]
 serde = { workspace = true, features = ["derive"] }
 bincode.workspace = true
 futures.workspace = true
 log.workspace = true
+thiserror.workspace = true
 
 wasmi = { workspace = true, optional = true }
-thiserror = { workspace = true, optional = true }
 
 [features]
 default = ["guest"]
-host = ["wasmi", "thiserror"]
+host = ["wasmi"]
 guest = []
 
 [dev-dependencies]

demo/frontend/public/robots.txt

@@ -17,7 +17,8 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
     log::info!("running query and execute at the same time");
 
     let x: Option<i64> = None;
-    let query_future = query!("SELECT * FROM foo WHERE bar = ?", "baz", 1, 1.23, x);
+    let query_future =
+        query!("SELECT * FROM foo WHERE bar = ?", "baz", 1, 1.23, x);
     let exec_future = execute!("SELECT * FROM foo WHERE bar = ?", "baz");
 
     let (result, result2) = futures::join!(query_future, exec_future);
@@ -35,6 +36,17 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
     let result = query_future.await;
     log::info!("final query result: {:?}", result);
 
+    log::info!("testing errors");
+    if let Err(err) = execute!("FAIL").await {
+        log::error!("error: {:?}", err);
+    } else {
+        panic!("expected error");
+    }
+
+    log::info!("make sure we can resume queries after an error");
+    let result = execute!("SELECT * FROM foo WHERE bar = ?", "baz").await;
+    log::info!("result: {:?}", result);
+
     Ok(())
 }
 

lib/sqlsync-react/package.json

@@ -5,7 +5,7 @@ use std::collections::BTreeMap;
 use serde::{Deserialize, Serialize};
 use sqlsync_reducer::{
     host_ffi::{register_log_handler, WasmFFI},
-    types::{ExecResponse, QueryResponse, Request},
+    types::{ErrorResponse, ExecResponse, QueryResponse, Request},
 };
 use wasmi::{Engine, Linker, Module, Store};
 
@@ -22,8 +22,9 @@ fn main() -> anyhow::Result<()> {
         .init()?;
 
     // build guest.wasm using: `cargo build --target wasm32-unknown-unknown --example guest`
-    let wasm_bytes =
-        include_bytes!("../../../target/wasm32-unknown-unknown/debug/examples/guest.wasm");
+    let wasm_bytes = include_bytes!(
+        "../../../target/wasm32-unknown-unknown/debug/examples/guest.wasm"
+    );
 
     let engine = Engine::default();
     let module = Module::new(&engine, &wasm_bytes[..])?;
@@ -32,7 +33,8 @@ fn main() -> anyhow::Result<()> {
     register_log_handler(&mut linker)?;
 
     let mut store = Store::new(&engine, WasmFFI::uninitialized());
-    let instance = linker.instantiate(&mut store, &module)?.start(&mut store)?;
+    let instance =
+        linker.instantiate(&mut store, &module)?.start(&mut store)?;
 
     // initialize the FFI
     let ffi = WasmFFI::initialized(&store, &instance)?;
@@ -52,21 +54,39 @@ fn main() -> anyhow::Result<()> {
         let mut responses = BTreeMap::new();
         for (id, req) in requests_inner {
             match req {
-                Request::Query { .. } => {
-                    log::info!("received query request: {:?}", req);
+                Request::Query { sql, params } => {
+                    log::info!("received query request: {} {:?}", sql, params);
                     let ptr = ffi.encode(
                         &mut store,
-                        &QueryResponse {
+                        &Ok::<_, ErrorResponse>(QueryResponse {
                             columns: vec!["foo".into(), "bar".into()],
                             rows: vec![vec!["baz".into(), "qux".into()].into()],
-                        },
+                        }),
                     )?;
                     responses.insert(id, ptr);
                 }
-                Request::Exec { .. } => {
-                    log::info!("received exec request: {:?}", req);
-                    let ptr = ffi.encode(&mut store, &ExecResponse { changes: 1 })?;
-                    responses.insert(id, ptr);
+                Request::Exec { sql, params } => {
+                    log::info!("received exec request: {} {:?}", sql, params);
+                    if sql == "FAIL" {
+                        let ptr = ffi.encode(
+                            &mut store,
+                            &Err::<ExecResponse, _>(
+                                ErrorResponse::SqliteError {
+                                    code: 1,
+                                    message: "error".to_string(),
+                                },
+                            ),
+                        )?;
+                        responses.insert(id, ptr);
+                    } else {
+                        let ptr = ffi.encode(
+                            &mut store,
+                            &Ok::<_, ErrorResponse>(ExecResponse {
+                                changes: 1,
+                            }),
+                        )?;
+                        responses.insert(id, ptr);
+                    }
                 }
             }
         }