reducer can now handle errors

carlsverre · carlsverre · commit 874f55e11433 · 2023-11-30T16:31:58.000-08:00
diff --git a/demo/demo-reducer/src/lib.rs b/demo/demo-reducer/src/lib.rs
@@ -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?;
         }
     }
 
diff --git a/lib/sqlsync-react/sqlsync-react-test-reducer/src/lib.rs b/lib/sqlsync-react/sqlsync-react-test-reducer/src/lib.rs
@@ -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?;
         }
     }
 
diff --git a/lib/sqlsync-reducer/Cargo.toml b/lib/sqlsync-reducer/Cargo.toml
@@ -15,13 +15,13 @@ 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]
diff --git a/lib/sqlsync-reducer/examples/host.rs b/lib/sqlsync-reducer/examples/host.rs
@@ -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, SqliteError},
+    types::{ErrorResponse, ExecResponse, QueryResponse, Request},
 };
 use wasmi::{Engine, Linker, Module, Store};
 
@@ -58,7 +58,7 @@ fn main() -> anyhow::Result<()> {
                     log::info!("received query request: {} {:?}", sql, params);
                     let ptr = ffi.encode(
                         &mut store,
-                        &Ok::<_, SqliteError>(QueryResponse {
+                        &Ok::<_, ErrorResponse>(QueryResponse {
                             columns: vec!["foo".into(), "bar".into()],
                             rows: vec![vec!["baz".into(), "qux".into()].into()],
                         }),
@@ -70,16 +70,20 @@ fn main() -> anyhow::Result<()> {
                     if sql == "FAIL" {
                         let ptr = ffi.encode(
                             &mut store,
-                            &Err::<ExecResponse, _>(SqliteError {
-                                code: Some(1),
-                                message: "error".to_string(),
-                            }),
+                            &Err::<ExecResponse, _>(
+                                ErrorResponse::SqliteError {
+                                    code: 1,
+                                    message: "error".to_string(),
+                                },
+                            ),
                         )?;
                         responses.insert(id, ptr);
                     } else {
                         let ptr = ffi.encode(
                             &mut store,
-                            &Ok::<_, SqliteError>(ExecResponse { changes: 1 }),
+                            &Ok::<_, ErrorResponse>(ExecResponse {
+                                changes: 1,
+                            }),
                         )?;
                         responses.insert(id, ptr);
                     }
diff --git a/lib/sqlsync-reducer/src/guest_reactor.rs b/lib/sqlsync-reducer/src/guest_reactor.rs
@@ -12,8 +12,8 @@ use serde::de::DeserializeOwned;
 use crate::{
     guest_ffi::{fbm, FFIBufPtr},
     types::{
-        ExecResponse, QueryResponse, ReducerError, Request, RequestId,
-        Requests, Responses, SqliteError, SqliteValue,
+        ErrorResponse, ExecResponse, QueryResponse, ReducerError, Request,
+        RequestId, Requests, Responses, SqliteValue,
     },
 };
 
@@ -107,6 +107,7 @@ impl Reactor {
     }
 }
 
+#[must_use]
 pub struct ResponseFuture<T: DeserializeOwned> {
     id: RequestId,
     _marker: std::marker::PhantomData<T>,
@@ -132,7 +133,7 @@ impl<T: DeserializeOwned> Future for ResponseFuture<T> {
 pub fn raw_query(
     sql: String,
     params: Vec<SqliteValue>,
-) -> ResponseFuture<Result<QueryResponse, SqliteError>> {
+) -> ResponseFuture<Result<QueryResponse, ErrorResponse>> {
     let request = Request::Query { sql, params };
     let id = reactor().queue_request(request);
     ResponseFuture::new(id)
@@ -141,7 +142,7 @@ pub fn raw_query(
 pub fn raw_execute(
     sql: String,
     params: Vec<SqliteValue>,
-) -> ResponseFuture<Result<ExecResponse, SqliteError>> {
+) -> ResponseFuture<Result<ExecResponse, ErrorResponse>> {
     let request = Request::Exec { sql, params };
     let id = reactor().queue_request(request);
     ResponseFuture::new(id)
diff --git a/lib/sqlsync-reducer/src/types.rs b/lib/sqlsync-reducer/src/types.rs
@@ -9,6 +9,7 @@ use std::{
 
 use log::Level;
 use serde::{Deserialize, Serialize};
+use thiserror::Error;
 
 pub type RequestId = u32;
 
@@ -62,10 +63,12 @@ pub struct ExecResponse {
     pub changes: usize,
 }
 
-#[derive(Serialize, Deserialize, Debug)]
-pub struct SqliteError {
-    pub code: Option<i32>,
-    pub message: String,
+#[derive(Serialize, Deserialize, Debug, Error)]
+pub enum ErrorResponse {
+    #[error("SQLite Error({code}): {message}")]
+    SqliteError { code: i32, message: String },
+    #[error("Unknown: {0}")]
+    Unknown(String),
 }
 
 #[derive(Serialize, Deserialize)]
diff --git a/lib/sqlsync/examples/counter-reducer.rs b/lib/sqlsync/examples/counter-reducer.rs
@@ -15,31 +15,32 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
     let mutation: Mutation = bincode::deserialize(&mutation)?;
     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 => {
             execute!(
                 "INSERT INTO counter (id, value) VALUES (0, 0)
                 ON CONFLICT (id) DO UPDATE SET value = value + 1"
             )
-            .await;
+            .await?;
         }
         Mutation::Decr => {
             execute!(
                 "INSERT INTO counter (id, value) VALUES (0, 0)
                 ON CONFLICT (id) DO UPDATE SET value = value - 1"
             )
-            .await;
+            .await?;
         }
     }
 
diff --git a/lib/sqlsync/examples/hello-reducer.rs b/lib/sqlsync/examples/hello-reducer.rs
@@ -18,10 +18,10 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
                 key,
                 value
             )
-            .await;
+            .await?;
         }
         Mutation::Delete(key) => {
-            execute!("DELETE FROM kv WHERE key = ?", key).await;
+            execute!("DELETE FROM kv WHERE key = ?", key).await?;
         }
     }
 
diff --git a/lib/sqlsync/examples/task-reducer.rs b/lib/sqlsync/examples/task-reducer.rs
@@ -29,7 +29,7 @@ enum Mutation {
 }
 
 async fn query_max_sort() -> Result<f64, ReducerError> {
-    let response = query!("select max(sort) from tasks").await;
+    let response = query!("select max(sort) from tasks").await?;
     assert!(response.rows.len() == 1, "expected 1 row");
     Ok(response.rows[0].maybe_get(0)?.unwrap_or(0.0))
 }
@@ -45,7 +45,7 @@ async fn query_sort_after(id: i64) -> Result<f64, ReducerError> {
         ",
         id
     )
-    .await;
+    .await?;
 
     if response.rows.len() == 0 {
         query_max_sort().await
@@ -73,7 +73,7 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
                     created_at TEXT NOT NULL
                 )"
             )
-            .await;
+            .await?;
         }
 
         Mutation::AppendTask { id, description } => {
@@ -86,18 +86,14 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
                 max_sort + 1.,
                 description
             )
-            .await;
+            .await?;
         }
 
         Mutation::RemoveTask { id } => {
-            execute!("delete from tasks where id = ?", id).await;
+            execute!("delete from tasks where id = ?", id).await?;
         }
 
-        Mutation::UpdateTask {
-            id,
-            description,
-            completed,
-        } => {
+        Mutation::UpdateTask { id, description, completed } => {
             execute!(
                 "update tasks set
                     description = IFNULL(?, description),
@@ -107,12 +103,13 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {
                 description,
                 completed
             )
-            .await;
+            .await?;
         }
 
         Mutation::MoveTask { id, after } => {
             let new_sort = query_sort_after(after).await?;
-            execute!("update tasks set sort = ? where id = ?", new_sort, id);
+            execute!("update tasks set sort = ? where id = ?", new_sort, id)
+                .await?;
         }
     }
 
diff --git a/lib/sqlsync/src/reducer.rs b/lib/sqlsync/src/reducer.rs

Original file line number	Diff line number	Diff line change
`@@ -29,7 +29,7 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {`
`29`	`29`	`created_at TEXT NOT NULL`
`30`	`30`	`)"`
`31`	`31`	`)`
`32`		`- .await;`
	`32`	`+ .await?;`
`33`	`33`	`}`
`34`	`34`
`35`	`35`	`Mutation::CreateTask { id, description } => {`
`@@ -40,19 +40,19 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {`
`40`	`40`	`id,`
`41`	`41`	`description`
`42`	`42`	`)`
`43`		`- .await;`
	`43`	`+ .await?;`
`44`	`44`	`}`
`45`	`45`
`46`	`46`	`Mutation::DeleteTask { id } => {`
`47`		`- execute!("delete from tasks where id = ?", id).await;`
	`47`	`+ execute!("delete from tasks where id = ?", id).await?;`
`48`	`48`	`}`
`49`	`49`
`50`	`50`	`Mutation::ToggleCompleted { id } => {`
`51`	`51`	`execute!(`
`52`	`52`	`"update tasks set completed = not completed where id = ?",`
`53`	`53`	`id`
`54`	`54`	`)`
`55`		`- .await;`
	`55`	`+ .await?;`
`56`	`56`	`}`
`57`	`57`	`}`
`58`	`58`
Original file line number	Diff line number	Diff line change
`@@ -18,10 +18,10 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {`
`18`	`18`	`key,`
`19`	`19`	`value`
`20`	`20`	`)`
`21`		`- .await;`
	`21`	`+ .await?;`
`22`	`22`	`}`
`23`	`23`	`Mutation::Delete(key) => {`
`24`		`- execute!("DELETE FROM kv WHERE key = ?", key).await;`
	`24`	`+ execute!("DELETE FROM kv WHERE key = ?", key).await?;`
`25`	`25`	`}`
`26`	`26`	`}`
`27`	`27`
Original file line number	Diff line number	Diff line change
`@@ -29,7 +29,7 @@ enum Mutation {`
`29`	`29`	`}`
`30`	`30`
`31`	`31`	`async fn query_max_sort() -> Result<f64, ReducerError> {`
`32`		`- let response = query!("select max(sort) from tasks").await;`
	`32`	`+ let response = query!("select max(sort) from tasks").await?;`
`33`	`33`	`assert!(response.rows.len() == 1, "expected 1 row");`
`34`	`34`	`Ok(response.rows[0].maybe_get(0)?.unwrap_or(0.0))`
`35`	`35`	`}`
`@@ -45,7 +45,7 @@ async fn query_sort_after(id: i64) -> Result<f64, ReducerError> {`
`45`	`45`	`",`
`46`	`46`	`id`
`47`	`47`	`)`
`48`		`- .await;`
	`48`	`+ .await?;`
`49`	`49`
`50`	`50`	`if response.rows.len() == 0 {`
`51`	`51`	`query_max_sort().await`
`@@ -73,7 +73,7 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {`
`73`	`73`	`created_at TEXT NOT NULL`
`74`	`74`	`)"`
`75`	`75`	`)`
`76`		`- .await;`
	`76`	`+ .await?;`
`77`	`77`	`}`
`78`	`78`
`79`	`79`	`Mutation::AppendTask { id, description } => {`
`@@ -86,18 +86,14 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {`
`86`	`86`	`max_sort + 1.,`
`87`	`87`	`description`
`88`	`88`	`)`
`89`		`- .await;`
	`89`	`+ .await?;`
`90`	`90`	`}`
`91`	`91`
`92`	`92`	`Mutation::RemoveTask { id } => {`
`93`		`- execute!("delete from tasks where id = ?", id).await;`
	`93`	`+ execute!("delete from tasks where id = ?", id).await?;`
`94`	`94`	`}`
`95`	`95`
`96`		`- Mutation::UpdateTask {`
`97`		`- id,`
`98`		`- description,`
`99`		`- completed,`
`100`		`- } => {`
	`96`	`+ Mutation::UpdateTask { id, description, completed } => {`
`101`	`97`	`execute!(`
`102`	`98`	`"update tasks set`
`103`	`99`	`description = IFNULL(?, description),`
`@@ -107,12 +103,13 @@ async fn reducer(mutation: Vec<u8>) -> Result<(), ReducerError> {`
`107`	`103`	`description,`
`108`	`104`	`completed`
`109`	`105`	`)`
`110`		`- .await;`
	`106`	`+ .await?;`
`111`	`107`	`}`
`112`	`108`
`113`	`109`	`Mutation::MoveTask { id, after } => {`
`114`	`110`	`let new_sort = query_sort_after(after).await?;`
`115`		`- execute!("update tasks set sort = ? where id = ?", new_sort, id);`
	`111`	`+ execute!("update tasks set sort = ? where id = ?", new_sort, id)`
	`112`	`+ .await?;`
`116`	`113`	`}`
`117`	`114`	`}`
`118`	`115`