diff --git a/Cargo.lock b/Cargo.lock index 0935c81..f96e478 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -308,6 +308,17 @@ version = "1.0.98" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e16d2d3311acee920a9eb8d33b8cbc1787ce4a264e85f964c2404b969bdcd487" +[[package]] +name = "async-trait" +version = "0.1.88" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e539d3fca749fcee5236ab05e93a52867dd549cc157c8cb7f99595f3cedffdb5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "autocfg" version = "1.5.0" @@ -1562,7 +1573,7 @@ dependencies = [ [[package]] name = "taskchampion-sync-server" -version = "0.6.2-pre" +version = "0.7.0-pre" dependencies = [ "actix-rt", "actix-web", @@ -1585,28 +1596,32 @@ dependencies = [ [[package]] name = "taskchampion-sync-server-core" -version = "0.6.2-pre" +version = "0.7.0-pre" dependencies = [ "anyhow", + "async-trait", "chrono", "env_logger", "log", "pretty_assertions", "thiserror", + "tokio", "uuid", ] [[package]] name = "taskchampion-sync-server-storage-sqlite" -version = "0.6.2-pre" +version = "0.7.0-pre" dependencies = [ "anyhow", + "async-trait", "chrono", "pretty_assertions", "rusqlite", "taskchampion-sync-server-core", "tempfile", "thiserror", + "tokio", "uuid", ] @@ -1709,9 +1724,21 @@ dependencies = [ "signal-hook-registry", "slab", "socket2", + "tokio-macros", "windows-sys 0.52.0", ] +[[package]] +name = "tokio-macros" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "tokio-util" version = "0.7.15" diff --git a/Cargo.toml b/Cargo.toml index ecdee0a..4bfe6d6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ rust-version = "1.82.0" # MSRV [workspace.dependencies] +async-trait = "0.1.88" uuid = { version = "^1.17.0", features = ["serde", "v4"] } actix-web = "^4.11.0" anyhow = "1.0" @@ -24,3 +25,4 @@ actix-rt = "2" tempfile = "3" pretty_assertions = "1" temp-env = "0.3" +tokio = { version = "*", features = ["rt", "macros"] } diff --git a/core/Cargo.toml b/core/Cargo.toml index ee6f890..68b9adf 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "taskchampion-sync-server-core" -version = "0.6.2-pre" +version = "0.7.0-pre" authors = ["Dustin J. Mitchell "] edition = "2021" description = "Core of sync protocol for TaskChampion" @@ -10,6 +10,7 @@ license = "MIT" [dependencies] uuid.workspace = true +async-trait.workspace = true anyhow.workspace = true thiserror.workspace = true log.workspace = true @@ -18,3 +19,4 @@ chrono.workspace = true [dev-dependencies] pretty_assertions.workspace = true +tokio.workspace = true diff --git a/core/src/inmemory.rs b/core/src/inmemory.rs index a6ace2f..07f1fc5 100644 --- a/core/src/inmemory.rs +++ b/core/src/inmemory.rs @@ -44,8 +44,9 @@ struct InnerTxn<'a> { committed: bool, } +#[async_trait::async_trait] impl Storage for InMemoryStorage { - fn txn(&self, client_id: Uuid) -> anyhow::Result> { + async fn txn(&self, client_id: Uuid) -> anyhow::Result> { Ok(Box::new(InnerTxn { client_id, guard: self.0.lock().expect("poisoned lock"), @@ -55,12 +56,13 @@ impl Storage for InMemoryStorage { } } +#[async_trait::async_trait(?Send)] impl StorageTxn for InnerTxn<'_> { - fn get_client(&mut self) -> anyhow::Result> { + async fn get_client(&mut self) -> anyhow::Result> { Ok(self.guard.clients.get(&self.client_id).cloned()) } - fn new_client(&mut self, latest_version_id: Uuid) -> anyhow::Result<()> { + async fn new_client(&mut self, latest_version_id: Uuid) -> anyhow::Result<()> { if self.guard.clients.contains_key(&self.client_id) { return Err(anyhow::anyhow!("Client {} already exists", self.client_id)); } @@ -75,7 +77,7 @@ impl StorageTxn for InnerTxn<'_> { Ok(()) } - fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec) -> anyhow::Result<()> { + async fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec) -> anyhow::Result<()> { let client = self .guard .clients @@ -87,7 +89,7 @@ impl StorageTxn for InnerTxn<'_> { Ok(()) } - fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result>> { + async fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result>> { // sanity check let client = self.guard.clients.get(&self.client_id); let client = client.ok_or_else(|| anyhow::anyhow!("no such client"))?; @@ -97,7 +99,7 @@ impl StorageTxn for InnerTxn<'_> { Ok(self.guard.snapshots.get(&self.client_id).cloned()) } - fn get_version_by_parent( + async fn get_version_by_parent( &mut self, parent_version_id: Uuid, ) -> anyhow::Result> { @@ -116,7 +118,7 @@ impl StorageTxn for InnerTxn<'_> { } } - fn get_version(&mut self, version_id: Uuid) -> anyhow::Result> { + async fn get_version(&mut self, version_id: Uuid) -> anyhow::Result> { Ok(self .guard .versions @@ -124,7 +126,7 @@ impl StorageTxn for InnerTxn<'_> { .cloned()) } - fn add_version( + async fn add_version( &mut self, version_id: Uuid, parent_version_id: Uuid, @@ -174,7 +176,7 @@ impl StorageTxn for InnerTxn<'_> { Ok(()) } - fn commit(&mut self) -> anyhow::Result<()> { + async fn commit(&mut self) -> anyhow::Result<()> { self.committed = true; Ok(()) } @@ -193,32 +195,33 @@ mod test { use super::*; use chrono::Utc; - #[test] - fn test_get_client_empty() -> anyhow::Result<()> { + #[tokio::test] + async fn test_get_client_empty() -> anyhow::Result<()> { let storage = InMemoryStorage::new(); - let mut txn = storage.txn(Uuid::new_v4())?; - let maybe_client = txn.get_client()?; + let mut txn = storage.txn(Uuid::new_v4()).await?; + let maybe_client = txn.get_client().await?; assert!(maybe_client.is_none()); Ok(()) } - #[test] - fn test_client_storage() -> anyhow::Result<()> { + #[tokio::test] + async fn test_client_storage() -> anyhow::Result<()> { let storage = InMemoryStorage::new(); let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; + let mut txn = storage.txn(client_id).await?; let latest_version_id = Uuid::new_v4(); - txn.new_client(latest_version_id)?; + txn.new_client(latest_version_id).await?; - let client = txn.get_client()?.unwrap(); + let client = txn.get_client().await?.unwrap(); assert_eq!(client.latest_version_id, latest_version_id); assert!(client.snapshot.is_none()); let latest_version_id = Uuid::new_v4(); - txn.add_version(latest_version_id, Uuid::new_v4(), vec![1, 1])?; + txn.add_version(latest_version_id, Uuid::new_v4(), vec![1, 1]) + .await?; - let client = txn.get_client()?.unwrap(); + let client = txn.get_client().await?.unwrap(); assert_eq!(client.latest_version_id, latest_version_id); assert!(client.snapshot.is_none()); @@ -227,38 +230,39 @@ mod test { timestamp: Utc::now(), versions_since: 4, }; - txn.set_snapshot(snap.clone(), vec![1, 2, 3])?; + txn.set_snapshot(snap.clone(), vec![1, 2, 3]).await?; - let client = txn.get_client()?.unwrap(); + let client = txn.get_client().await?.unwrap(); assert_eq!(client.latest_version_id, latest_version_id); assert_eq!(client.snapshot.unwrap(), snap); - txn.commit()?; + txn.commit().await?; Ok(()) } - #[test] - fn test_gvbp_empty() -> anyhow::Result<()> { + #[tokio::test] + async fn test_gvbp_empty() -> anyhow::Result<()> { let storage = InMemoryStorage::new(); let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; - let maybe_version = txn.get_version_by_parent(Uuid::new_v4())?; + let mut txn = storage.txn(client_id).await?; + let maybe_version = txn.get_version_by_parent(Uuid::new_v4()).await?; assert!(maybe_version.is_none()); Ok(()) } - #[test] - fn test_add_version_and_get_version() -> anyhow::Result<()> { + #[tokio::test] + async fn test_add_version_and_get_version() -> anyhow::Result<()> { let storage = InMemoryStorage::new(); let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; + let mut txn = storage.txn(client_id).await?; let version_id = Uuid::new_v4(); let parent_version_id = Uuid::new_v4(); let history_segment = b"abc".to_vec(); - txn.new_client(parent_version_id)?; - txn.add_version(version_id, parent_version_id, history_segment.clone())?; + txn.new_client(parent_version_id).await?; + txn.add_version(version_id, parent_version_id, history_segment.clone()) + .await?; let expected = Version { version_id, @@ -266,74 +270,76 @@ mod test { history_segment, }; - let version = txn.get_version_by_parent(parent_version_id)?.unwrap(); + let version = txn.get_version_by_parent(parent_version_id).await?.unwrap(); assert_eq!(version, expected); - let version = txn.get_version(version_id)?.unwrap(); + let version = txn.get_version(version_id).await?.unwrap(); assert_eq!(version, expected); - txn.commit()?; + txn.commit().await?; Ok(()) } - #[test] - fn test_add_version_exists() -> anyhow::Result<()> { + #[tokio::test] + async fn test_add_version_exists() -> anyhow::Result<()> { let storage = InMemoryStorage::new(); let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; + let mut txn = storage.txn(client_id).await?; let version_id = Uuid::new_v4(); let parent_version_id = Uuid::new_v4(); let history_segment = b"abc".to_vec(); - txn.new_client(parent_version_id)?; - txn.add_version(version_id, parent_version_id, history_segment.clone())?; + txn.new_client(parent_version_id).await?; + txn.add_version(version_id, parent_version_id, history_segment.clone()) + .await?; assert!(txn .add_version(version_id, parent_version_id, history_segment.clone()) + .await .is_err()); - txn.commit()?; + txn.commit().await?; Ok(()) } - #[test] - fn test_snapshots() -> anyhow::Result<()> { + #[tokio::test] + async fn test_snapshots() -> anyhow::Result<()> { let storage = InMemoryStorage::new(); let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; + let mut txn = storage.txn(client_id).await?; - txn.new_client(Uuid::new_v4())?; - assert!(txn.get_client()?.unwrap().snapshot.is_none()); + txn.new_client(Uuid::new_v4()).await?; + assert!(txn.get_client().await?.unwrap().snapshot.is_none()); let snap = Snapshot { version_id: Uuid::new_v4(), timestamp: Utc::now(), versions_since: 3, }; - txn.set_snapshot(snap.clone(), vec![9, 8, 9])?; + txn.set_snapshot(snap.clone(), vec![9, 8, 9]).await?; assert_eq!( - txn.get_snapshot_data(snap.version_id)?.unwrap(), + txn.get_snapshot_data(snap.version_id).await?.unwrap(), vec![9, 8, 9] ); - assert_eq!(txn.get_client()?.unwrap().snapshot, Some(snap)); + assert_eq!(txn.get_client().await?.unwrap().snapshot, Some(snap)); let snap2 = Snapshot { version_id: Uuid::new_v4(), timestamp: Utc::now(), versions_since: 10, }; - txn.set_snapshot(snap2.clone(), vec![0, 2, 4, 6])?; + txn.set_snapshot(snap2.clone(), vec![0, 2, 4, 6]).await?; assert_eq!( - txn.get_snapshot_data(snap2.version_id)?.unwrap(), + txn.get_snapshot_data(snap2.version_id).await?.unwrap(), vec![0, 2, 4, 6] ); - assert_eq!(txn.get_client()?.unwrap().snapshot, Some(snap2)); + assert_eq!(txn.get_client().await?.unwrap().snapshot, Some(snap2)); // check that mismatched version is detected - assert!(txn.get_snapshot_data(Uuid::new_v4()).is_err()); + assert!(txn.get_snapshot_data(Uuid::new_v4()).await.is_err()); - txn.commit()?; + txn.commit().await?; Ok(()) } } diff --git a/core/src/server.rs b/core/src/server.rs index 09222d5..88ae50a 100644 --- a/core/src/server.rs +++ b/core/src/server.rs @@ -106,17 +106,17 @@ impl Server { } /// Implementation of the GetChildVersion protocol transaction. - pub fn get_child_version( + pub async fn get_child_version( &self, client_id: ClientId, parent_version_id: VersionId, ) -> Result { - let mut txn = self.storage.txn(client_id)?; - let client = txn.get_client()?.ok_or(ServerError::NoSuchClient)?; + let mut txn = self.txn(client_id).await?; + let client = txn.get_client().await?.ok_or(ServerError::NoSuchClient)?; // If a version with parentVersionId equal to the requested parentVersionId exists, it is // returned. - if let Some(version) = txn.get_version_by_parent(parent_version_id)? { + if let Some(version) = txn.get_version_by_parent(parent_version_id).await? { return Ok(GetVersionResult::Success { version_id: version.version_id, parent_version_id: version.parent_version_id, @@ -142,7 +142,7 @@ impl Server { } /// Implementation of the AddVersion protocol transaction - pub fn add_version( + pub async fn add_version( &self, client_id: ClientId, parent_version_id: VersionId, @@ -150,8 +150,8 @@ impl Server { ) -> Result<(AddVersionResult, SnapshotUrgency), ServerError> { log::debug!("add_version(client_id: {client_id}, parent_version_id: {parent_version_id})"); - let mut txn = self.storage.txn(client_id)?; - let client = txn.get_client()?.ok_or(ServerError::NoSuchClient)?; + let mut txn = self.txn(client_id).await?; + let client = txn.get_client().await?.ok_or(ServerError::NoSuchClient)?; // check if this version is acceptable, under the protection of the transaction if client.latest_version_id != NIL_VERSION_ID @@ -169,8 +169,9 @@ impl Server { log::debug!("add_version request accepted: new version_id: {version_id}"); // update the DB - txn.add_version(version_id, parent_version_id, history_segment)?; - txn.commit()?; + txn.add_version(version_id, parent_version_id, history_segment) + .await?; + txn.commit().await?; // calculate the urgency let time_urgency = match client.snapshot { @@ -194,7 +195,7 @@ impl Server { } /// Implementation of the AddSnapshot protocol transaction - pub fn add_snapshot( + pub async fn add_snapshot( &self, client_id: ClientId, version_id: VersionId, @@ -202,8 +203,8 @@ impl Server { ) -> Result<(), ServerError> { log::debug!("add_snapshot(client_id: {client_id}, version_id: {version_id})"); - let mut txn = self.storage.txn(client_id)?; - let client = txn.get_client()?.ok_or(ServerError::NoSuchClient)?; + let mut txn = self.txn(client_id).await?; + let client = txn.get_client().await?.ok_or(ServerError::NoSuchClient)?; // NOTE: if the snapshot is rejected, this function logs about it and returns // Ok(()), as there's no reason to report an errot to the client / user. @@ -239,7 +240,7 @@ impl Server { } // get the parent version ID - if let Some(parent) = txn.get_version(vid)? { + if let Some(parent) = txn.get_version(vid).await? { vid = parent.parent_version_id; } else { // this version does not exist; "this should not happen" but if it does, @@ -257,21 +258,23 @@ impl Server { versions_since: 0, }, data, - )?; - txn.commit()?; + ) + .await?; + txn.commit().await?; Ok(()) } /// Implementation of the GetSnapshot protocol transaction - pub fn get_snapshot( + pub async fn get_snapshot( &self, client_id: ClientId, ) -> Result)>, ServerError> { - let mut txn = self.storage.txn(client_id)?; - let client = txn.get_client()?.ok_or(ServerError::NoSuchClient)?; + let mut txn = self.txn(client_id).await?; + let client = txn.get_client().await?.ok_or(ServerError::NoSuchClient)?; Ok(if let Some(snap) = client.snapshot { - txn.get_snapshot_data(snap.version_id)? + txn.get_snapshot_data(snap.version_id) + .await? .map(|data| (snap.version_id, data)) } else { None @@ -279,8 +282,8 @@ impl Server { } /// Convenience method to get a transaction for the embedded storage. - pub fn txn(&self, client_id: Uuid) -> Result, ServerError> { - Ok(self.storage.txn(client_id)?) + pub async fn txn(&self, client_id: Uuid) -> Result, ServerError> { + Ok(self.storage.txn(client_id).await?) } } @@ -288,68 +291,70 @@ impl Server { mod test { use super::*; use crate::inmemory::InMemoryStorage; - use crate::storage::{Snapshot, Storage, StorageTxn}; + use crate::storage::{Snapshot, Storage}; use chrono::{Duration, TimeZone, Utc}; use pretty_assertions::assert_eq; - fn setup(init: INIT) -> anyhow::Result<(Server, RES)> - where - INIT: FnOnce(&mut dyn StorageTxn, Uuid) -> anyhow::Result, - { + /// Set up for a test, returning storage and a client_id. + fn setup() -> (InMemoryStorage, Uuid) { let _ = env_logger::builder().is_test(true).try_init(); let storage = InMemoryStorage::new(); let client_id = Uuid::new_v4(); - let res; - { - let mut txn = storage.txn(client_id)?; - res = init(txn.as_mut(), client_id)?; - txn.commit()?; - } - Ok((Server::new(ServerConfig::default(), storage), res)) + (storage, client_id) } - /// Utility setup function for add_version tests - fn av_setup( + /// Convert storage into a Server. + fn into_server(storage: InMemoryStorage) -> Server { + Server::new(ServerConfig::default(), storage) + } + + /// Add versions to the DB for the given client. + async fn add_versions( + storage: &InMemoryStorage, + client_id: Uuid, num_versions: u32, snapshot_version: Option, snapshot_days_ago: Option, - ) -> anyhow::Result<(Server, Uuid, Vec)> { - let (server, (client_id, versions)) = setup(|txn, client_id| { - let mut versions = vec![]; + ) -> anyhow::Result> { + let mut txn = storage.txn(client_id).await?; + let mut versions = vec![]; - let mut version_id = Uuid::nil(); - txn.new_client(Uuid::nil())?; - debug_assert!(num_versions < u8::MAX.into()); - for vnum in 0..num_versions { - let parent_version_id = version_id; - version_id = Uuid::new_v4(); - versions.push(version_id); - txn.add_version( - version_id, - parent_version_id, - // Generate some unique data for this version. - vec![0, 0, vnum as u8], - )?; - if Some(vnum) == snapshot_version { - txn.set_snapshot( - Snapshot { - version_id, - versions_since: 0, - timestamp: Utc::now() - Duration::days(snapshot_days_ago.unwrap_or(0)), - }, - // Generate some unique data for this snapshot. - vec![vnum as u8], - )?; - } + let mut version_id = Uuid::nil(); + txn.new_client(Uuid::nil()).await?; + assert!( + num_versions < u8::MAX.into(), + "we cast the version number to u8" + ); + for vnum in 0..num_versions { + let parent_version_id = version_id; + version_id = Uuid::new_v4(); + versions.push(version_id); + txn.add_version( + version_id, + parent_version_id, + // Generate some unique data for this version. + vec![0, 0, vnum as u8], + ) + .await?; + if Some(vnum) == snapshot_version { + txn.set_snapshot( + Snapshot { + version_id, + versions_since: 0, + timestamp: Utc::now() - Duration::days(snapshot_days_ago.unwrap_or(0)), + }, + // Generate some unique data for this snapshot. + vec![vnum as u8], + ) + .await?; } - - Ok((client_id, versions)) - })?; - Ok((server, client_id, versions)) + } + txn.commit().await?; + Ok(versions) } /// Utility function to check the results of an add_version call - fn av_success_check( + async fn av_success_check( server: &Server, client_id: Uuid, existing_versions: &[Uuid], @@ -364,17 +369,17 @@ mod test { } // verify that the storage was updated - let mut txn = server.txn(client_id)?; - let client = txn.get_client()?.unwrap(); + let mut txn = server.txn(client_id).await?; + let client = txn.get_client().await?.unwrap(); assert_eq!(client.latest_version_id, new_version_id); let parent_version_id = existing_versions.last().cloned().unwrap_or_else(Uuid::nil); - let version = txn.get_version(new_version_id)?.unwrap(); + let version = txn.get_version(new_version_id).await?.unwrap(); assert_eq!(version.version_id, new_version_id); assert_eq!(version.parent_version_id, parent_version_id); assert_eq!(version.history_segment, expected_history); } else { - panic!("did not get Ok from add_version: {:?}", add_version_result); + panic!("did not get Ok from add_version: {add_version_result:?}"); } assert_eq!(snapshot_urgency, expected_urgency); @@ -426,89 +431,108 @@ mod test { ); } - #[test] - fn get_child_version_not_found_initial_nil() -> anyhow::Result<()> { - let (server, client_id) = setup(|txn, client_id| { - txn.new_client(NIL_VERSION_ID)?; + #[tokio::test] + async fn get_child_version_not_found_initial_nil() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + { + let mut txn = storage.txn(client_id).await?; + txn.new_client(NIL_VERSION_ID).await?; + txn.commit().await?; + } + + let server = into_server(storage); - Ok(client_id) - })?; // when no latest version exists, the first version is NotFound assert_eq!( - server.get_child_version(client_id, NIL_VERSION_ID)?, + server.get_child_version(client_id, NIL_VERSION_ID).await?, GetVersionResult::NotFound ); Ok(()) } - #[test] - fn get_child_version_not_found_initial_continuing() -> anyhow::Result<()> { - let (server, client_id) = setup(|txn, client_id| { - txn.new_client(NIL_VERSION_ID)?; + #[tokio::test] + async fn get_child_version_not_found_initial_continuing() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + { + let mut txn = storage.txn(client_id).await?; + txn.new_client(NIL_VERSION_ID).await?; + txn.commit().await?; + } - Ok(client_id) - })?; + let server = into_server(storage); // when no latest version exists, _any_ child version is NOT_FOUND. This allows syncs to // start to a new server even if the client already has been uploading to another service. assert_eq!( - server.get_child_version(client_id, Uuid::new_v4(),)?, + server.get_child_version(client_id, Uuid::new_v4(),).await?, GetVersionResult::NotFound ); Ok(()) } - #[test] - fn get_child_version_not_found_up_to_date() -> anyhow::Result<()> { - let (server, (client_id, parent_version_id)) = setup(|txn, client_id| { + #[tokio::test] + async fn get_child_version_not_found_up_to_date() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let parent_version_id = Uuid::new_v4(); + { + let mut txn = storage.txn(client_id).await?; // add a parent version, but not the requested child version - let parent_version_id = Uuid::new_v4(); - txn.new_client(parent_version_id)?; - txn.add_version(parent_version_id, NIL_VERSION_ID, vec![])?; - - Ok((client_id, parent_version_id)) - })?; + txn.new_client(parent_version_id).await?; + txn.add_version(parent_version_id, NIL_VERSION_ID, vec![]) + .await?; + txn.commit().await?; + } + let server = into_server(storage); assert_eq!( - server.get_child_version(client_id, parent_version_id)?, + server + .get_child_version(client_id, parent_version_id) + .await?, GetVersionResult::NotFound ); Ok(()) } - #[test] - fn get_child_version_gone_not_latest() -> anyhow::Result<()> { - let (server, client_id) = setup(|txn, client_id| { + #[tokio::test] + async fn get_child_version_gone_not_latest() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let parent_version_id = Uuid::new_v4(); + { + let mut txn = storage.txn(client_id).await?; // Add a parent version, but not the requested parent version - let parent_version_id = Uuid::new_v4(); - txn.new_client(parent_version_id)?; - txn.add_version(parent_version_id, NIL_VERSION_ID, vec![])?; - - Ok(client_id) - })?; + txn.new_client(parent_version_id).await?; + txn.add_version(parent_version_id, NIL_VERSION_ID, vec![]) + .await?; + txn.commit().await?; + } + let server = into_server(storage); assert_eq!( - server.get_child_version(client_id, Uuid::new_v4(),)?, + server.get_child_version(client_id, Uuid::new_v4(),).await?, GetVersionResult::Gone ); Ok(()) } - #[test] - fn get_child_version_found() -> anyhow::Result<()> { - let (server, (client_id, version_id, parent_version_id, history_segment)) = - setup(|txn, client_id| { - let version_id = Uuid::new_v4(); - let parent_version_id = Uuid::new_v4(); - let history_segment = b"abcd".to_vec(); + #[tokio::test] + async fn get_child_version_found() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let version_id = Uuid::new_v4(); + let parent_version_id = Uuid::new_v4(); + let history_segment = b"abcd".to_vec(); + { + let mut txn = storage.txn(client_id).await?; + txn.new_client(version_id).await?; + txn.add_version(version_id, parent_version_id, history_segment.clone()) + .await?; + txn.commit().await?; + } - txn.new_client(version_id)?; - txn.add_version(version_id, parent_version_id, history_segment.clone())?; - - Ok((client_id, version_id, parent_version_id, history_segment)) - })?; + let server = into_server(storage); assert_eq!( - server.get_child_version(client_id, parent_version_id)?, + server + .get_child_version(client_id, parent_version_id) + .await?, GetVersionResult::Success { version_id, parent_version_id, @@ -518,29 +542,41 @@ mod test { Ok(()) } - #[test] - fn add_version_conflict() -> anyhow::Result<()> { - let (server, client_id, versions) = av_setup(3, None, None)?; + #[tokio::test] + async fn add_version_conflict() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let versions = add_versions(&storage, client_id, 3, None, None).await?; // try to add a child of a version other than the latest + let server = into_server(storage); assert_eq!( - server.add_version(client_id, versions[1], vec![3, 6, 9])?.0, + server + .add_version(client_id, versions[1], vec![3, 6, 9]) + .await? + .0, AddVersionResult::ExpectedParentVersion(versions[2]) ); // verify that the storage wasn't updated - let mut txn = server.txn(client_id)?; - assert_eq!(txn.get_client()?.unwrap().latest_version_id, versions[2]); - assert_eq!(txn.get_version_by_parent(versions[2])?, None); + let mut txn = server.txn(client_id).await?; + assert_eq!( + txn.get_client().await?.unwrap().latest_version_id, + versions[2] + ); + assert_eq!(txn.get_version_by_parent(versions[2]).await?, None); Ok(()) } - #[test] - fn add_version_with_existing_history() -> anyhow::Result<()> { - let (server, client_id, versions) = av_setup(1, None, None)?; + #[tokio::test] + async fn add_version_with_existing_history() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let versions = add_versions(&storage, client_id, 1, None, None).await?; - let result = server.add_version(client_id, versions[0], vec![3, 6, 9])?; + let server = into_server(storage); + let result = server + .add_version(client_id, versions[0], vec![3, 6, 9]) + .await?; av_success_check( &server, @@ -550,17 +586,22 @@ mod test { vec![3, 6, 9], // urgency=high because there are no snapshots yet SnapshotUrgency::High, - )?; + ) + .await?; Ok(()) } - #[test] - fn add_version_with_no_history() -> anyhow::Result<()> { - let (server, client_id, versions) = av_setup(0, None, None)?; + #[tokio::test] + async fn add_version_with_no_history() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let versions = add_versions(&storage, client_id, 0, None, None).await?; + let server = into_server(storage); let parent_version_id = Uuid::nil(); - let result = server.add_version(client_id, parent_version_id, vec![3, 6, 9])?; + let result = server + .add_version(client_id, parent_version_id, vec![3, 6, 9]) + .await?; av_success_check( &server, @@ -570,16 +611,21 @@ mod test { vec![3, 6, 9], // urgency=high because there are no snapshots yet SnapshotUrgency::High, - )?; + ) + .await?; Ok(()) } - #[test] - fn add_version_success_recent_snapshot() -> anyhow::Result<()> { - let (server, client_id, versions) = av_setup(1, Some(0), None)?; + #[tokio::test] + async fn add_version_success_recent_snapshot() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let versions = add_versions(&storage, client_id, 1, Some(0), None).await?; - let result = server.add_version(client_id, versions[0], vec![1, 2, 3])?; + let server = into_server(storage); + let result = server + .add_version(client_id, versions[0], vec![1, 2, 3]) + .await?; av_success_check( &server, @@ -589,17 +635,22 @@ mod test { vec![1, 2, 3], // no snapshot request since the previous version has a snapshot SnapshotUrgency::None, - )?; + ) + .await?; Ok(()) } - #[test] - fn add_version_success_aged_snapshot() -> anyhow::Result<()> { + #[tokio::test] + async fn add_version_success_aged_snapshot() -> anyhow::Result<()> { // one snapshot, but it was 50 days ago - let (server, client_id, versions) = av_setup(1, Some(0), Some(50))?; + let (storage, client_id) = setup(); + let versions = add_versions(&storage, client_id, 1, Some(0), Some(50)).await?; - let result = server.add_version(client_id, versions[0], vec![1, 2, 3])?; + let server = into_server(storage); + let result = server + .add_version(client_id, versions[0], vec![1, 2, 3]) + .await?; av_success_check( &server, @@ -609,18 +660,24 @@ mod test { vec![1, 2, 3], // urgency=high due to days since the snapshot SnapshotUrgency::High, - )?; + ) + .await?; Ok(()) } - #[test] - fn add_version_success_snapshot_many_versions_ago() -> anyhow::Result<()> { + #[tokio::test] + async fn add_version_success_snapshot_many_versions_ago() -> anyhow::Result<()> { // one snapshot, but it was 50 versions ago - let (mut server, client_id, versions) = av_setup(50, Some(0), None)?; + let (storage, client_id) = setup(); + let versions = add_versions(&storage, client_id, 50, Some(0), None).await?; + + let mut server = into_server(storage); server.config.snapshot_versions = 30; - let result = server.add_version(client_id, versions[49], vec![1, 2, 3])?; + let result = server + .add_version(client_id, versions[49], vec![1, 2, 3]) + .await?; av_success_check( &server, @@ -630,136 +687,165 @@ mod test { vec![1, 2, 3], // urgency=high due to number of versions since the snapshot SnapshotUrgency::High, - )?; + ) + .await?; Ok(()) } - #[test] - fn add_snapshot_success_latest() -> anyhow::Result<()> { - let (server, (client_id, version_id)) = setup(|txn, client_id| { - let version_id = Uuid::new_v4(); + #[tokio::test] + async fn add_snapshot_success_latest() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let version_id = Uuid::new_v4(); + { + let mut txn = storage.txn(client_id).await?; // set up a task DB with one version in it - txn.new_client(version_id)?; - txn.add_version(version_id, NIL_VERSION_ID, vec![])?; + txn.new_client(version_id).await?; + txn.add_version(version_id, NIL_VERSION_ID, vec![]).await?; - // add a snapshot for that version - Ok((client_id, version_id)) - })?; - server.add_snapshot(client_id, version_id, vec![1, 2, 3])?; + txn.commit().await?; + } + + let server = into_server(storage); + server + .add_snapshot(client_id, version_id, vec![1, 2, 3]) + .await?; // verify the snapshot - let mut txn = server.txn(client_id)?; - let client = txn.get_client()?.unwrap(); + let mut txn = server.txn(client_id).await?; + let client = txn.get_client().await?.unwrap(); let snapshot = client.snapshot.unwrap(); assert_eq!(snapshot.version_id, version_id); assert_eq!(snapshot.versions_since, 0); assert_eq!( - txn.get_snapshot_data(version_id).unwrap(), + txn.get_snapshot_data(version_id).await.unwrap(), Some(vec![1, 2, 3]) ); Ok(()) } - #[test] - fn add_snapshot_success_older() -> anyhow::Result<()> { - let (server, (client_id, version_id_1)) = setup(|txn, client_id| { - let version_id_1 = Uuid::new_v4(); - let version_id_2 = Uuid::new_v4(); + #[tokio::test] + async fn add_snapshot_success_older() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let version_id_1 = Uuid::new_v4(); + let version_id_2 = Uuid::new_v4(); + { + let mut txn = storage.txn(client_id).await?; // set up a task DB with two versions in it - txn.new_client(version_id_2)?; - txn.add_version(version_id_1, NIL_VERSION_ID, vec![])?; - txn.add_version(version_id_2, version_id_1, vec![])?; + txn.new_client(version_id_2).await?; + txn.add_version(version_id_1, NIL_VERSION_ID, vec![]) + .await?; + txn.add_version(version_id_2, version_id_1, vec![]).await?; + + txn.commit().await?; + } - Ok((client_id, version_id_1)) - })?; // add a snapshot for version 1 - server.add_snapshot(client_id, version_id_1, vec![1, 2, 3])?; + let server = into_server(storage); + server + .add_snapshot(client_id, version_id_1, vec![1, 2, 3]) + .await?; // verify the snapshot - let mut txn = server.txn(client_id)?; - let client = txn.get_client()?.unwrap(); + let mut txn = server.txn(client_id).await?; + let client = txn.get_client().await?.unwrap(); let snapshot = client.snapshot.unwrap(); assert_eq!(snapshot.version_id, version_id_1); assert_eq!(snapshot.versions_since, 0); assert_eq!( - txn.get_snapshot_data(version_id_1).unwrap(), + txn.get_snapshot_data(version_id_1).await.unwrap(), Some(vec![1, 2, 3]) ); Ok(()) } - #[test] - fn add_snapshot_fails_no_such() -> anyhow::Result<()> { - let (server, client_id) = setup(|txn, client_id| { - let version_id_1 = Uuid::new_v4(); - let version_id_2 = Uuid::new_v4(); + #[tokio::test] + async fn add_snapshot_fails_no_such() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let version_id_1 = Uuid::new_v4(); + let version_id_2 = Uuid::new_v4(); + { + let mut txn = storage.txn(client_id).await?; // set up a task DB with two versions in it - txn.new_client(version_id_2)?; - txn.add_version(version_id_1, NIL_VERSION_ID, vec![])?; - txn.add_version(version_id_2, version_id_1, vec![])?; + txn.new_client(version_id_2).await?; + txn.add_version(version_id_1, NIL_VERSION_ID, vec![]) + .await?; + txn.add_version(version_id_2, version_id_1, vec![]).await?; - // add a snapshot for unknown version - Ok(client_id) - })?; + txn.commit().await?; + } + // add a snapshot for unknown version + let server = into_server(storage); let version_id_unk = Uuid::new_v4(); - server.add_snapshot(client_id, version_id_unk, vec![1, 2, 3])?; + server + .add_snapshot(client_id, version_id_unk, vec![1, 2, 3]) + .await?; // verify the snapshot does not exist - let mut txn = server.txn(client_id)?; - let client = txn.get_client()?.unwrap(); + let mut txn = server.txn(client_id).await?; + let client = txn.get_client().await?.unwrap(); assert!(client.snapshot.is_none()); Ok(()) } - #[test] - fn add_snapshot_fails_too_old() -> anyhow::Result<()> { - let (server, (client_id, version_ids)) = setup(|txn, client_id| { - let mut version_id = Uuid::new_v4(); - let mut parent_version_id = Uuid::nil(); - let mut version_ids = vec![]; + #[tokio::test] + async fn add_snapshot_fails_too_old() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let mut version_id = Uuid::new_v4(); + let mut parent_version_id = Uuid::nil(); + let mut version_ids = vec![]; + { + let mut txn = storage.txn(client_id).await?; // set up a task DB with 10 versions in it (oldest to newest) - txn.new_client(Uuid::nil())?; + txn.new_client(Uuid::nil()).await?; for _ in 0..10 { - txn.add_version(version_id, parent_version_id, vec![])?; + txn.add_version(version_id, parent_version_id, vec![]) + .await?; version_ids.push(version_id); parent_version_id = version_id; version_id = Uuid::new_v4(); } - // add a snapshot for the earliest of those - Ok((client_id, version_ids)) - })?; - server.add_snapshot(client_id, version_ids[0], vec![1, 2, 3])?; + txn.commit().await?; + } + + // add a snapshot for the earliest of those + let server = into_server(storage); + server + .add_snapshot(client_id, version_ids[0], vec![1, 2, 3]) + .await?; // verify the snapshot does not exist - let mut txn = server.txn(client_id)?; - let client = txn.get_client()?.unwrap(); + let mut txn = server.txn(client_id).await?; + let client = txn.get_client().await?.unwrap(); assert!(client.snapshot.is_none()); Ok(()) } - #[test] - fn add_snapshot_fails_newer_exists() -> anyhow::Result<()> { - let (server, (client_id, version_ids)) = setup(|txn, client_id| { - let mut version_id = Uuid::new_v4(); - let mut parent_version_id = Uuid::nil(); - let mut version_ids = vec![]; + #[tokio::test] + async fn add_snapshot_fails_newer_exists() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let mut version_id = Uuid::new_v4(); + let mut parent_version_id = Uuid::nil(); + let mut version_ids = vec![]; + { + let mut txn = storage.txn(client_id).await?; // set up a task DB with 5 versions in it (oldest to newest) and a snapshot of the // middle one - txn.new_client(Uuid::nil())?; + txn.new_client(Uuid::nil()).await?; for _ in 0..5 { - txn.add_version(version_id, parent_version_id, vec![])?; + txn.add_version(version_id, parent_version_id, vec![]) + .await?; version_ids.push(version_id); parent_version_id = version_id; version_id = Uuid::new_v4(); @@ -771,55 +857,64 @@ mod test { timestamp: Utc.with_ymd_and_hms(2001, 9, 9, 1, 46, 40).unwrap(), }, vec![1, 2, 3], - )?; + ) + .await?; - // add a snapshot for the earliest of those - Ok((client_id, version_ids)) - })?; + txn.commit().await?; + } - server.add_snapshot(client_id, version_ids[0], vec![9, 9, 9])?; + // add a snapshot for the earliest of those + let server = into_server(storage); + server + .add_snapshot(client_id, version_ids[0], vec![9, 9, 9]) + .await?; // verify the snapshot was not replaced - let mut txn = server.txn(client_id)?; - let client = txn.get_client()?.unwrap(); + let mut txn = server.txn(client_id).await?; + let client = txn.get_client().await?.unwrap(); let snapshot = client.snapshot.unwrap(); assert_eq!(snapshot.version_id, version_ids[2]); assert_eq!(snapshot.versions_since, 2); assert_eq!( - txn.get_snapshot_data(version_ids[2]).unwrap(), + txn.get_snapshot_data(version_ids[2]).await.unwrap(), Some(vec![1, 2, 3]) ); Ok(()) } - #[test] - fn add_snapshot_fails_nil_version() -> anyhow::Result<()> { - let (server, client_id) = setup(|txn, client_id| { + #[tokio::test] + async fn add_snapshot_fails_nil_version() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + { + let mut txn = storage.txn(client_id).await?; // just set up the client - txn.new_client(NIL_VERSION_ID)?; + txn.new_client(NIL_VERSION_ID).await?; + txn.commit().await?; + } - // add a snapshot for the nil version - Ok(client_id) - })?; - - server.add_snapshot(client_id, NIL_VERSION_ID, vec![9, 9, 9])?; + let server = into_server(storage); + server + .add_snapshot(client_id, NIL_VERSION_ID, vec![9, 9, 9]) + .await?; // verify the snapshot does not exist - let mut txn = server.txn(client_id)?; - let client = txn.get_client()?.unwrap(); + let mut txn = server.txn(client_id).await?; + let client = txn.get_client().await?.unwrap(); assert!(client.snapshot.is_none()); Ok(()) } - #[test] - fn get_snapshot_found() -> anyhow::Result<()> { - let (server, (client_id, data, snapshot_version_id)) = setup(|txn, client_id| { - let data = vec![1, 2, 3]; - let snapshot_version_id = Uuid::new_v4(); + #[tokio::test] + async fn get_snapshot_found() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + let data = vec![1, 2, 3]; + let snapshot_version_id = Uuid::new_v4(); - txn.new_client(snapshot_version_id)?; + { + let mut txn = storage.txn(client_id).await?; + txn.new_client(snapshot_version_id).await?; txn.set_snapshot( Snapshot { version_id: snapshot_version_id, @@ -827,25 +922,31 @@ mod test { timestamp: Utc.with_ymd_and_hms(2001, 9, 9, 1, 46, 40).unwrap(), }, data.clone(), - )?; - Ok((client_id, data, snapshot_version_id)) - })?; + ) + .await?; + txn.commit().await?; + } + + let server = into_server(storage); assert_eq!( - server.get_snapshot(client_id)?, + server.get_snapshot(client_id).await?, Some((snapshot_version_id, data)) ); Ok(()) } - #[test] - fn get_snapshot_not_found() -> anyhow::Result<()> { - let (server, client_id) = setup(|txn, client_id| { - txn.new_client(NIL_VERSION_ID)?; - Ok(client_id) - })?; + #[tokio::test] + async fn get_snapshot_not_found() -> anyhow::Result<()> { + let (storage, client_id) = setup(); + { + let mut txn = storage.txn(client_id).await?; + txn.new_client(NIL_VERSION_ID).await?; + txn.commit().await?; + } - assert_eq!(server.get_snapshot(client_id)?, None); + let server = into_server(storage); + assert_eq!(server.get_snapshot(client_id).await?, None); Ok(()) } diff --git a/core/src/storage.rs b/core/src/storage.rs index 011f231..e1c1882 100644 --- a/core/src/storage.rs +++ b/core/src/storage.rs @@ -44,32 +44,35 @@ pub struct Version { /// /// Changes in a transaction that is dropped without calling `commit` must not appear in any other /// transaction. +#[async_trait::async_trait(?Send)] pub trait StorageTxn { /// Get information about the client for this transaction - fn get_client(&mut self) -> anyhow::Result>; + async fn get_client(&mut self) -> anyhow::Result>; /// Create the client for this transaction, with the given latest_version_id. The client must /// not already exist. - fn new_client(&mut self, latest_version_id: Uuid) -> anyhow::Result<()>; + async fn new_client(&mut self, latest_version_id: Uuid) -> anyhow::Result<()>; /// Set the client's most recent snapshot. - fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec) -> anyhow::Result<()>; + async fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec) -> anyhow::Result<()>; /// Get the data for the most recent snapshot. The version_id /// is used to verify that the snapshot is for the correct version. - fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result>>; + async fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result>>; /// Get a version, indexed by parent version id - fn get_version_by_parent(&mut self, parent_version_id: Uuid) - -> anyhow::Result>; + async fn get_version_by_parent( + &mut self, + parent_version_id: Uuid, + ) -> anyhow::Result>; /// Get a version, indexed by its own version id - fn get_version(&mut self, version_id: Uuid) -> anyhow::Result>; + async fn get_version(&mut self, version_id: Uuid) -> anyhow::Result>; /// Add a version (that must not already exist), and /// - update latest_version_id /// - increment snapshot.versions_since - fn add_version( + async fn add_version( &mut self, version_id: Uuid, parent_version_id: Uuid, @@ -78,12 +81,13 @@ pub trait StorageTxn { /// Commit any changes made in the transaction. It is an error to call this more than /// once. It is safe to skip this call for read-only operations. - fn commit(&mut self) -> anyhow::Result<()>; + async fn commit(&mut self) -> anyhow::Result<()>; } /// A trait for objects able to act as storage. Most of the interesting behavior is in the /// [`crate::storage::StorageTxn`] trait. +#[async_trait::async_trait] pub trait Storage: Send + Sync { /// Begin a transaction for the given client ID. - fn txn(&self, client_id: Uuid) -> anyhow::Result>; + async fn txn(&self, client_id: Uuid) -> anyhow::Result>; } diff --git a/server/Cargo.toml b/server/Cargo.toml index 9aea205..6c9d2fb 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "taskchampion-sync-server" -version = "0.6.2-pre" +version = "0.7.0-pre" authors = ["Dustin J. Mitchell "] edition = "2021" publish = false diff --git a/server/src/api/add_snapshot.rs b/server/src/api/add_snapshot.rs index da6b654..13914eb 100644 --- a/server/src/api/add_snapshot.rs +++ b/server/src/api/add_snapshot.rs @@ -49,6 +49,7 @@ pub(crate) async fn service( server_state .server .add_snapshot(client_id, version_id, body.to_vec()) + .await .map_err(server_error_to_actix)?; Ok(HttpResponse::Ok().body("")) } @@ -70,10 +71,10 @@ mod test { // set up the storage contents.. { - let mut txn = storage.txn(client_id).unwrap(); - txn.new_client(version_id).unwrap(); - txn.add_version(version_id, NIL_VERSION_ID, vec![])?; - txn.commit()?; + let mut txn = storage.txn(client_id).await.unwrap(); + txn.new_client(version_id).await.unwrap(); + txn.add_version(version_id, NIL_VERSION_ID, vec![]).await?; + txn.commit().await?; } let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); @@ -114,9 +115,9 @@ mod test { // set up the storage contents.. { - let mut txn = storage.txn(client_id).unwrap(); - txn.new_client(NIL_VERSION_ID).unwrap(); - txn.commit().unwrap(); + let mut txn = storage.txn(client_id).await.unwrap(); + txn.new_client(NIL_VERSION_ID).await.unwrap(); + txn.commit().await.unwrap(); } let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); diff --git a/server/src/api/add_version.rs b/server/src/api/add_version.rs index 7f7e57b..77d8630 100644 --- a/server/src/api/add_version.rs +++ b/server/src/api/add_version.rs @@ -60,6 +60,7 @@ pub(crate) async fn service( return match server_state .server .add_version(client_id, parent_version_id, body.to_vec()) + .await { Ok((AddVersionResult::Ok(version_id), snap_urgency)) => { let mut rb = HttpResponse::Ok(); @@ -85,9 +86,12 @@ pub(crate) async fn service( let mut txn = server_state .server .txn(client_id) + .await .map_err(server_error_to_actix)?; - txn.new_client(NIL_VERSION_ID).map_err(failure_to_ise)?; - txn.commit().map_err(failure_to_ise)?; + txn.new_client(NIL_VERSION_ID) + .await + .map_err(failure_to_ise)?; + txn.commit().await.map_err(failure_to_ise)?; continue; } Err(e) => Err(server_error_to_actix(e)), @@ -113,9 +117,9 @@ mod test { // set up the storage contents.. { - let mut txn = storage.txn(client_id).unwrap(); - txn.new_client(Uuid::nil()).unwrap(); - txn.commit().unwrap(); + let mut txn = storage.txn(client_id).await.unwrap(); + txn.new_client(Uuid::nil()).await.unwrap(); + txn.commit().await.unwrap(); } let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); @@ -187,8 +191,8 @@ mod test { // Check that the client really was created { - let mut txn = server.server_state.server.txn(client_id).unwrap(); - let client = txn.get_client().unwrap().unwrap(); + let mut txn = server.server_state.server.txn(client_id).await.unwrap(); + let client = txn.get_client().await.unwrap().unwrap(); assert_eq!(client.latest_version_id, new_version_id); assert_eq!(client.snapshot, None); } @@ -233,9 +237,9 @@ mod test { // set up the storage contents.. { - let mut txn = storage.txn(client_id).unwrap(); - txn.new_client(version_id).unwrap(); - txn.commit().unwrap(); + let mut txn = storage.txn(client_id).await.unwrap(); + txn.new_client(version_id).await.unwrap(); + txn.commit().await.unwrap(); } let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); diff --git a/server/src/api/get_child_version.rs b/server/src/api/get_child_version.rs index 9f21b0f..1a70e53 100644 --- a/server/src/api/get_child_version.rs +++ b/server/src/api/get_child_version.rs @@ -26,6 +26,7 @@ pub(crate) async fn service( match server_state .server .get_child_version(client_id, parent_version_id) + .await { Ok(GetVersionResult::Success { version_id, @@ -64,11 +65,12 @@ mod test { // set up the storage contents.. { - let mut txn = storage.txn(client_id).unwrap(); - txn.new_client(Uuid::new_v4()).unwrap(); + let mut txn = storage.txn(client_id).await.unwrap(); + txn.new_client(Uuid::new_v4()).await.unwrap(); txn.add_version(version_id, parent_version_id, b"abcd".to_vec()) + .await .unwrap(); - txn.commit().unwrap(); + txn.commit().await.unwrap(); } let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); @@ -128,11 +130,12 @@ mod test { // create the client and a single version. { - let mut txn = storage.txn(client_id).unwrap(); - txn.new_client(Uuid::new_v4()).unwrap(); + let mut txn = storage.txn(client_id).await.unwrap(); + txn.new_client(Uuid::new_v4()).await.unwrap(); txn.add_version(test_version_id, NIL_VERSION_ID, b"vers".to_vec()) + .await .unwrap(); - txn.commit().unwrap(); + txn.commit().await.unwrap(); } let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); let app = App::new().configure(|sc| server.config(sc)); diff --git a/server/src/api/get_snapshot.rs b/server/src/api/get_snapshot.rs index d7aabdb..3940662 100644 --- a/server/src/api/get_snapshot.rs +++ b/server/src/api/get_snapshot.rs @@ -20,6 +20,7 @@ pub(crate) async fn service( if let Some((version_id, data)) = server_state .server .get_snapshot(client_id) + .await .map_err(server_error_to_actix)? { Ok(HttpResponse::Ok() @@ -48,9 +49,9 @@ mod test { // set up the storage contents.. { - let mut txn = storage.txn(client_id).unwrap(); - txn.new_client(Uuid::new_v4()).unwrap(); - txn.commit().unwrap(); + let mut txn = storage.txn(client_id).await.unwrap(); + txn.new_client(Uuid::new_v4()).await.unwrap(); + txn.commit().await.unwrap(); } let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); @@ -75,8 +76,8 @@ mod test { // set up the storage contents.. { - let mut txn = storage.txn(client_id).unwrap(); - txn.new_client(Uuid::new_v4()).unwrap(); + let mut txn = storage.txn(client_id).await.unwrap(); + txn.new_client(Uuid::new_v4()).await.unwrap(); txn.set_snapshot( Snapshot { version_id, @@ -85,8 +86,9 @@ mod test { }, snapshot_data.clone(), ) + .await .unwrap(); - txn.commit().unwrap(); + txn.commit().await.unwrap(); } let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); diff --git a/sqlite/Cargo.toml b/sqlite/Cargo.toml index 6718f19..c3c03d0 100644 --- a/sqlite/Cargo.toml +++ b/sqlite/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "taskchampion-sync-server-storage-sqlite" -version = "0.6.2-pre" +version = "0.7.0-pre" authors = ["Dustin J. Mitchell "] edition = "2021" description = "SQLite backend for TaskChampion-sync-server" @@ -9,7 +9,8 @@ repository = "https://github.com/GothenburgBitFactory/taskchampion-sync-server" license = "MIT" [dependencies] -taskchampion-sync-server-core = { path = "../core", version = "0.6.2-pre" } +taskchampion-sync-server-core = { path = "../core", version = "0.7.0-pre" } +async-trait.workspace = true uuid.workspace = true anyhow.workspace = true thiserror.workspace = true @@ -19,3 +20,4 @@ chrono.workspace = true [dev-dependencies] tempfile.workspace = true pretty_assertions.workspace = true +tokio.workspace = true diff --git a/sqlite/src/lib.rs b/sqlite/src/lib.rs index bd9f93f..233a658 100644 --- a/sqlite/src/lib.rs +++ b/sqlite/src/lib.rs @@ -77,8 +77,9 @@ impl SqliteStorage { } } +#[async_trait::async_trait] impl Storage for SqliteStorage { - fn txn(&self, client_id: Uuid) -> anyhow::Result> { + async fn txn(&self, client_id: Uuid) -> anyhow::Result> { let con = self.new_connection()?; // Begin the transaction on this new connection. An IMMEDIATE connection is in // write (exclusive) mode from the start. @@ -126,8 +127,9 @@ impl Txn { } } +#[async_trait::async_trait(?Send)] impl StorageTxn for Txn { - fn get_client(&mut self) -> anyhow::Result> { + async fn get_client(&mut self) -> anyhow::Result> { let result: Option = self .con .query_row( @@ -171,7 +173,7 @@ impl StorageTxn for Txn { Ok(result) } - fn new_client(&mut self, latest_version_id: Uuid) -> anyhow::Result<()> { + async fn new_client(&mut self, latest_version_id: Uuid) -> anyhow::Result<()> { self.con .execute( "INSERT OR REPLACE INTO clients (client_id, latest_version_id) VALUES (?, ?)", @@ -181,7 +183,7 @@ impl StorageTxn for Txn { Ok(()) } - fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec) -> anyhow::Result<()> { + async fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec) -> anyhow::Result<()> { self.con .execute( "UPDATE clients @@ -203,7 +205,7 @@ impl StorageTxn for Txn { Ok(()) } - fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result>> { + async fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result>> { let r = self .con .query_row( @@ -227,7 +229,7 @@ impl StorageTxn for Txn { .transpose() } - fn get_version_by_parent( + async fn get_version_by_parent( &mut self, parent_version_id: Uuid, @@ -238,14 +240,14 @@ impl StorageTxn for Txn { parent_version_id) } - fn get_version(&mut self, version_id: Uuid) -> anyhow::Result> { + async fn get_version(&mut self, version_id: Uuid) -> anyhow::Result> { self.get_version_impl( "SELECT version_id, parent_version_id, history_segment FROM versions WHERE version_id = ? AND client_id = ?", self.client_id, version_id) } - fn add_version( + async fn add_version( &mut self, version_id: Uuid, @@ -276,7 +278,7 @@ impl StorageTxn for Txn { Ok(()) } - fn commit(&mut self) -> anyhow::Result<()> { + async fn commit(&mut self) -> anyhow::Result<()> { self.con.execute("COMMIT", [])?; Ok(()) } @@ -289,47 +291,48 @@ mod test { use pretty_assertions::assert_eq; use tempfile::TempDir; - #[test] - fn test_emtpy_dir() -> anyhow::Result<()> { + #[tokio::test] + async fn test_emtpy_dir() -> anyhow::Result<()> { let tmp_dir = TempDir::new()?; let non_existant = tmp_dir.path().join("subdir"); let storage = SqliteStorage::new(non_existant)?; let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; - let maybe_client = txn.get_client()?; + let mut txn = storage.txn(client_id).await?; + let maybe_client = txn.get_client().await?; assert!(maybe_client.is_none()); Ok(()) } - #[test] - fn test_get_client_empty() -> anyhow::Result<()> { + #[tokio::test] + async fn test_get_client_empty() -> anyhow::Result<()> { let tmp_dir = TempDir::new()?; let storage = SqliteStorage::new(tmp_dir.path())?; let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; - let maybe_client = txn.get_client()?; + let mut txn = storage.txn(client_id).await?; + let maybe_client = txn.get_client().await?; assert!(maybe_client.is_none()); Ok(()) } - #[test] - fn test_client_storage() -> anyhow::Result<()> { + #[tokio::test] + async fn test_client_storage() -> anyhow::Result<()> { let tmp_dir = TempDir::new()?; let storage = SqliteStorage::new(tmp_dir.path())?; let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; + let mut txn = storage.txn(client_id).await?; let latest_version_id = Uuid::new_v4(); - txn.new_client(latest_version_id)?; + txn.new_client(latest_version_id).await?; - let client = txn.get_client()?.unwrap(); + let client = txn.get_client().await?.unwrap(); assert_eq!(client.latest_version_id, latest_version_id); assert!(client.snapshot.is_none()); let latest_version_id = Uuid::new_v4(); - txn.add_version(latest_version_id, Uuid::new_v4(), vec![1, 1])?; + txn.add_version(latest_version_id, Uuid::new_v4(), vec![1, 1]) + .await?; - let client = txn.get_client()?.unwrap(); + let client = txn.get_client().await?.unwrap(); assert_eq!(client.latest_version_id, latest_version_id); assert!(client.snapshot.is_none()); @@ -338,37 +341,38 @@ mod test { timestamp: "2014-11-28T12:00:09Z".parse::>().unwrap(), versions_since: 4, }; - txn.set_snapshot(snap.clone(), vec![1, 2, 3])?; + txn.set_snapshot(snap.clone(), vec![1, 2, 3]).await?; - let client = txn.get_client()?.unwrap(); + let client = txn.get_client().await?.unwrap(); assert_eq!(client.latest_version_id, latest_version_id); assert_eq!(client.snapshot.unwrap(), snap); Ok(()) } - #[test] - fn test_gvbp_empty() -> anyhow::Result<()> { + #[tokio::test] + async fn test_gvbp_empty() -> anyhow::Result<()> { let tmp_dir = TempDir::new()?; let storage = SqliteStorage::new(tmp_dir.path())?; let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; - let maybe_version = txn.get_version_by_parent(Uuid::new_v4())?; + let mut txn = storage.txn(client_id).await?; + let maybe_version = txn.get_version_by_parent(Uuid::new_v4()).await?; assert!(maybe_version.is_none()); Ok(()) } - #[test] - fn test_add_version_and_get_version() -> anyhow::Result<()> { + #[tokio::test] + async fn test_add_version_and_get_version() -> anyhow::Result<()> { let tmp_dir = TempDir::new()?; let storage = SqliteStorage::new(tmp_dir.path())?; let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; + let mut txn = storage.txn(client_id).await?; let version_id = Uuid::new_v4(); let parent_version_id = Uuid::new_v4(); let history_segment = b"abc".to_vec(); - txn.add_version(version_id, parent_version_id, history_segment.clone())?; + txn.add_version(version_id, parent_version_id, history_segment.clone()) + .await?; let expected = Version { version_id, @@ -376,70 +380,72 @@ mod test { history_segment, }; - let version = txn.get_version_by_parent(parent_version_id)?.unwrap(); + let version = txn.get_version_by_parent(parent_version_id).await?.unwrap(); assert_eq!(version, expected); - let version = txn.get_version(version_id)?.unwrap(); + let version = txn.get_version(version_id).await?.unwrap(); assert_eq!(version, expected); Ok(()) } - #[test] - fn test_add_version_exists() -> anyhow::Result<()> { + #[tokio::test] + async fn test_add_version_exists() -> anyhow::Result<()> { let tmp_dir = TempDir::new()?; let storage = SqliteStorage::new(tmp_dir.path())?; let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; + let mut txn = storage.txn(client_id).await?; let version_id = Uuid::new_v4(); let parent_version_id = Uuid::new_v4(); let history_segment = b"abc".to_vec(); - txn.add_version(version_id, parent_version_id, history_segment.clone())?; + txn.add_version(version_id, parent_version_id, history_segment.clone()) + .await?; assert!(txn .add_version(version_id, parent_version_id, history_segment.clone()) + .await .is_err()); Ok(()) } - #[test] - fn test_snapshots() -> anyhow::Result<()> { + #[tokio::test] + async fn test_snapshots() -> anyhow::Result<()> { let tmp_dir = TempDir::new()?; let storage = SqliteStorage::new(tmp_dir.path())?; let client_id = Uuid::new_v4(); - let mut txn = storage.txn(client_id)?; + let mut txn = storage.txn(client_id).await?; - txn.new_client(Uuid::new_v4())?; - assert!(txn.get_client()?.unwrap().snapshot.is_none()); + txn.new_client(Uuid::new_v4()).await?; + assert!(txn.get_client().await?.unwrap().snapshot.is_none()); let snap = Snapshot { version_id: Uuid::new_v4(), timestamp: "2013-10-08T12:00:09Z".parse::>().unwrap(), versions_since: 3, }; - txn.set_snapshot(snap.clone(), vec![9, 8, 9])?; + txn.set_snapshot(snap.clone(), vec![9, 8, 9]).await?; assert_eq!( - txn.get_snapshot_data(snap.version_id)?.unwrap(), + txn.get_snapshot_data(snap.version_id).await?.unwrap(), vec![9, 8, 9] ); - assert_eq!(txn.get_client()?.unwrap().snapshot, Some(snap)); + assert_eq!(txn.get_client().await?.unwrap().snapshot, Some(snap)); let snap2 = Snapshot { version_id: Uuid::new_v4(), timestamp: "2014-11-28T12:00:09Z".parse::>().unwrap(), versions_since: 10, }; - txn.set_snapshot(snap2.clone(), vec![0, 2, 4, 6])?; + txn.set_snapshot(snap2.clone(), vec![0, 2, 4, 6]).await?; assert_eq!( - txn.get_snapshot_data(snap2.version_id)?.unwrap(), + txn.get_snapshot_data(snap2.version_id).await?.unwrap(), vec![0, 2, 4, 6] ); - assert_eq!(txn.get_client()?.unwrap().snapshot, Some(snap2)); + assert_eq!(txn.get_client().await?.unwrap().snapshot, Some(snap2)); // check that mismatched version is detected - assert!(txn.get_snapshot_data(Uuid::new_v4()).is_err()); + assert!(txn.get_snapshot_data(Uuid::new_v4()).await.is_err()); Ok(()) } diff --git a/sqlite/tests/concurrency.rs b/sqlite/tests/concurrency.rs index 5a3a37a..55f3209 100644 --- a/sqlite/tests/concurrency.rs +++ b/sqlite/tests/concurrency.rs @@ -2,45 +2,54 @@ use std::thread; use taskchampion_sync_server_core::{Storage, NIL_VERSION_ID}; use taskchampion_sync_server_storage_sqlite::SqliteStorage; use tempfile::TempDir; +use tokio::runtime; use uuid::Uuid; /// Test that calls to `add_version` from different threads maintain sequential consistency. -#[test] -fn add_version_concurrency() -> anyhow::Result<()> { +/// +/// This uses `std::thread` to ensure actual parallelism, with a different, single-threaded Tokio runtime +/// in each thread. Asynchronous concurrency does not actually test consistency. +#[tokio::test] +async fn add_version_concurrency() -> anyhow::Result<()> { let tmp_dir = TempDir::new()?; let client_id = Uuid::new_v4(); { let con = SqliteStorage::new(tmp_dir.path())?; - let mut txn = con.txn(client_id)?; - txn.new_client(NIL_VERSION_ID)?; - txn.commit()?; + let mut txn = con.txn(client_id).await?; + txn.new_client(NIL_VERSION_ID).await?; + txn.commit().await?; } const N: i32 = 100; const T: i32 = 4; // Add N versions to the DB. - let add_versions = || { - let con = SqliteStorage::new(tmp_dir.path())?; + let add_versions = |tmp_dir, client_id| { + let rt = runtime::Builder::new_current_thread().build()?; + rt.block_on(async { + let con = SqliteStorage::new(tmp_dir)?; - for _ in 0..N { - let mut txn = con.txn(client_id)?; - let client = txn.get_client()?.unwrap(); - let version_id = Uuid::new_v4(); - let parent_version_id = client.latest_version_id; - std::thread::yield_now(); // Make failure more likely. - txn.add_version(version_id, parent_version_id, b"data".to_vec())?; - txn.commit()?; - } + for _ in 0..N { + let mut txn = con.txn(client_id).await?; + let client = txn.get_client().await?.unwrap(); + let version_id = Uuid::new_v4(); + let parent_version_id = client.latest_version_id; + std::thread::yield_now(); // Make failure more likely. + txn.add_version(version_id, parent_version_id, b"data".to_vec()) + .await?; + txn.commit().await?; + } - Ok::<_, anyhow::Error>(()) + Ok::<_, anyhow::Error>(()) + }) }; thread::scope(|s| { // Spawn T threads. for _ in 0..T { - s.spawn(add_versions); + let tmp_dir = tmp_dir.path(); + s.spawn(move || add_versions(tmp_dir, client_id)); } }); @@ -49,13 +58,16 @@ fn add_version_concurrency() -> anyhow::Result<()> { // same `parent_version_id`. { let con = SqliteStorage::new(tmp_dir.path())?; - let mut txn = con.txn(client_id)?; - let client = txn.get_client()?.unwrap(); + let mut txn = con.txn(client_id).await?; + let client = txn.get_client().await?.unwrap(); let mut n = 0; let mut version_id = client.latest_version_id; while version_id != NIL_VERSION_ID { - let version = txn.get_version(version_id)?.expect("version should exist"); + let version = txn + .get_version(version_id) + .await? + .expect("version should exist"); n += 1; version_id = version.parent_version_id; }