use std::error::Error;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::value::RawValue;
use super::*;
use crate::serve::protocol::{Change, CollectionWrite, DocumentWrite, Fork, Mutation};
fn fixture() -> Result<(Service, tempfile::TempDir), Box<dyn Error>> {
let directory = tempfile::tempdir()?;
let mut storage = Storage::create(
DatabaseConfig {
data_dir: directory.path().join("database"),
shard_count: 1,
distributed: None,
executor_threads: Some(1),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
fixture_long_wait_interval(),
)?;
let expected = storage
.capture(&Selection {
branch: "main".to_owned(),
version: None,
})?
.version;
let collection = storage.mutate(&Mutation {
request_id: "prepare-admission-collection".to_owned(),
branch: "main".to_owned(),
expected,
change: Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}),
})?;
let fork = storage.mutate(&Mutation {
request_id: "prepare-admission-fork".to_owned(),
branch: "main".to_owned(),
expected: collection.version,
change: Change::Fork(Fork {
name: "feature".to_owned(),
from: collection.version,
}),
})?;
storage.mutate(&Mutation {
request_id: "prepare-admission-document".to_owned(),
branch: "feature".to_owned(),
expected: fork.version,
change: Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: "one".to_owned(),
document: RawValue::from_string(r#"{"n":1}"#.to_owned())?,
}),
})?;
let service = Service::new(storage, NonZeroU32::new(64).ok_or("zero page")?)?;
Ok((service, directory))
}
#[test]
fn merge_prepare_answers_while_writer_admission_is_held() -> Result<(), Box<dyn Error>> {
let (service, directory) = fixture()?;
let guard = service.storage.lock().map_err(|error| error.to_string())?;
let before = guard
.capture(&Selection {
branch: "feature".to_owned(),
version: None,
})?
.version;
let answer = service.dispatch(Operation::MergePrepare(ReviewPage {
branch: "feature".to_owned(),
after: None,
limit: NonZeroU32::new(64).ok_or("zero page")?,
}))?;
assert_eq!(answer["done"], true);
assert_eq!(answer["lines"].as_array().ok_or("missing lines")?.len(), 1);
assert_eq!(answer["lines"][0]["state"], "source_only");
assert_eq!(
guard
.capture(&Selection {
branch: "feature".to_owned(),
version: None
})?
.version,
before
);
drop(guard);
assert!(directory.path().join("database").is_dir());
drop(service);
drop(directory);
Ok(())
}
#[test]
fn merge_dangling_answers_while_writer_admission_is_held() -> Result<(), Box<dyn Error>> {
let (service, directory) = fixture()?;
let guard = service.storage.lock().map_err(|error| error.to_string())?;
let before = guard
.capture(&Selection {
branch: "feature".to_owned(),
version: None,
})?
.version;
let mut after = None;
for _ in 0..128 {
let answer = service.dispatch(Operation::MergeDangling(ReviewPage {
branch: "feature".to_owned(),
after,
limit: NonZeroU32::MIN,
}))?;
assert!(
answer["lines"]
.as_array()
.ok_or("missing lines")?
.is_empty()
);
after = answer["next_after"].as_str().map(str::to_owned);
if after.is_none() {
assert_eq!(answer["done"], true);
assert_eq!(
guard
.capture(&Selection {
branch: "feature".to_owned(),
version: None,
})?
.version,
before
);
drop(guard);
assert!(directory.path().join("database").is_dir());
drop(service);
drop(directory);
return Ok(());
}
assert_eq!(answer["done"], false);
}
Err("dangling read did not exhaust its fixture".into())
}
const fn fixture_long_wait_interval() -> std::time::Duration {
std::time::Duration::from_secs(1)
}