use bytes::Bytes;
use futures::StreamExt as _;
use crate::{
AppendOutcome, AppendRequest, ChildrenRequest, CommitCondition, CommitConflict, CommitMutation,
CommitOutcome, CommitRequest, ForkRequest, IdempotencyKey, IdempotencyOutcome, ReadRequest,
StreamError, StreamPath, StreamProvider,
};
pub const SUITE: &[u8] = include_bytes!("../conformance/stream.json");
#[allow(
clippy::too_many_lines,
reason = "linear conformance walkthrough; each check is a distinct provider-contract \
assertion, and splitting it would only move the same sequential checks behind \
indirection"
)]
pub async fn verify(provider: &dyn StreamProvider) -> Result<(), String> {
if SUITE.is_empty() {
return Err("Stream conformance inventory is empty".into());
}
let source = path("conformance/source")?;
let child = path("conformance/child")?;
let append_key = key(b"stream-append")?;
if provider
.inspect_idempotency(append_key.clone())
.await
.map_err(|err| error(&err))?
.is_some()
{
return Err("unknown idempotency identity was reported as retained".into());
}
let initial = AppendRequest {
path: source.clone(),
records: vec![Bytes::from_static(b"one"), Bytes::from_static(b"two")],
if_tail: Some(0),
idempotency_key: Some(append_key.clone()),
};
let first = provider
.append(initial.clone())
.await
.map_err(|err| error(&err))?;
let AppendOutcome::Committed(first_receipt) = &first else {
return Err("initial tail CAS conflicted".into());
};
if first_receipt.start != 0
|| first_receipt.end != 2
|| first_receipt.tail != 2
|| provider.append(initial).await.map_err(|err| error(&err))? != first
{
return Err("atomic append or exact replay changed its receipt".into());
}
if provider
.append(AppendRequest {
path: source.clone(),
records: vec![Bytes::from_static(b"different")],
if_tail: Some(0),
idempotency_key: Some(append_key.clone()),
})
.await
!= Err(StreamError::IdempotencyMismatch)
{
return Err("changed idempotency arguments were accepted".into());
}
let append_observation = provider
.inspect_idempotency(append_key.clone())
.await
.map_err(|err| error(&err))?
.ok_or_else(|| "committed append identity was not retained".to_owned())?;
if append_observation.idempotency_key != append_key
|| append_observation.request_digest == [0; 32]
|| append_observation.outcome != IdempotencyOutcome::Append(first.clone())
{
return Err("append inspection did not return the exact retained outcome".into());
}
if provider
.append(AppendRequest {
path: source.clone(),
records: vec![Bytes::from_static(b"never")],
if_tail: Some(1),
idempotency_key: None,
})
.await
.map_err(|err| error(&err))?
!= (AppendOutcome::TailConflict { actual_tail: 2 })
{
return Err("tail conflict was not returned as data".into());
}
let fork = provider
.fork(ForkRequest {
source: source.clone(),
destination: child.clone(),
at_tail: Some(1),
idempotency_key: Some(key(b"stream-fork")?),
})
.await
.map_err(|err| error(&err))?;
if fork.forked_at != 1 || fork.tail != 1 {
return Err("fork did not retain the exact immutable prefix".into());
}
let inherited = provider
.read(ReadRequest {
path: child.clone(),
from: 0,
limit: 8,
})
.await
.map_err(|err| error(&err))?
.collect::<Vec<_>>()
.await;
let [item] = inherited.as_slice() else {
return Err("forked history was not exact".into());
};
if item.as_ref().map_err(ToString::to_string)?.value != Bytes::from_static(b"one") {
return Err("forked history was not exact".into());
}
let children = provider
.children(ChildrenRequest {
parent: Some(path("conformance")?),
limit: 8,
})
.await
.map_err(|err| error(&err))?
.collect::<Vec<_>>()
.await
.into_iter()
.collect::<Result<Vec<_>, _>>()
.map_err(|err| error(&err))?;
let mut child_paths = children
.into_iter()
.map(|child| child.path)
.collect::<Vec<_>>();
child_paths.sort();
if child_paths != vec![path("conformance/child")?, path("conformance/source")?] {
return Err("direct child listing changed its fixed snapshot".into());
}
let mut follow = provider
.follow(child.clone(), 1)
.await
.map_err(|err| error(&err))?;
provider
.append(AppendRequest {
path: child.clone(),
records: vec![Bytes::from_static(b"live")],
if_tail: Some(1),
idempotency_key: None,
})
.await
.map_err(|err| error(&err))?;
let live = tokio::time::timeout(std::time::Duration::from_secs(1), follow.next())
.await
.map_err(|_| "follow did not make bounded progress".to_owned())?
.ok_or_else(|| "follow ended".to_owned())?
.map_err(|err| error(&err))?;
if live.sequence != 1 || live.value != Bytes::from_static(b"live") {
return Err("follow returned a gap or duplicate".into());
}
let trim_key = key(b"stream-trim")?;
let trimmed = provider
.trim(child.clone(), 1, trim_key.clone())
.await
.map_err(|err| error(&err))?;
if trimmed.trim_point != 1
|| provider
.trim(child.clone(), 1, trim_key.clone())
.await
.map_err(|err| error(&err))?
!= trimmed
|| provider.trim(child.clone(), 2, trim_key).await != Err(StreamError::IdempotencyMismatch)
{
return Err("logical trim replay or mismatch semantics changed".into());
}
let trimmed_read = provider
.read(ReadRequest {
path: child.clone(),
from: 0,
limit: 1,
})
.await;
let trimmed = match trimmed_read {
Err(StreamError::OutOfRange) => true,
Ok(mut records) => matches!(records.next().await, Some(Err(StreamError::OutOfRange))),
_ => false,
};
if !trimmed {
return Err("trimmed history remained publicly readable".into());
}
let delete_key = key(b"stream-delete")?;
let deleted = provider
.delete(child.clone(), delete_key.clone())
.await
.map_err(|err| error(&err))?;
if provider
.delete(child.clone(), delete_key)
.await
.map_err(|err| error(&err))?
!= deleted
|| provider.tail(child.clone()).await != Err(StreamError::Retired)
{
return Err("permanent deletion or exact replay changed".into());
}
let committed_path = path("conformance/committed")?;
let request = CommitRequest {
conditions: vec![
CommitCondition::Tail {
path: source.clone(),
expected: 2,
},
CommitCondition::Absent {
path: committed_path.clone(),
},
],
mutations: vec![CommitMutation::Fork {
source: source.clone(),
destination: committed_path,
at_tail: 2,
}],
idempotency_key: key(b"stream-commit")?,
};
let committed = provider
.commit(request.clone())
.await
.map_err(|err| error(&err))?;
let CommitOutcome::Committed(envelope) = &committed else {
return Err("valid coordinated commit conflicted".into());
};
if provider.commit(request).await.map_err(|err| error(&err))? != committed
|| provider
.read_commit(envelope.commit_id)
.await
.map_err(|err| error(&err))?
!= *envelope
{
return Err("coordinated commit replay or envelope changed".into());
}
verify_stale_tail_condition(provider, source.clone()).await?;
let absent = path("conformance/absent-tail")?;
let absent_key = key(b"stream-absent-tail")?;
let absent_request = CommitRequest {
conditions: vec![CommitCondition::Tail {
path: absent.clone(),
expected: 0,
}],
mutations: vec![CommitMutation::Append {
path: absent.clone(),
records: vec![Bytes::from_static(b"must-not-commit")],
}],
idempotency_key: absent_key.clone(),
};
let absent_conflict = CommitOutcome::Conflict(vec![CommitConflict::Tail {
path: absent.clone(),
expected: 0,
actual: None,
}]);
if provider
.commit(absent_request.clone())
.await
.map_err(|err| error(&err))?
!= absent_conflict
|| provider
.commit(absent_request)
.await
.map_err(|err| error(&err))?
!= absent_conflict
|| provider.tail(absent).await != Err(StreamError::NotFound)
{
return Err("absent tail condition did not return one replayable conflict".into());
}
let absent_observation = provider
.inspect_idempotency(absent_key.clone())
.await
.map_err(|err| error(&err))?
.ok_or_else(|| "commit conflict identity was not retained".to_owned())?;
if absent_observation.idempotency_key != absent_key
|| absent_observation.request_digest == [0; 32]
|| absent_observation.outcome != IdempotencyOutcome::Commit(absent_conflict)
{
return Err("commit-conflict inspection did not preserve the terminal result".into());
}
Ok(())
}
async fn verify_stale_tail_condition(
provider: &dyn StreamProvider,
source: StreamPath,
) -> Result<(), String> {
let stale_path = path("conformance/stale-commit")?;
let stale_conflict = CommitOutcome::Conflict(vec![CommitConflict::Tail {
path: source.clone(),
expected: 1,
actual: Some(2),
}]);
if provider
.commit(CommitRequest {
conditions: vec![
CommitCondition::Tail {
path: source.clone(),
expected: 1,
},
CommitCondition::Absent {
path: stale_path.clone(),
},
],
mutations: vec![CommitMutation::Fork {
source: source.clone(),
destination: stale_path.clone(),
at_tail: 2,
}],
idempotency_key: key(b"stream-stale-commit")?,
})
.await
.map_err(|err| error(&err))?
!= stale_conflict
|| provider.tail(stale_path).await != Err(StreamError::NotFound)
|| provider.tail(source).await.map_err(|err| error(&err))? != 2
{
return Err("stale tail condition mutated a coordinated commit".into());
}
Ok(())
}
fn path(value: &str) -> Result<StreamPath, String> {
StreamPath::new(value).map_err(|err| error(&err))
}
fn key(value: &'static [u8]) -> Result<IdempotencyKey, String> {
IdempotencyKey::new(Bytes::from_static(value)).map_err(|err| error(&err))
}
fn error(error: &impl ToString) -> String {
error.to_string()
}
#[cfg(test)]
mod tests {
use super::verify;
use crate::{MemoryLimits, MemoryStream};
#[tokio::test]
async fn memory_provider_passes_the_public_suite() -> Result<(), String> {
verify(&MemoryStream::new(MemoryLimits::default())).await
}
}