use std::num::{NonZeroU32, NonZeroUsize};
use std::time::Duration;
use haematite::branch::KeptLimit;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use haematite_query::page::Origin;
use serde_json::value::RawValue;
use crate::serve::access::{self, TestSource};
use crate::serve::protocol::{Change, CollectionWrite, Mutation, Selection, Version};
use crate::serve::query::parser::{Stage, window};
use crate::serve::query::run::{Context, window as execution};
use crate::serve::query::{Seal, admission::Slots};
use crate::serve::storage::Storage;
type TestResult = Result<(), Box<dyn std::error::Error>>;
fn moved() -> Result<(tempfile::TempDir, Storage), Box<dyn std::error::Error>> {
let directory = tempfile::tempdir()?;
let mut storage = Storage::create_kept(
DatabaseConfig {
data_dir: directory.path().join("database"),
shard_count: 2,
distributed: None,
executor_threads: Some(1),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
KeptLimit::new(2)?,
Duration::from_secs(1),
)?;
for index in 0..5 {
let expected = storage.capture(&selection())?.version;
storage.mutate(&Mutation {
request_id: format!("clock-{index}"),
branch: "main".to_owned(),
expected,
change: Change::Collection(CollectionWrite {
collection: format!("c{index}"),
schema: None,
}),
})?;
}
Ok((directory, storage))
}
fn selection() -> Selection {
Selection {
branch: "main".to_owned(),
version: None,
}
}
fn rfc3339(nanos: u64) -> Result<String, Box<dyn std::error::Error>> {
let seconds = nanos / 1_000_000_000;
let days = i64::try_from(seconds / 86_400)? + 719_468;
let era = days / 146_097;
let within = days - era * 146_097;
let year = (within - within / 1460 + within / 36_524 - within / 146_096) / 365;
let ordinal = within - (365 * year + year / 4 - year / 100);
let month = (5 * ordinal + 2) / 153;
let day = ordinal - (153 * month + 2) / 5 + 1;
let month = month + if month < 10 { 3 } else { -9 };
let year = year + era * 400 + i64::from(month <= 2);
Ok(format!(
"{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}.{:09}Z",
seconds % 86_400 / 3600,
seconds % 3600 / 60,
seconds % 60,
nanos % 1_000_000_000
))
}
#[test]
fn window_instants_resolve_the_same_versions_as_history_list_across_spilled_and_kept_rows()
-> TestResult {
let fixture = moved()?;
let storage = &fixture.1;
let at = storage.capture(&selection())?;
let catalogue = storage.catalogue();
let clock = catalogue.window_clock(&at, NonZeroUsize::MAX)?;
let listed = catalogue.history("main", None, usize::MAX)?;
assert_eq!(
clock
.iter()
.map(|(version, _)| *version)
.collect::<Vec<_>>(),
listed.iter().map(|row| row.version).collect::<Vec<_>>()
);
assert_eq!(clock.len(), 6);
let access = access::request(&TestSource::denying(&[]))?;
let reader = storage.reader()?;
let slots = Slots::new(NonZeroUsize::MIN);
let seal = Seal::new()?;
let context = Context {
access: &access,
reader: &reader,
catalogue: &catalogue,
slots: &slots,
settings: None,
max_page: NonZeroU32::MAX,
seal: &seal,
};
let now = RawValue::from_string("9999999999.000000000".to_owned())?;
let (_, upper) = clock.last().ok_or("no clock head")?;
let upper_text = rfc3339(*upper)?;
for (version, timestamp) in &clock {
let text = rfc3339(*timestamp)?;
assert_eq!(window::instant(&text).map_err(str::to_owned)?, *timestamp);
let stage = window::read(&format!("window(\"{text}\"; \"{upper_text}\")"), 0)
.map_err(|error| error.to_string())?;
let Stage::Window(asked) = stage.value else {
return Err("wrong instant stage".into());
};
let actual = execution::resolve(&context, &at, &now, 100, &asked)?;
let first = clock
.iter()
.find(|(_, time)| time >= timestamp)
.ok_or("no matching history instant")?
.0;
let end = clock
.iter()
.find(|(_, time)| time >= upper)
.ok_or("no upper instant")?
.0;
for row in &listed {
let origin = Origin {
collection: "patients".to_owned(),
id: row.version.sequence.to_string(),
version: serde_json::value::to_raw_value(&row.version)?,
score: None,
};
assert_eq!(
actual.contains(&origin)?,
first.sequence <= row.version.sequence && row.version.sequence < end.sequence,
"instant for version {}",
version.sequence
);
}
}
Ok(())
}
#[test]
fn window_history_resolution_refuses_a_partial_scan() -> TestResult {
let fixture = moved()?;
let storage = &fixture.1;
let at = storage.capture(&selection())?;
let failure = storage
.catalogue()
.window_clock(&at, NonZeroUsize::MIN)
.err()
.ok_or("partial history clock accepted")?;
assert_eq!(failure.code, "query_window_history_limit");
Ok(())
}
#[test]
fn window_history_clock_stays_at_the_captured_version() -> TestResult {
let mut fixture = moved()?;
let storage = &mut fixture.1;
let at = storage.capture(&selection())?;
storage.mutate(&Mutation {
request_id: "later".to_owned(),
branch: "main".to_owned(),
expected: at.version,
change: Change::Collection(CollectionWrite {
collection: "later".to_owned(),
schema: None,
}),
})?;
let clock = storage.catalogue().window_clock(&at, NonZeroUsize::MAX)?;
assert_eq!(clock.last().ok_or("no clock")?.0, at.version);
Ok(())
}
#[test]
fn window_instant_boundary_refuses_released_or_unordered_history() -> TestResult {
let version = |sequence| Version {
generation: 7,
sequence,
};
let retained = [(version(2), 200), (version(3), 300)];
let failure = execution::version_at(&retained, version(3), 100)
.err()
.ok_or("released instant accepted")?;
assert_eq!(failure.code, "query_window_history_released");
let unordered = [(version(0), 200), (version(1), 100)];
let failure = execution::version_at(&unordered, version(1), 100)
.err()
.ok_or("unordered history accepted")?;
assert_eq!(failure.code, "query_window_history_clock");
assert_eq!(
execution::version_at(
&[(version(0), 100), (version(1), 100), (version(2), 200)],
version(2),
100
)?,
0
);
assert_eq!(
execution::version_at(&[(version(0), 100), (version(1), 200)], version(1), 201)?,
2
);
Ok(())
}