GothenburgBitFactory/taskchampion-sync-server
1
0
Fork 0
mirror of https://github.com/GothenburgBitFactory/taskchampion-sync-server.git synced 2025-08-01 20:20:25 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #126 from djmitche/async

Browse source 
Make Storage methods async
This commit is contained in:
Dustin J. Mitchell 2025-07-13 09:25:43 -04:00 • committed by GitHub
commit c539e604d9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 594 additions and 422 deletions
Download patch file Download diff file Expand all files Collapse all files

33
Cargo.lock generated
View file

@ -308,6 +308,17 @@ version = "1.0.98"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e16d2d3311acee920a9eb8d33b8cbc1787ce4a264e85f964c2404b969bdcd487" 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]] [[package]]
name = "autocfg" name = "autocfg"
version = "1.5.0" version = "1.5.0"
@ -1562,7 +1573,7 @@ dependencies = [
[[package]] [[package]]
name = "taskchampion-sync-server" name = "taskchampion-sync-server"
version = "0.6.2-pre" version = "0.7.0-pre"
dependencies = [ dependencies = [
"actix-rt", "actix-rt",
"actix-web", "actix-web",
@ -1585,28 +1596,32 @@ dependencies = [
[[package]] [[package]]
name = "taskchampion-sync-server-core" name = "taskchampion-sync-server-core"
version = "0.6.2-pre" version = "0.7.0-pre"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-trait",
"chrono", "chrono",
"env_logger", "env_logger",
"log", "log",
"pretty_assertions", "pretty_assertions",
"thiserror", "thiserror",
"tokio",
"uuid", "uuid",
] ]
[[package]] [[package]]
name = "taskchampion-sync-server-storage-sqlite" name = "taskchampion-sync-server-storage-sqlite"
version = "0.6.2-pre" version = "0.7.0-pre"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-trait",
"chrono", "chrono",
"pretty_assertions", "pretty_assertions",
"rusqlite", "rusqlite",
"taskchampion-sync-server-core", "taskchampion-sync-server-core",
"tempfile", "tempfile",
"thiserror", "thiserror",
"tokio",
"uuid", "uuid",
] ]
@ -1709,9 +1724,21 @@ dependencies = [
"signal-hook-registry", "signal-hook-registry",
"slab", "slab",
"socket2", "socket2",
"tokio-macros",
"windows-sys 0.52.0", "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]] [[package]]
name = "tokio-util" name = "tokio-util"
version = "0.7.15" version = "0.7.15"

2
Cargo.toml
View file

@ -8,6 +8,7 @@ members = [
rust-version = "1.82.0" # MSRV rust-version = "1.82.0" # MSRV
[workspace.dependencies] [workspace.dependencies]
async-trait = "0.1.88"
uuid = { version = "^1.17.0", features = ["serde", "v4"] } uuid = { version = "^1.17.0", features = ["serde", "v4"] }
actix-web = "^4.11.0" actix-web = "^4.11.0"
anyhow = "1.0" anyhow = "1.0"
@ -24,3 +25,4 @@ actix-rt = "2"
tempfile = "3" tempfile = "3"
pretty_assertions = "1" pretty_assertions = "1"
temp-env = "0.3" temp-env = "0.3"
tokio = { version = "*", features = ["rt", "macros"] }

4
core/Cargo.toml
View file

@ -1,6 +1,6 @@
[package] [package]
name = "taskchampion-sync-server-core" name = "taskchampion-sync-server-core"
version = "0.6.2-pre" version = "0.7.0-pre"
authors = ["Dustin J. Mitchell <dustin@mozilla.com>"] authors = ["Dustin J. Mitchell <dustin@mozilla.com>"]
edition = "2021" edition = "2021"
description = "Core of sync protocol for TaskChampion" description = "Core of sync protocol for TaskChampion"
@ -10,6 +10,7 @@ license = "MIT"
[dependencies] [dependencies]
uuid.workspace = true uuid.workspace = true
async-trait.workspace = true
anyhow.workspace = true anyhow.workspace = true
thiserror.workspace = true thiserror.workspace = true
log.workspace = true log.workspace = true
@ -18,3 +19,4 @@ chrono.workspace = true
[dev-dependencies] [dev-dependencies]
pretty_assertions.workspace = true pretty_assertions.workspace = true
tokio.workspace = true

114
core/src/inmemory.rs
View file

@ -44,8 +44,9 @@ struct InnerTxn<'a> {
committed: bool, committed: bool,
} }
#[async_trait::async_trait]
impl Storage for InMemoryStorage { impl Storage for InMemoryStorage {
fn txn(&self, client_id: Uuid) -> anyhow::Result<Box<dyn StorageTxn + '_>> { async fn txn(&self, client_id: Uuid) -> anyhow::Result<Box<dyn StorageTxn + '_>> {
Ok(Box::new(InnerTxn { Ok(Box::new(InnerTxn {
client_id, client_id,
guard: self.0.lock().expect("poisoned lock"), guard: self.0.lock().expect("poisoned lock"),
@ -55,12 +56,13 @@ impl Storage for InMemoryStorage {
} }
} }
#[async_trait::async_trait(?Send)]
impl StorageTxn for InnerTxn<'_> { impl StorageTxn for InnerTxn<'_> {
fn get_client(&mut self) -> anyhow::Result<Option<Client>> { async fn get_client(&mut self) -> anyhow::Result<Option<Client>> {
Ok(self.guard.clients.get(&self.client_id).cloned()) 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) { if self.guard.clients.contains_key(&self.client_id) {
return Err(anyhow::anyhow!("Client {} already exists", self.client_id)); return Err(anyhow::anyhow!("Client {} already exists", self.client_id));
} }
@ -75,7 +77,7 @@ impl StorageTxn for InnerTxn<'_> {
Ok(()) Ok(())
} }
fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec<u8>) -> anyhow::Result<()> { async fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec<u8>) -> anyhow::Result<()> {
let client = self let client = self
.guard .guard
.clients .clients
@ -87,7 +89,7 @@ impl StorageTxn for InnerTxn<'_> {
Ok(()) Ok(())
} }
fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result<Option<Vec<u8>>> { async fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result<Option<Vec<u8>>> {
// sanity check // sanity check
let client = self.guard.clients.get(&self.client_id); let client = self.guard.clients.get(&self.client_id);
let client = client.ok_or_else(|| anyhow::anyhow!("no such client"))?; 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()) Ok(self.guard.snapshots.get(&self.client_id).cloned())
} }
fn get_version_by_parent( async fn get_version_by_parent(
&mut self, &mut self,
parent_version_id: Uuid, parent_version_id: Uuid,
) -> anyhow::Result<Option<Version>> { ) -> anyhow::Result<Option<Version>> {
@ -116,7 +118,7 @@ impl StorageTxn for InnerTxn<'_> {
} }
} }
fn get_version(&mut self, version_id: Uuid) -> anyhow::Result<Option<Version>> { async fn get_version(&mut self, version_id: Uuid) -> anyhow::Result<Option<Version>> {
Ok(self Ok(self
.guard .guard
.versions .versions
@ -124,7 +126,7 @@ impl StorageTxn for InnerTxn<'_> {
.cloned()) .cloned())
} }
fn add_version( async fn add_version(
&mut self, &mut self,
version_id: Uuid, version_id: Uuid,
parent_version_id: Uuid, parent_version_id: Uuid,
@ -174,7 +176,7 @@ impl StorageTxn for InnerTxn<'_> {
Ok(()) Ok(())
} }
fn commit(&mut self) -> anyhow::Result<()> { async fn commit(&mut self) -> anyhow::Result<()> {
self.committed = true; self.committed = true;
Ok(()) Ok(())
} }
@ -193,32 +195,33 @@ mod test {
use super::*; use super::*;
use chrono::Utc; use chrono::Utc;
#[test] #[tokio::test]
fn test_get_client_empty() -> anyhow::Result<()> { async fn test_get_client_empty() -> anyhow::Result<()> {
let storage = InMemoryStorage::new(); let storage = InMemoryStorage::new();
let mut txn = storage.txn(Uuid::new_v4())?; let mut txn = storage.txn(Uuid::new_v4()).await?;
let maybe_client = txn.get_client()?; let maybe_client = txn.get_client().await?;
assert!(maybe_client.is_none()); assert!(maybe_client.is_none());
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_client_storage() -> anyhow::Result<()> { async fn test_client_storage() -> anyhow::Result<()> {
let storage = InMemoryStorage::new(); let storage = InMemoryStorage::new();
let client_id = Uuid::new_v4(); 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(); 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_eq!(client.latest_version_id, latest_version_id);
assert!(client.snapshot.is_none()); assert!(client.snapshot.is_none());
let latest_version_id = Uuid::new_v4(); 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_eq!(client.latest_version_id, latest_version_id);
assert!(client.snapshot.is_none()); assert!(client.snapshot.is_none());
@ -227,38 +230,39 @@ mod test {
timestamp: Utc::now(), timestamp: Utc::now(),
versions_since: 4, 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.latest_version_id, latest_version_id);
assert_eq!(client.snapshot.unwrap(), snap); assert_eq!(client.snapshot.unwrap(), snap);
txn.commit()?; txn.commit().await?;
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_gvbp_empty() -> anyhow::Result<()> { async fn test_gvbp_empty() -> anyhow::Result<()> {
let storage = InMemoryStorage::new(); let storage = InMemoryStorage::new();
let client_id = Uuid::new_v4(); let client_id = Uuid::new_v4();
let mut txn = storage.txn(client_id)?; let mut txn = storage.txn(client_id).await?;
let maybe_version = txn.get_version_by_parent(Uuid::new_v4())?; let maybe_version = txn.get_version_by_parent(Uuid::new_v4()).await?;
assert!(maybe_version.is_none()); assert!(maybe_version.is_none());
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_add_version_and_get_version() -> anyhow::Result<()> { async fn test_add_version_and_get_version() -> anyhow::Result<()> {
let storage = InMemoryStorage::new(); let storage = InMemoryStorage::new();
let client_id = Uuid::new_v4(); 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 version_id = Uuid::new_v4();
let parent_version_id = Uuid::new_v4(); let parent_version_id = Uuid::new_v4();
let history_segment = b"abc".to_vec(); let history_segment = b"abc".to_vec();
txn.new_client(parent_version_id)?; txn.new_client(parent_version_id).await?;
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 { let expected = Version {
version_id, version_id,
@ -266,74 +270,76 @@ mod test {
history_segment, 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); 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); assert_eq!(version, expected);
txn.commit()?; txn.commit().await?;
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_add_version_exists() -> anyhow::Result<()> { async fn test_add_version_exists() -> anyhow::Result<()> {
let storage = InMemoryStorage::new(); let storage = InMemoryStorage::new();
let client_id = Uuid::new_v4(); 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 version_id = Uuid::new_v4();
let parent_version_id = Uuid::new_v4(); let parent_version_id = Uuid::new_v4();
let history_segment = b"abc".to_vec(); let history_segment = b"abc".to_vec();
txn.new_client(parent_version_id)?; txn.new_client(parent_version_id).await?;
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 assert!(txn
.add_version(version_id, parent_version_id, history_segment.clone()) .add_version(version_id, parent_version_id, history_segment.clone())
.await
.is_err()); .is_err());
txn.commit()?; txn.commit().await?;
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_snapshots() -> anyhow::Result<()> { async fn test_snapshots() -> anyhow::Result<()> {
let storage = InMemoryStorage::new(); let storage = InMemoryStorage::new();
let client_id = Uuid::new_v4(); 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())?; txn.new_client(Uuid::new_v4()).await?;
assert!(txn.get_client()?.unwrap().snapshot.is_none()); assert!(txn.get_client().await?.unwrap().snapshot.is_none());
let snap = Snapshot { let snap = Snapshot {
version_id: Uuid::new_v4(), version_id: Uuid::new_v4(),
timestamp: Utc::now(), timestamp: Utc::now(),
versions_since: 3, versions_since: 3,
}; };
txn.set_snapshot(snap.clone(), vec![9, 8, 9])?; txn.set_snapshot(snap.clone(), vec![9, 8, 9]).await?;
assert_eq!( assert_eq!(
txn.get_snapshot_data(snap.version_id)?.unwrap(), txn.get_snapshot_data(snap.version_id).await?.unwrap(),
vec![9, 8, 9] 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 { let snap2 = Snapshot {
version_id: Uuid::new_v4(), version_id: Uuid::new_v4(),
timestamp: Utc::now(), timestamp: Utc::now(),
versions_since: 10, 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!( assert_eq!(
txn.get_snapshot_data(snap2.version_id)?.unwrap(), txn.get_snapshot_data(snap2.version_id).await?.unwrap(),
vec![0, 2, 4, 6] 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 // 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(()) Ok(())
} }
} }

599
core/src/server.rs
View file

@ -106,17 +106,17 @@ impl Server {
} }
/// Implementation of the GetChildVersion protocol transaction. /// Implementation of the GetChildVersion protocol transaction.
pub fn get_child_version( pub async fn get_child_version(
&self, &self,
client_id: ClientId, client_id: ClientId,
parent_version_id: VersionId, parent_version_id: VersionId,
) -> Result<GetVersionResult, ServerError> { ) -> Result<GetVersionResult, ServerError> {
let mut txn = self.storage.txn(client_id)?; let mut txn = self.txn(client_id).await?;
let client = txn.get_client()?.ok_or(ServerError::NoSuchClient)?; let client = txn.get_client().await?.ok_or(ServerError::NoSuchClient)?;
// If a version with parentVersionId equal to the requested parentVersionId exists, it is // If a version with parentVersionId equal to the requested parentVersionId exists, it is
// returned. // 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 { return Ok(GetVersionResult::Success {
version_id: version.version_id, version_id: version.version_id,
parent_version_id: version.parent_version_id, parent_version_id: version.parent_version_id,
@ -142,7 +142,7 @@ impl Server {
} }
/// Implementation of the AddVersion protocol transaction /// Implementation of the AddVersion protocol transaction
pub fn add_version( pub async fn add_version(
&self, &self,
client_id: ClientId, client_id: ClientId,
parent_version_id: VersionId, parent_version_id: VersionId,
@ -150,8 +150,8 @@ impl Server {
) -> Result<(AddVersionResult, SnapshotUrgency), ServerError> { ) -> Result<(AddVersionResult, SnapshotUrgency), ServerError> {
log::debug!("add_version(client_id: {client_id}, parent_version_id: {parent_version_id})"); log::debug!("add_version(client_id: {client_id}, parent_version_id: {parent_version_id})");
let mut txn = self.storage.txn(client_id)?; let mut txn = self.txn(client_id).await?;
let client = txn.get_client()?.ok_or(ServerError::NoSuchClient)?; let client = txn.get_client().await?.ok_or(ServerError::NoSuchClient)?;
// check if this version is acceptable, under the protection of the transaction // check if this version is acceptable, under the protection of the transaction
if client.latest_version_id != NIL_VERSION_ID 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}"); log::debug!("add_version request accepted: new version_id: {version_id}");
// update the DB // update the DB
txn.add_version(version_id, parent_version_id, history_segment)?; txn.add_version(version_id, parent_version_id, history_segment)
txn.commit()?; .await?;
txn.commit().await?;
// calculate the urgency // calculate the urgency
let time_urgency = match client.snapshot { let time_urgency = match client.snapshot {
@ -194,7 +195,7 @@ impl Server {
} }
/// Implementation of the AddSnapshot protocol transaction /// Implementation of the AddSnapshot protocol transaction
pub fn add_snapshot( pub async fn add_snapshot(
&self, &self,
client_id: ClientId, client_id: ClientId,
version_id: VersionId, version_id: VersionId,
@ -202,8 +203,8 @@ impl Server {
) -> Result<(), ServerError> { ) -> Result<(), ServerError> {
log::debug!("add_snapshot(client_id: {client_id}, version_id: {version_id})"); log::debug!("add_snapshot(client_id: {client_id}, version_id: {version_id})");
let mut txn = self.storage.txn(client_id)?; let mut txn = self.txn(client_id).await?;
let client = txn.get_client()?.ok_or(ServerError::NoSuchClient)?; let client = txn.get_client().await?.ok_or(ServerError::NoSuchClient)?;
// NOTE: if the snapshot is rejected, this function logs about it and returns // 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. // 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 // 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; vid = parent.parent_version_id;
} else { } else {
// this version does not exist; "this should not happen" but if it does, // this version does not exist; "this should not happen" but if it does,
@ -257,21 +258,23 @@ impl Server {
versions_since: 0, versions_since: 0,
}, },
data, data,
)?; )
txn.commit()?; .await?;
txn.commit().await?;
Ok(()) Ok(())
} }
/// Implementation of the GetSnapshot protocol transaction /// Implementation of the GetSnapshot protocol transaction
pub fn get_snapshot( pub async fn get_snapshot(
&self, &self,
client_id: ClientId, client_id: ClientId,
) -> Result<Option<(Uuid, Vec<u8>)>, ServerError> { ) -> Result<Option<(Uuid, Vec<u8>)>, ServerError> {
let mut txn = self.storage.txn(client_id)?; let mut txn = self.txn(client_id).await?;
let client = txn.get_client()?.ok_or(ServerError::NoSuchClient)?; let client = txn.get_client().await?.ok_or(ServerError::NoSuchClient)?;
Ok(if let Some(snap) = client.snapshot { 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)) .map(|data| (snap.version_id, data))
} else { } else {
None None
@ -279,8 +282,8 @@ impl Server {
} }
/// Convenience method to get a transaction for the embedded storage. /// Convenience method to get a transaction for the embedded storage.
pub fn txn(&self, client_id: Uuid) -> Result<Box<dyn StorageTxn + '_>, ServerError> { pub async fn txn(&self, client_id: Uuid) -> Result<Box<dyn StorageTxn + '_>, ServerError> {
Ok(self.storage.txn(client_id)?) Ok(self.storage.txn(client_id).await?)
} }
} }
@ -288,68 +291,70 @@ impl Server {
mod test { mod test {
use super::*; use super::*;
use crate::inmemory::InMemoryStorage; use crate::inmemory::InMemoryStorage;
use crate::storage::{Snapshot, Storage, StorageTxn}; use crate::storage::{Snapshot, Storage};
use chrono::{Duration, TimeZone, Utc}; use chrono::{Duration, TimeZone, Utc};
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
fn setup<INIT, RES>(init: INIT) -> anyhow::Result<(Server, RES)> /// Set up for a test, returning storage and a client_id.
where fn setup() -> (InMemoryStorage, Uuid) {
INIT: FnOnce(&mut dyn StorageTxn, Uuid) -> anyhow::Result<RES>,
{
let _ = env_logger::builder().is_test(true).try_init(); let _ = env_logger::builder().is_test(true).try_init();
let storage = InMemoryStorage::new(); let storage = InMemoryStorage::new();
let client_id = Uuid::new_v4(); let client_id = Uuid::new_v4();
let res; (storage, client_id)
{
let mut txn = storage.txn(client_id)?;
res = init(txn.as_mut(), client_id)?;
txn.commit()?;
}
Ok((Server::new(ServerConfig::default(), storage), res))
} }
/// Utility setup function for add_version tests /// Convert storage into a Server.
fn av_setup( 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, num_versions: u32,
snapshot_version: Option<u32>, snapshot_version: Option<u32>,
snapshot_days_ago: Option<i64>, snapshot_days_ago: Option<i64>,
) -> anyhow::Result<(Server, Uuid, Vec<Uuid>)> { ) -> anyhow::Result<Vec<Uuid>> {
let (server, (client_id, versions)) = setup(|txn, client_id| { let mut txn = storage.txn(client_id).await?;
let mut versions = vec![]; let mut versions = vec![];
let mut version_id = Uuid::nil(); let mut version_id = Uuid::nil();
txn.new_client(Uuid::nil())?; txn.new_client(Uuid::nil()).await?;
debug_assert!(num_versions < u8::MAX.into()); assert!(
for vnum in 0..num_versions { num_versions < u8::MAX.into(),
let parent_version_id = version_id; "we cast the version number to u8"
version_id = Uuid::new_v4(); );
versions.push(version_id); for vnum in 0..num_versions {
txn.add_version( let parent_version_id = version_id;
version_id, version_id = Uuid::new_v4();
parent_version_id, versions.push(version_id);
// Generate some unique data for this version. txn.add_version(
vec![0, 0, vnum as u8], version_id,
)?; parent_version_id,
if Some(vnum) == snapshot_version { // Generate some unique data for this version.
txn.set_snapshot( vec![0, 0, vnum as u8],
Snapshot { )
version_id, .await?;
versions_since: 0, if Some(vnum) == snapshot_version {
timestamp: Utc::now() - Duration::days(snapshot_days_ago.unwrap_or(0)), txn.set_snapshot(
}, Snapshot {
// Generate some unique data for this snapshot. version_id,
vec![vnum as u8], 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)) txn.commit().await?;
})?; Ok(versions)
Ok((server, client_id, versions))
} }
/// Utility function to check the results of an add_version call /// Utility function to check the results of an add_version call
fn av_success_check( async fn av_success_check(
server: &Server, server: &Server,
client_id: Uuid, client_id: Uuid,
existing_versions: &[Uuid], existing_versions: &[Uuid],
@ -364,17 +369,17 @@ mod test {
} }
// verify that the storage was updated // verify that the storage was updated
let mut txn = server.txn(client_id)?; let mut txn = server.txn(client_id).await?;
let client = txn.get_client()?.unwrap(); let client = txn.get_client().await?.unwrap();
assert_eq!(client.latest_version_id, new_version_id); assert_eq!(client.latest_version_id, new_version_id);
let parent_version_id = existing_versions.last().cloned().unwrap_or_else(Uuid::nil); 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.version_id, new_version_id);
assert_eq!(version.parent_version_id, parent_version_id); assert_eq!(version.parent_version_id, parent_version_id);
assert_eq!(version.history_segment, expected_history); assert_eq!(version.history_segment, expected_history);
} else { } 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); assert_eq!(snapshot_urgency, expected_urgency);
@ -426,89 +431,108 @@ mod test {
); );
} }
#[test] #[tokio::test]
fn get_child_version_not_found_initial_nil() -> anyhow::Result<()> { async fn get_child_version_not_found_initial_nil() -> anyhow::Result<()> {
let (server, client_id) = setup(|txn, client_id| { let (storage, client_id) = setup();
txn.new_client(NIL_VERSION_ID)?; {
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 // when no latest version exists, the first version is NotFound
assert_eq!( assert_eq!(
server.get_child_version(client_id, NIL_VERSION_ID)?, server.get_child_version(client_id, NIL_VERSION_ID).await?,
GetVersionResult::NotFound GetVersionResult::NotFound
); );
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn get_child_version_not_found_initial_continuing() -> anyhow::Result<()> { async fn get_child_version_not_found_initial_continuing() -> anyhow::Result<()> {
let (server, client_id) = setup(|txn, client_id| { let (storage, client_id) = setup();
txn.new_client(NIL_VERSION_ID)?; {
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 // 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. // start to a new server even if the client already has been uploading to another service.
assert_eq!( assert_eq!(
server.get_child_version(client_id, Uuid::new_v4(),)?, server.get_child_version(client_id, Uuid::new_v4(),).await?,
GetVersionResult::NotFound GetVersionResult::NotFound
); );
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn get_child_version_not_found_up_to_date() -> anyhow::Result<()> { async fn get_child_version_not_found_up_to_date() -> anyhow::Result<()> {
let (server, (client_id, parent_version_id)) = setup(|txn, client_id| { 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 // add a parent version, but not the requested child version
let parent_version_id = Uuid::new_v4(); txn.new_client(parent_version_id).await?;
txn.new_client(parent_version_id)?; txn.add_version(parent_version_id, NIL_VERSION_ID, vec![])
txn.add_version(parent_version_id, NIL_VERSION_ID, vec![])?; .await?;
txn.commit().await?;
Ok((client_id, parent_version_id)) }
})?;
let server = into_server(storage);
assert_eq!( assert_eq!(
server.get_child_version(client_id, parent_version_id)?, server
.get_child_version(client_id, parent_version_id)
.await?,
GetVersionResult::NotFound GetVersionResult::NotFound
); );
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn get_child_version_gone_not_latest() -> anyhow::Result<()> { async fn get_child_version_gone_not_latest() -> anyhow::Result<()> {
let (server, client_id) = setup(|txn, client_id| { 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 // Add a parent version, but not the requested parent version
let parent_version_id = Uuid::new_v4(); txn.new_client(parent_version_id).await?;
txn.new_client(parent_version_id)?; txn.add_version(parent_version_id, NIL_VERSION_ID, vec![])
txn.add_version(parent_version_id, NIL_VERSION_ID, vec![])?; .await?;
txn.commit().await?;
Ok(client_id) }
})?;
let server = into_server(storage);
assert_eq!( assert_eq!(
server.get_child_version(client_id, Uuid::new_v4(),)?, server.get_child_version(client_id, Uuid::new_v4(),).await?,
GetVersionResult::Gone GetVersionResult::Gone
); );
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn get_child_version_found() -> anyhow::Result<()> { async fn get_child_version_found() -> anyhow::Result<()> {
let (server, (client_id, version_id, parent_version_id, history_segment)) = let (storage, client_id) = setup();
setup(|txn, client_id| { let version_id = Uuid::new_v4();
let version_id = Uuid::new_v4(); let parent_version_id = Uuid::new_v4();
let parent_version_id = Uuid::new_v4(); let history_segment = b"abcd".to_vec();
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)?; let server = into_server(storage);
txn.add_version(version_id, parent_version_id, history_segment.clone())?;
Ok((client_id, version_id, parent_version_id, history_segment))
})?;
assert_eq!( assert_eq!(
server.get_child_version(client_id, parent_version_id)?, server
.get_child_version(client_id, parent_version_id)
.await?,
GetVersionResult::Success { GetVersionResult::Success {
version_id, version_id,
parent_version_id, parent_version_id,
@ -518,29 +542,41 @@ mod test {
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_version_conflict() -> anyhow::Result<()> { async fn add_version_conflict() -> anyhow::Result<()> {
let (server, client_id, versions) = av_setup(3, None, None)?; 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 // try to add a child of a version other than the latest
let server = into_server(storage);
assert_eq!( 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]) AddVersionResult::ExpectedParentVersion(versions[2])
); );
// verify that the storage wasn't updated // verify that the storage wasn't updated
let mut txn = server.txn(client_id)?; let mut txn = server.txn(client_id).await?;
assert_eq!(txn.get_client()?.unwrap().latest_version_id, versions[2]); assert_eq!(
assert_eq!(txn.get_version_by_parent(versions[2])?, None); txn.get_client().await?.unwrap().latest_version_id,
versions[2]
);
assert_eq!(txn.get_version_by_parent(versions[2]).await?, None);
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_version_with_existing_history() -> anyhow::Result<()> { async fn add_version_with_existing_history() -> anyhow::Result<()> {
let (server, client_id, versions) = av_setup(1, None, None)?; 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( av_success_check(
&server, &server,
@ -550,17 +586,22 @@ mod test {
vec![3, 6, 9], vec![3, 6, 9],
// urgency=high because there are no snapshots yet // urgency=high because there are no snapshots yet
SnapshotUrgency::High, SnapshotUrgency::High,
)?; )
.await?;
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_version_with_no_history() -> anyhow::Result<()> { async fn add_version_with_no_history() -> anyhow::Result<()> {
let (server, client_id, versions) = av_setup(0, None, None)?; 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 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( av_success_check(
&server, &server,
@ -570,16 +611,21 @@ mod test {
vec![3, 6, 9], vec![3, 6, 9],
// urgency=high because there are no snapshots yet // urgency=high because there are no snapshots yet
SnapshotUrgency::High, SnapshotUrgency::High,
)?; )
.await?;
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_version_success_recent_snapshot() -> anyhow::Result<()> { async fn add_version_success_recent_snapshot() -> anyhow::Result<()> {
let (server, client_id, versions) = av_setup(1, Some(0), None)?; 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( av_success_check(
&server, &server,
@ -589,17 +635,22 @@ mod test {
vec![1, 2, 3], vec![1, 2, 3],
// no snapshot request since the previous version has a snapshot // no snapshot request since the previous version has a snapshot
SnapshotUrgency::None, SnapshotUrgency::None,
)?; )
.await?;
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_version_success_aged_snapshot() -> anyhow::Result<()> { async fn add_version_success_aged_snapshot() -> anyhow::Result<()> {
// one snapshot, but it was 50 days ago // 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( av_success_check(
&server, &server,
@ -609,18 +660,24 @@ mod test {
vec![1, 2, 3], vec![1, 2, 3],
// urgency=high due to days since the snapshot // urgency=high due to days since the snapshot
SnapshotUrgency::High, SnapshotUrgency::High,
)?; )
.await?;
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_version_success_snapshot_many_versions_ago() -> anyhow::Result<()> { async fn add_version_success_snapshot_many_versions_ago() -> anyhow::Result<()> {
// one snapshot, but it was 50 versions ago // 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; 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( av_success_check(
&server, &server,
@ -630,136 +687,165 @@ mod test {
vec![1, 2, 3], vec![1, 2, 3],
// urgency=high due to number of versions since the snapshot // urgency=high due to number of versions since the snapshot
SnapshotUrgency::High, SnapshotUrgency::High,
)?; )
.await?;
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_snapshot_success_latest() -> anyhow::Result<()> { async fn add_snapshot_success_latest() -> anyhow::Result<()> {
let (server, (client_id, version_id)) = setup(|txn, client_id| { let (storage, client_id) = setup();
let version_id = Uuid::new_v4(); let version_id = Uuid::new_v4();
{
let mut txn = storage.txn(client_id).await?;
// set up a task DB with one version in it // set up a task DB with one version in it
txn.new_client(version_id)?; txn.new_client(version_id).await?;
txn.add_version(version_id, NIL_VERSION_ID, vec![])?; txn.add_version(version_id, NIL_VERSION_ID, vec![]).await?;
// add a snapshot for that version txn.commit().await?;
Ok((client_id, version_id)) }
})?;
server.add_snapshot(client_id, version_id, vec![1, 2, 3])?; let server = into_server(storage);
server
.add_snapshot(client_id, version_id, vec![1, 2, 3])
.await?;
// verify the snapshot // verify the snapshot
let mut txn = server.txn(client_id)?; let mut txn = server.txn(client_id).await?;
let client = txn.get_client()?.unwrap(); let client = txn.get_client().await?.unwrap();
let snapshot = client.snapshot.unwrap(); let snapshot = client.snapshot.unwrap();
assert_eq!(snapshot.version_id, version_id); assert_eq!(snapshot.version_id, version_id);
assert_eq!(snapshot.versions_since, 0); assert_eq!(snapshot.versions_since, 0);
assert_eq!( assert_eq!(
txn.get_snapshot_data(version_id).unwrap(), txn.get_snapshot_data(version_id).await.unwrap(),
Some(vec![1, 2, 3]) Some(vec![1, 2, 3])
); );
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_snapshot_success_older() -> anyhow::Result<()> { async fn add_snapshot_success_older() -> anyhow::Result<()> {
let (server, (client_id, version_id_1)) = setup(|txn, client_id| { let (storage, client_id) = setup();
let version_id_1 = Uuid::new_v4(); let version_id_1 = Uuid::new_v4();
let version_id_2 = 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 // set up a task DB with two versions in it
txn.new_client(version_id_2)?; txn.new_client(version_id_2).await?;
txn.add_version(version_id_1, NIL_VERSION_ID, vec![])?; txn.add_version(version_id_1, NIL_VERSION_ID, vec![])
txn.add_version(version_id_2, version_id_1, 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 // 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 // verify the snapshot
let mut txn = server.txn(client_id)?; let mut txn = server.txn(client_id).await?;
let client = txn.get_client()?.unwrap(); let client = txn.get_client().await?.unwrap();
let snapshot = client.snapshot.unwrap(); let snapshot = client.snapshot.unwrap();
assert_eq!(snapshot.version_id, version_id_1); assert_eq!(snapshot.version_id, version_id_1);
assert_eq!(snapshot.versions_since, 0); assert_eq!(snapshot.versions_since, 0);
assert_eq!( assert_eq!(
txn.get_snapshot_data(version_id_1).unwrap(), txn.get_snapshot_data(version_id_1).await.unwrap(),
Some(vec![1, 2, 3]) Some(vec![1, 2, 3])
); );
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_snapshot_fails_no_such() -> anyhow::Result<()> { async fn add_snapshot_fails_no_such() -> anyhow::Result<()> {
let (server, client_id) = setup(|txn, client_id| { let (storage, client_id) = setup();
let version_id_1 = Uuid::new_v4(); let version_id_1 = Uuid::new_v4();
let version_id_2 = 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 // set up a task DB with two versions in it
txn.new_client(version_id_2)?; txn.new_client(version_id_2).await?;
txn.add_version(version_id_1, NIL_VERSION_ID, vec![])?; txn.add_version(version_id_1, NIL_VERSION_ID, vec![])
txn.add_version(version_id_2, version_id_1, vec![])?; .await?;
txn.add_version(version_id_2, version_id_1, vec![]).await?;
// add a snapshot for unknown version txn.commit().await?;
Ok(client_id) }
})?;
// add a snapshot for unknown version
let server = into_server(storage);
let version_id_unk = Uuid::new_v4(); 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 // verify the snapshot does not exist
let mut txn = server.txn(client_id)?; let mut txn = server.txn(client_id).await?;
let client = txn.get_client()?.unwrap(); let client = txn.get_client().await?.unwrap();
assert!(client.snapshot.is_none()); assert!(client.snapshot.is_none());
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_snapshot_fails_too_old() -> anyhow::Result<()> { async fn add_snapshot_fails_too_old() -> anyhow::Result<()> {
let (server, (client_id, version_ids)) = setup(|txn, client_id| { let (storage, client_id) = setup();
let mut version_id = Uuid::new_v4(); let mut version_id = Uuid::new_v4();
let mut parent_version_id = Uuid::nil(); let mut parent_version_id = Uuid::nil();
let mut version_ids = vec![]; 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) // 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 { 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); version_ids.push(version_id);
parent_version_id = version_id; parent_version_id = version_id;
version_id = Uuid::new_v4(); version_id = Uuid::new_v4();
} }
// add a snapshot for the earliest of those txn.commit().await?;
Ok((client_id, version_ids)) }
})?;
server.add_snapshot(client_id, version_ids[0], vec![1, 2, 3])?; // 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 // verify the snapshot does not exist
let mut txn = server.txn(client_id)?; let mut txn = server.txn(client_id).await?;
let client = txn.get_client()?.unwrap(); let client = txn.get_client().await?.unwrap();
assert!(client.snapshot.is_none()); assert!(client.snapshot.is_none());
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_snapshot_fails_newer_exists() -> anyhow::Result<()> { async fn add_snapshot_fails_newer_exists() -> anyhow::Result<()> {
let (server, (client_id, version_ids)) = setup(|txn, client_id| { let (storage, client_id) = setup();
let mut version_id = Uuid::new_v4(); let mut version_id = Uuid::new_v4();
let mut parent_version_id = Uuid::nil(); let mut parent_version_id = Uuid::nil();
let mut version_ids = vec![]; 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 // set up a task DB with 5 versions in it (oldest to newest) and a snapshot of the
// middle one // middle one
txn.new_client(Uuid::nil())?; txn.new_client(Uuid::nil()).await?;
for _ in 0..5 { 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); version_ids.push(version_id);
parent_version_id = version_id; parent_version_id = version_id;
version_id = Uuid::new_v4(); version_id = Uuid::new_v4();
@ -771,55 +857,64 @@ mod test {
timestamp: Utc.with_ymd_and_hms(2001, 9, 9, 1, 46, 40).unwrap(), timestamp: Utc.with_ymd_and_hms(2001, 9, 9, 1, 46, 40).unwrap(),
}, },
vec![1, 2, 3], vec![1, 2, 3],
)?; )
.await?;
// add a snapshot for the earliest of those txn.commit().await?;
Ok((client_id, version_ids)) }
})?;
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 // verify the snapshot was not replaced
let mut txn = server.txn(client_id)?; let mut txn = server.txn(client_id).await?;
let client = txn.get_client()?.unwrap(); let client = txn.get_client().await?.unwrap();
let snapshot = client.snapshot.unwrap(); let snapshot = client.snapshot.unwrap();
assert_eq!(snapshot.version_id, version_ids[2]); assert_eq!(snapshot.version_id, version_ids[2]);
assert_eq!(snapshot.versions_since, 2); assert_eq!(snapshot.versions_since, 2);
assert_eq!( assert_eq!(
txn.get_snapshot_data(version_ids[2]).unwrap(), txn.get_snapshot_data(version_ids[2]).await.unwrap(),
Some(vec![1, 2, 3]) Some(vec![1, 2, 3])
); );
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn add_snapshot_fails_nil_version() -> anyhow::Result<()> { async fn add_snapshot_fails_nil_version() -> anyhow::Result<()> {
let (server, client_id) = setup(|txn, client_id| { let (storage, client_id) = setup();
{
let mut txn = storage.txn(client_id).await?;
// just set up the client // 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 let server = into_server(storage);
Ok(client_id) server
})?; .add_snapshot(client_id, NIL_VERSION_ID, vec![9, 9, 9])
.await?;
server.add_snapshot(client_id, NIL_VERSION_ID, vec![9, 9, 9])?;
// verify the snapshot does not exist // verify the snapshot does not exist
let mut txn = server.txn(client_id)?; let mut txn = server.txn(client_id).await?;
let client = txn.get_client()?.unwrap(); let client = txn.get_client().await?.unwrap();
assert!(client.snapshot.is_none()); assert!(client.snapshot.is_none());
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn get_snapshot_found() -> anyhow::Result<()> { async fn get_snapshot_found() -> anyhow::Result<()> {
let (server, (client_id, data, snapshot_version_id)) = setup(|txn, client_id| { let (storage, client_id) = setup();
let data = vec![1, 2, 3]; let data = vec![1, 2, 3];
let snapshot_version_id = Uuid::new_v4(); 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( txn.set_snapshot(
Snapshot { Snapshot {
version_id: snapshot_version_id, version_id: snapshot_version_id,
@ -827,25 +922,31 @@ mod test {
timestamp: Utc.with_ymd_and_hms(2001, 9, 9, 1, 46, 40).unwrap(), timestamp: Utc.with_ymd_and_hms(2001, 9, 9, 1, 46, 40).unwrap(),
}, },
data.clone(), data.clone(),
)?; )
Ok((client_id, data, snapshot_version_id)) .await?;
})?; txn.commit().await?;
}
let server = into_server(storage);
assert_eq!( assert_eq!(
server.get_snapshot(client_id)?, server.get_snapshot(client_id).await?,
Some((snapshot_version_id, data)) Some((snapshot_version_id, data))
); );
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn get_snapshot_not_found() -> anyhow::Result<()> { async fn get_snapshot_not_found() -> anyhow::Result<()> {
let (server, client_id) = setup(|txn, client_id| { let (storage, client_id) = setup();
txn.new_client(NIL_VERSION_ID)?; {
Ok(client_id) 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(()) Ok(())
} }

24
core/src/storage.rs
View file

@ -44,32 +44,35 @@ pub struct Version {
/// ///
/// Changes in a transaction that is dropped without calling `commit` must not appear in any other /// Changes in a transaction that is dropped without calling `commit` must not appear in any other
/// transaction. /// transaction.
#[async_trait::async_trait(?Send)]
pub trait StorageTxn { pub trait StorageTxn {
/// Get information about the client for this transaction /// Get information about the client for this transaction
fn get_client(&mut self) -> anyhow::Result<Option<Client>>; async fn get_client(&mut self) -> anyhow::Result<Option<Client>>;
/// Create the client for this transaction, with the given latest_version_id. The client must /// Create the client for this transaction, with the given latest_version_id. The client must
/// not already exist. /// 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. /// Set the client's most recent snapshot.
fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec<u8>) -> anyhow::Result<()>; async fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec<u8>) -> anyhow::Result<()>;
/// Get the data for the most recent snapshot. The version_id /// Get the data for the most recent snapshot. The version_id
/// is used to verify that the snapshot is for the correct version. /// is used to verify that the snapshot is for the correct version.
fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result<Option<Vec<u8>>>; async fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result<Option<Vec<u8>>>;
/// Get a version, indexed by parent version id /// Get a version, indexed by parent version id
fn get_version_by_parent(&mut self, parent_version_id: Uuid) async fn get_version_by_parent(
-> anyhow::Result<Option<Version>>; &mut self,
parent_version_id: Uuid,
) -> anyhow::Result<Option<Version>>;
/// Get a version, indexed by its own version id /// Get a version, indexed by its own version id
fn get_version(&mut self, version_id: Uuid) -> anyhow::Result<Option<Version>>; async fn get_version(&mut self, version_id: Uuid) -> anyhow::Result<Option<Version>>;
/// Add a version (that must not already exist), and /// Add a version (that must not already exist), and
/// - update latest_version_id /// - update latest_version_id
/// - increment snapshot.versions_since /// - increment snapshot.versions_since
fn add_version( async fn add_version(
&mut self, &mut self,
version_id: Uuid, version_id: Uuid,
parent_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 /// 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. /// 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 /// A trait for objects able to act as storage. Most of the interesting behavior is in the
/// [`crate::storage::StorageTxn`] trait. /// [`crate::storage::StorageTxn`] trait.
#[async_trait::async_trait]
pub trait Storage: Send + Sync { pub trait Storage: Send + Sync {
/// Begin a transaction for the given client ID. /// Begin a transaction for the given client ID.
fn txn(&self, client_id: Uuid) -> anyhow::Result<Box<dyn StorageTxn + '_>>; async fn txn(&self, client_id: Uuid) -> anyhow::Result<Box<dyn StorageTxn + '_>>;
} }

2
server/Cargo.toml
View file

@ -1,6 +1,6 @@
[package] [package]
name = "taskchampion-sync-server" name = "taskchampion-sync-server"
version = "0.6.2-pre" version = "0.7.0-pre"
authors = ["Dustin J. Mitchell <dustin@mozilla.com>"] authors = ["Dustin J. Mitchell <dustin@mozilla.com>"]
edition = "2021" edition = "2021"
publish = false publish = false

15
server/src/api/add_snapshot.rs
View file

@ -49,6 +49,7 @@ pub(crate) async fn service(
server_state server_state
.server .server
.add_snapshot(client_id, version_id, body.to_vec()) .add_snapshot(client_id, version_id, body.to_vec())
.await
.map_err(server_error_to_actix)?; .map_err(server_error_to_actix)?;
Ok(HttpResponse::Ok().body("")) Ok(HttpResponse::Ok().body(""))
} }
@ -70,10 +71,10 @@ mod test {
// set up the storage contents.. // set up the storage contents..
{ {
let mut txn = storage.txn(client_id).unwrap(); let mut txn = storage.txn(client_id).await.unwrap();
txn.new_client(version_id).unwrap(); txn.new_client(version_id).await.unwrap();
txn.add_version(version_id, NIL_VERSION_ID, vec![])?; txn.add_version(version_id, NIL_VERSION_ID, vec![]).await?;
txn.commit()?; txn.commit().await?;
} }
let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage);
@ -114,9 +115,9 @@ mod test {
// set up the storage contents.. // set up the storage contents..
{ {
let mut txn = storage.txn(client_id).unwrap(); let mut txn = storage.txn(client_id).await.unwrap();
txn.new_client(NIL_VERSION_ID).unwrap(); txn.new_client(NIL_VERSION_ID).await.unwrap();
txn.commit().unwrap(); txn.commit().await.unwrap();
} }
let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage);

24
server/src/api/add_version.rs
View file

@ -60,6 +60,7 @@ pub(crate) async fn service(
return match server_state return match server_state
.server .server
.add_version(client_id, parent_version_id, body.to_vec()) .add_version(client_id, parent_version_id, body.to_vec())
.await
{ {
Ok((AddVersionResult::Ok(version_id), snap_urgency)) => { Ok((AddVersionResult::Ok(version_id), snap_urgency)) => {
let mut rb = HttpResponse::Ok(); let mut rb = HttpResponse::Ok();
@ -85,9 +86,12 @@ pub(crate) async fn service(
let mut txn = server_state let mut txn = server_state
.server .server
.txn(client_id) .txn(client_id)
.await
.map_err(server_error_to_actix)?; .map_err(server_error_to_actix)?;
txn.new_client(NIL_VERSION_ID).map_err(failure_to_ise)?; txn.new_client(NIL_VERSION_ID)
txn.commit().map_err(failure_to_ise)?; .await
.map_err(failure_to_ise)?;
txn.commit().await.map_err(failure_to_ise)?;
continue; continue;
} }
Err(e) => Err(server_error_to_actix(e)), Err(e) => Err(server_error_to_actix(e)),
@ -113,9 +117,9 @@ mod test {
// set up the storage contents.. // set up the storage contents..
{ {
let mut txn = storage.txn(client_id).unwrap(); let mut txn = storage.txn(client_id).await.unwrap();
txn.new_client(Uuid::nil()).unwrap(); txn.new_client(Uuid::nil()).await.unwrap();
txn.commit().unwrap(); txn.commit().await.unwrap();
} }
let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage);
@ -187,8 +191,8 @@ mod test {
// Check that the client really was created // Check that the client really was created
{ {
let mut txn = server.server_state.server.txn(client_id).unwrap(); let mut txn = server.server_state.server.txn(client_id).await.unwrap();
let client = txn.get_client().unwrap().unwrap(); let client = txn.get_client().await.unwrap().unwrap();
assert_eq!(client.latest_version_id, new_version_id); assert_eq!(client.latest_version_id, new_version_id);
assert_eq!(client.snapshot, None); assert_eq!(client.snapshot, None);
} }
@ -233,9 +237,9 @@ mod test {
// set up the storage contents.. // set up the storage contents..
{ {
let mut txn = storage.txn(client_id).unwrap(); let mut txn = storage.txn(client_id).await.unwrap();
txn.new_client(version_id).unwrap(); txn.new_client(version_id).await.unwrap();
txn.commit().unwrap(); txn.commit().await.unwrap();
} }
let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage);

15
server/src/api/get_child_version.rs
View file

@ -26,6 +26,7 @@ pub(crate) async fn service(
match server_state match server_state
.server .server
.get_child_version(client_id, parent_version_id) .get_child_version(client_id, parent_version_id)
.await
{ {
Ok(GetVersionResult::Success { Ok(GetVersionResult::Success {
version_id, version_id,
@ -64,11 +65,12 @@ mod test {
// set up the storage contents.. // set up the storage contents..
{ {
let mut txn = storage.txn(client_id).unwrap(); let mut txn = storage.txn(client_id).await.unwrap();
txn.new_client(Uuid::new_v4()).unwrap(); txn.new_client(Uuid::new_v4()).await.unwrap();
txn.add_version(version_id, parent_version_id, b"abcd".to_vec()) txn.add_version(version_id, parent_version_id, b"abcd".to_vec())
.await
.unwrap(); .unwrap();
txn.commit().unwrap(); txn.commit().await.unwrap();
} }
let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage);
@ -128,11 +130,12 @@ mod test {
// create the client and a single version. // create the client and a single version.
{ {
let mut txn = storage.txn(client_id).unwrap(); let mut txn = storage.txn(client_id).await.unwrap();
txn.new_client(Uuid::new_v4()).unwrap(); txn.new_client(Uuid::new_v4()).await.unwrap();
txn.add_version(test_version_id, NIL_VERSION_ID, b"vers".to_vec()) txn.add_version(test_version_id, NIL_VERSION_ID, b"vers".to_vec())
.await
.unwrap(); .unwrap();
txn.commit().unwrap(); txn.commit().await.unwrap();
} }
let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage);
let app = App::new().configure(|sc| server.config(sc)); let app = App::new().configure(|sc| server.config(sc));

14
server/src/api/get_snapshot.rs
View file

@ -20,6 +20,7 @@ pub(crate) async fn service(
if let Some((version_id, data)) = server_state if let Some((version_id, data)) = server_state
.server .server
.get_snapshot(client_id) .get_snapshot(client_id)
.await
.map_err(server_error_to_actix)? .map_err(server_error_to_actix)?
{ {
Ok(HttpResponse::Ok() Ok(HttpResponse::Ok()
@ -48,9 +49,9 @@ mod test {
// set up the storage contents.. // set up the storage contents..
{ {
let mut txn = storage.txn(client_id).unwrap(); let mut txn = storage.txn(client_id).await.unwrap();
txn.new_client(Uuid::new_v4()).unwrap(); txn.new_client(Uuid::new_v4()).await.unwrap();
txn.commit().unwrap(); txn.commit().await.unwrap();
} }
let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage);
@ -75,8 +76,8 @@ mod test {
// set up the storage contents.. // set up the storage contents..
{ {
let mut txn = storage.txn(client_id).unwrap(); let mut txn = storage.txn(client_id).await.unwrap();
txn.new_client(Uuid::new_v4()).unwrap(); txn.new_client(Uuid::new_v4()).await.unwrap();
txn.set_snapshot( txn.set_snapshot(
Snapshot { Snapshot {
version_id, version_id,
@ -85,8 +86,9 @@ mod test {
}, },
snapshot_data.clone(), snapshot_data.clone(),
) )
.await
.unwrap(); .unwrap();
txn.commit().unwrap(); txn.commit().await.unwrap();
} }
let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage); let server = WebServer::new(ServerConfig::default(), WebConfig::default(), storage);

6
sqlite/Cargo.toml
View file

@ -1,6 +1,6 @@
[package] [package]
name = "taskchampion-sync-server-storage-sqlite" name = "taskchampion-sync-server-storage-sqlite"
version = "0.6.2-pre" version = "0.7.0-pre"
authors = ["Dustin J. Mitchell <dustin@mozilla.com>"] authors = ["Dustin J. Mitchell <dustin@mozilla.com>"]
edition = "2021" edition = "2021"
description = "SQLite backend for TaskChampion-sync-server" description = "SQLite backend for TaskChampion-sync-server"
@ -9,7 +9,8 @@ repository = "https://github.com/GothenburgBitFactory/taskchampion-sync-server"
license = "MIT" license = "MIT"
[dependencies] [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 uuid.workspace = true
anyhow.workspace = true anyhow.workspace = true
thiserror.workspace = true thiserror.workspace = true
@ -19,3 +20,4 @@ chrono.workspace = true
[dev-dependencies] [dev-dependencies]
tempfile.workspace = true tempfile.workspace = true
pretty_assertions.workspace = true pretty_assertions.workspace = true
tokio.workspace = true

110
sqlite/src/lib.rs
View file

@ -77,8 +77,9 @@ impl SqliteStorage {
} }
} }
#[async_trait::async_trait]
impl Storage for SqliteStorage { impl Storage for SqliteStorage {
fn txn(&self, client_id: Uuid) -> anyhow::Result<Box<dyn StorageTxn + '_>> { async fn txn(&self, client_id: Uuid) -> anyhow::Result<Box<dyn StorageTxn + '_>> {
let con = self.new_connection()?; let con = self.new_connection()?;
// Begin the transaction on this new connection. An IMMEDIATE connection is in // Begin the transaction on this new connection. An IMMEDIATE connection is in
// write (exclusive) mode from the start. // write (exclusive) mode from the start.
@ -126,8 +127,9 @@ impl Txn {
} }
} }
#[async_trait::async_trait(?Send)]
impl StorageTxn for Txn { impl StorageTxn for Txn {
fn get_client(&mut self) -> anyhow::Result<Option<Client>> { async fn get_client(&mut self) -> anyhow::Result<Option<Client>> {
let result: Option<Client> = self let result: Option<Client> = self
.con .con
.query_row( .query_row(
@ -171,7 +173,7 @@ impl StorageTxn for Txn {
Ok(result) 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 self.con
.execute( .execute(
"INSERT OR REPLACE INTO clients (client_id, latest_version_id) VALUES (?, ?)", "INSERT OR REPLACE INTO clients (client_id, latest_version_id) VALUES (?, ?)",
@ -181,7 +183,7 @@ impl StorageTxn for Txn {
Ok(()) Ok(())
} }
fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec<u8>) -> anyhow::Result<()> { async fn set_snapshot(&mut self, snapshot: Snapshot, data: Vec<u8>) -> anyhow::Result<()> {
self.con self.con
.execute( .execute(
"UPDATE clients "UPDATE clients
@ -203,7 +205,7 @@ impl StorageTxn for Txn {
Ok(()) Ok(())
} }
fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result<Option<Vec<u8>>> { async fn get_snapshot_data(&mut self, version_id: Uuid) -> anyhow::Result<Option<Vec<u8>>> {
let r = self let r = self
.con .con
.query_row( .query_row(
@ -227,7 +229,7 @@ impl StorageTxn for Txn {
.transpose() .transpose()
} }
fn get_version_by_parent( async fn get_version_by_parent(
&mut self, &mut self,
parent_version_id: Uuid, parent_version_id: Uuid,
@ -238,14 +240,14 @@ impl StorageTxn for Txn {
parent_version_id) parent_version_id)
} }
fn get_version(&mut self, version_id: Uuid) -> anyhow::Result<Option<Version>> { async fn get_version(&mut self, version_id: Uuid) -> anyhow::Result<Option<Version>> {
self.get_version_impl( self.get_version_impl(
"SELECT version_id, parent_version_id, history_segment FROM versions WHERE version_id = ? AND client_id = ?", "SELECT version_id, parent_version_id, history_segment FROM versions WHERE version_id = ? AND client_id = ?",
self.client_id, self.client_id,
version_id) version_id)
} }
fn add_version( async fn add_version(
&mut self, &mut self,
version_id: Uuid, version_id: Uuid,
@ -276,7 +278,7 @@ impl StorageTxn for Txn {
Ok(()) Ok(())
} }
fn commit(&mut self) -> anyhow::Result<()> { async fn commit(&mut self) -> anyhow::Result<()> {
self.con.execute("COMMIT", [])?; self.con.execute("COMMIT", [])?;
Ok(()) Ok(())
} }
@ -289,47 +291,48 @@ mod test {
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
use tempfile::TempDir; use tempfile::TempDir;
#[test] #[tokio::test]
fn test_emtpy_dir() -> anyhow::Result<()> { async fn test_emtpy_dir() -> anyhow::Result<()> {
let tmp_dir = TempDir::new()?; let tmp_dir = TempDir::new()?;
let non_existant = tmp_dir.path().join("subdir"); let non_existant = tmp_dir.path().join("subdir");
let storage = SqliteStorage::new(non_existant)?; let storage = SqliteStorage::new(non_existant)?;
let client_id = Uuid::new_v4(); let client_id = Uuid::new_v4();
let mut txn = storage.txn(client_id)?; let mut txn = storage.txn(client_id).await?;
let maybe_client = txn.get_client()?; let maybe_client = txn.get_client().await?;
assert!(maybe_client.is_none()); assert!(maybe_client.is_none());
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_get_client_empty() -> anyhow::Result<()> { async fn test_get_client_empty() -> anyhow::Result<()> {
let tmp_dir = TempDir::new()?; let tmp_dir = TempDir::new()?;
let storage = SqliteStorage::new(tmp_dir.path())?; let storage = SqliteStorage::new(tmp_dir.path())?;
let client_id = Uuid::new_v4(); let client_id = Uuid::new_v4();
let mut txn = storage.txn(client_id)?; let mut txn = storage.txn(client_id).await?;
let maybe_client = txn.get_client()?; let maybe_client = txn.get_client().await?;
assert!(maybe_client.is_none()); assert!(maybe_client.is_none());
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_client_storage() -> anyhow::Result<()> { async fn test_client_storage() -> anyhow::Result<()> {
let tmp_dir = TempDir::new()?; let tmp_dir = TempDir::new()?;
let storage = SqliteStorage::new(tmp_dir.path())?; let storage = SqliteStorage::new(tmp_dir.path())?;
let client_id = Uuid::new_v4(); 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(); 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_eq!(client.latest_version_id, latest_version_id);
assert!(client.snapshot.is_none()); assert!(client.snapshot.is_none());
let latest_version_id = Uuid::new_v4(); 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_eq!(client.latest_version_id, latest_version_id);
assert!(client.snapshot.is_none()); assert!(client.snapshot.is_none());
@ -338,37 +341,38 @@ mod test {
timestamp: "2014-11-28T12:00:09Z".parse::<DateTime<Utc>>().unwrap(), timestamp: "2014-11-28T12:00:09Z".parse::<DateTime<Utc>>().unwrap(),
versions_since: 4, 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.latest_version_id, latest_version_id);
assert_eq!(client.snapshot.unwrap(), snap); assert_eq!(client.snapshot.unwrap(), snap);
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_gvbp_empty() -> anyhow::Result<()> { async fn test_gvbp_empty() -> anyhow::Result<()> {
let tmp_dir = TempDir::new()?; let tmp_dir = TempDir::new()?;
let storage = SqliteStorage::new(tmp_dir.path())?; let storage = SqliteStorage::new(tmp_dir.path())?;
let client_id = Uuid::new_v4(); let client_id = Uuid::new_v4();
let mut txn = storage.txn(client_id)?; let mut txn = storage.txn(client_id).await?;
let maybe_version = txn.get_version_by_parent(Uuid::new_v4())?; let maybe_version = txn.get_version_by_parent(Uuid::new_v4()).await?;
assert!(maybe_version.is_none()); assert!(maybe_version.is_none());
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_add_version_and_get_version() -> anyhow::Result<()> { async fn test_add_version_and_get_version() -> anyhow::Result<()> {
let tmp_dir = TempDir::new()?; let tmp_dir = TempDir::new()?;
let storage = SqliteStorage::new(tmp_dir.path())?; let storage = SqliteStorage::new(tmp_dir.path())?;
let client_id = Uuid::new_v4(); 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 version_id = Uuid::new_v4();
let parent_version_id = Uuid::new_v4(); let parent_version_id = Uuid::new_v4();
let history_segment = b"abc".to_vec(); 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 { let expected = Version {
version_id, version_id,
@ -376,70 +380,72 @@ mod test {
history_segment, 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); 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); assert_eq!(version, expected);
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_add_version_exists() -> anyhow::Result<()> { async fn test_add_version_exists() -> anyhow::Result<()> {
let tmp_dir = TempDir::new()?; let tmp_dir = TempDir::new()?;
let storage = SqliteStorage::new(tmp_dir.path())?; let storage = SqliteStorage::new(tmp_dir.path())?;
let client_id = Uuid::new_v4(); 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 version_id = Uuid::new_v4();
let parent_version_id = Uuid::new_v4(); let parent_version_id = Uuid::new_v4();
let history_segment = b"abc".to_vec(); 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 assert!(txn
.add_version(version_id, parent_version_id, history_segment.clone()) .add_version(version_id, parent_version_id, history_segment.clone())
.await
.is_err()); .is_err());
Ok(()) Ok(())
} }
#[test] #[tokio::test]
fn test_snapshots() -> anyhow::Result<()> { async fn test_snapshots() -> anyhow::Result<()> {
let tmp_dir = TempDir::new()?; let tmp_dir = TempDir::new()?;
let storage = SqliteStorage::new(tmp_dir.path())?; let storage = SqliteStorage::new(tmp_dir.path())?;
let client_id = Uuid::new_v4(); 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())?; txn.new_client(Uuid::new_v4()).await?;
assert!(txn.get_client()?.unwrap().snapshot.is_none()); assert!(txn.get_client().await?.unwrap().snapshot.is_none());
let snap = Snapshot { let snap = Snapshot {
version_id: Uuid::new_v4(), version_id: Uuid::new_v4(),
timestamp: "2013-10-08T12:00:09Z".parse::<DateTime<Utc>>().unwrap(), timestamp: "2013-10-08T12:00:09Z".parse::<DateTime<Utc>>().unwrap(),
versions_since: 3, versions_since: 3,
}; };
txn.set_snapshot(snap.clone(), vec![9, 8, 9])?; txn.set_snapshot(snap.clone(), vec![9, 8, 9]).await?;
assert_eq!( assert_eq!(
txn.get_snapshot_data(snap.version_id)?.unwrap(), txn.get_snapshot_data(snap.version_id).await?.unwrap(),
vec![9, 8, 9] 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 { let snap2 = Snapshot {
version_id: Uuid::new_v4(), version_id: Uuid::new_v4(),
timestamp: "2014-11-28T12:00:09Z".parse::<DateTime<Utc>>().unwrap(), timestamp: "2014-11-28T12:00:09Z".parse::<DateTime<Utc>>().unwrap(),
versions_since: 10, 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!( assert_eq!(
txn.get_snapshot_data(snap2.version_id)?.unwrap(), txn.get_snapshot_data(snap2.version_id).await?.unwrap(),
vec![0, 2, 4, 6] 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 // 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(()) Ok(())
} }

54
sqlite/tests/concurrency.rs
View file

@ -2,45 +2,54 @@ use std::thread;
use taskchampion_sync_server_core::{Storage, NIL_VERSION_ID}; use taskchampion_sync_server_core::{Storage, NIL_VERSION_ID};
use taskchampion_sync_server_storage_sqlite::SqliteStorage; use taskchampion_sync_server_storage_sqlite::SqliteStorage;
use tempfile::TempDir; use tempfile::TempDir;
use tokio::runtime;
use uuid::Uuid; use uuid::Uuid;
/// Test that calls to `add_version` from different threads maintain sequential consistency. /// 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 tmp_dir = TempDir::new()?;
let client_id = Uuid::new_v4(); let client_id = Uuid::new_v4();
{ {
let con = SqliteStorage::new(tmp_dir.path())?; let con = SqliteStorage::new(tmp_dir.path())?;
let mut txn = con.txn(client_id)?; let mut txn = con.txn(client_id).await?;
txn.new_client(NIL_VERSION_ID)?; txn.new_client(NIL_VERSION_ID).await?;
txn.commit()?; txn.commit().await?;
} }
const N: i32 = 100; const N: i32 = 100;
const T: i32 = 4; const T: i32 = 4;
// Add N versions to the DB. // Add N versions to the DB.
let add_versions = || { let add_versions = |tmp_dir, client_id| {
let con = SqliteStorage::new(tmp_dir.path())?; let rt = runtime::Builder::new_current_thread().build()?;
rt.block_on(async {
let con = SqliteStorage::new(tmp_dir)?;
for _ in 0..N { for _ in 0..N {
let mut txn = con.txn(client_id)?; let mut txn = con.txn(client_id).await?;
let client = txn.get_client()?.unwrap(); let client = txn.get_client().await?.unwrap();
let version_id = Uuid::new_v4(); let version_id = Uuid::new_v4();
let parent_version_id = client.latest_version_id; let parent_version_id = client.latest_version_id;
std::thread::yield_now(); // Make failure more likely. std::thread::yield_now(); // Make failure more likely.
txn.add_version(version_id, parent_version_id, b"data".to_vec())?; txn.add_version(version_id, parent_version_id, b"data".to_vec())
txn.commit()?; .await?;
} txn.commit().await?;
}
Ok::<_, anyhow::Error>(()) Ok::<_, anyhow::Error>(())
})
}; };
thread::scope(|s| { thread::scope(|s| {
// Spawn T threads. // Spawn T threads.
for _ in 0..T { 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`. // same `parent_version_id`.
{ {
let con = SqliteStorage::new(tmp_dir.path())?; let con = SqliteStorage::new(tmp_dir.path())?;
let mut txn = con.txn(client_id)?; let mut txn = con.txn(client_id).await?;
let client = txn.get_client()?.unwrap(); let client = txn.get_client().await?.unwrap();
let mut n = 0; let mut n = 0;
let mut version_id = client.latest_version_id; let mut version_id = client.latest_version_id;
while version_id != NIL_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; n += 1;
version_id = version.parent_version_id; version_id = version.parent_version_id;
} }