#![forbid(unsafe_code)]
#![warn(missing_docs)]
use std::sync::Mutex;
use pacta_contract::lifecycle::{self, State};
use pacta_contract::{Claim, Pact, Registry, Retainer, Timestamp, Transition};
use uuid::Uuid;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct NotHeld;
impl std::fmt::Display for NotHeld {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "retainer is not the current holder of any claim")
}
}
impl std::error::Error for NotHeld {}
impl From<lifecycle::NotCurrentHolder> for NotHeld {
fn from(_: lifecycle::NotCurrentHolder) -> Self {
NotHeld
}
}
struct Record {
pact: Pact,
state: State,
}
struct Store {
records: Mutex<Vec<Record>>,
lease_millis: u64,
}
impl Store {
fn seeded(pacts: Vec<Pact>, lease_millis: u64) -> Self {
Self {
records: Mutex::new(
pacts
.into_iter()
.map(|pact| Record {
pact,
state: State::Available,
})
.collect(),
),
lease_millis,
}
}
fn lease_millis(&self) -> u64 {
self.lease_millis
}
fn claim(&self, dockets: &[&str], now: Timestamp) -> Option<Claim> {
let mut records = self
.records
.lock()
.expect("registry mutex should not be poisoned");
let index = records.iter().position(|record| {
dockets.contains(&record.pact.docket.as_str())
&& lifecycle::is_claimable(&record.state, now)
})?;
let retainer = Retainer::new(Uuid::new_v4());
records[index].state = lifecycle::on_claim(&retainer, now, self.lease_millis);
let expiry = lifecycle::lease_expiry(now, self.lease_millis);
Some(Claim::new(records[index].pact.clone(), retainer, expiry))
}
fn apply(&self, retainer: &Retainer, transition: &Transition<'_>) -> Result<(), NotHeld> {
let mut records = self
.records
.lock()
.expect("registry mutex should not be poisoned");
let record = records
.iter_mut()
.find(|record| matches!(&record.state, State::Held { retainer: held, .. } if held == retainer))
.ok_or(NotHeld)?;
record.state = transition(&record.state)?;
Ok(())
}
}
pub struct MemoryRegistry {
store: Store,
}
impl MemoryRegistry {
#[must_use]
pub fn new(lease_millis: u64) -> Self {
Self::seeded(Vec::new(), lease_millis)
}
#[must_use]
pub fn seeded(pacts: Vec<Pact>, lease_millis: u64) -> Self {
Self {
store: Store::seeded(pacts, lease_millis),
}
}
}
impl Registry for MemoryRegistry {
type Error = NotHeld;
fn claim(&self, dockets: &[&str], now: Timestamp) -> Result<Option<Claim>, Self::Error> {
Ok(self.store.claim(dockets, now))
}
fn lease_millis(&self) -> u64 {
self.store.lease_millis()
}
fn apply(&self, retainer: &Retainer, transition: &Transition<'_>) -> Result<(), Self::Error> {
self.store.apply(retainer, transition)
}
}
#[cfg(feature = "async")]
pub struct MemoryRegistryAsync {
store: Store,
}
#[cfg(feature = "async")]
impl MemoryRegistryAsync {
#[must_use]
pub fn new(lease_millis: u64) -> Self {
Self::seeded(Vec::new(), lease_millis)
}
#[must_use]
pub fn seeded(pacts: Vec<Pact>, lease_millis: u64) -> Self {
Self {
store: Store::seeded(pacts, lease_millis),
}
}
}
#[cfg(feature = "async")]
impl pacta_contract::AsyncRegistry for MemoryRegistryAsync {
type Error = NotHeld;
async fn claim(&self, dockets: &[&str], now: Timestamp) -> Result<Option<Claim>, NotHeld> {
Ok(self.store.claim(dockets, now))
}
fn lease_millis(&self) -> u64 {
self.store.lease_millis()
}
async fn apply(&self, retainer: &Retainer, transition: &Transition<'_>) -> Result<(), NotHeld> {
self.store.apply(retainer, transition)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::marker::PhantomData;
use std::rc::Rc;
struct LocalRegistry {
inner: MemoryRegistry,
_local: Rc<()>,
}
impl LocalRegistry {
fn seeded(pacts: Vec<Pact>, lease_millis: u64) -> Self {
Self {
inner: MemoryRegistry::seeded(pacts, lease_millis),
_local: Rc::new(()),
}
}
}
impl Registry for LocalRegistry {
type Error = NotHeld;
fn claim(&self, dockets: &[&str], now: Timestamp) -> Result<Option<Claim>, Self::Error> {
self.inner.claim(dockets, now)
}
fn lease_millis(&self) -> u64 {
self.inner.lease_millis()
}
fn apply(
&self,
retainer: &Retainer,
transition: &Transition<'_>,
) -> Result<(), Self::Error> {
self.inner.apply(retainer, transition)
}
}
#[cfg(feature = "async")]
struct LocalRegistryAsync {
inner: MemoryRegistryAsync,
_local: Rc<()>,
}
#[cfg(feature = "async")]
impl LocalRegistryAsync {
fn seeded(pacts: Vec<Pact>, lease_millis: u64) -> Self {
Self {
inner: MemoryRegistryAsync::seeded(pacts, lease_millis),
_local: Rc::new(()),
}
}
}
#[cfg(feature = "async")]
impl pacta_contract::AsyncRegistry for LocalRegistryAsync {
type Error = NotHeld;
async fn claim(
&self,
dockets: &[&str],
now: Timestamp,
) -> Result<Option<Claim>, Self::Error> {
pacta_contract::AsyncRegistry::claim(&self.inner, dockets, now).await
}
fn lease_millis(&self) -> u64 {
pacta_contract::AsyncRegistry::lease_millis(&self.inner)
}
async fn apply(
&self,
retainer: &Retainer,
transition: &Transition<'_>,
) -> Result<(), Self::Error> {
pacta_contract::AsyncRegistry::apply(&self.inner, retainer, transition).await
}
}
#[cfg(feature = "async")]
#[derive(Clone, Copy)]
struct LocalDriver(PhantomData<Rc<()>>);
#[cfg(feature = "async")]
impl pacta_conformance::BlockingDriver for LocalDriver {
fn drive<F: core::future::Future>(&self, future: F) -> F::Output {
pacta_conformance::BlockingDriver::drive(&pacta_conformance::SelfProgress, future)
}
}
#[test]
fn passes_registry_conformance() {
pacta_conformance::run(MemoryRegistry::seeded);
}
#[test]
fn local_sync_backend_passes_sequential_conformance() {
pacta_conformance::run(LocalRegistry::seeded);
}
#[test]
fn passes_sync_contention() {
pacta_conformance::run_contention(MemoryRegistry::seeded);
}
fn a_pact() -> Pact {
Pact::new(Uuid::new_v4(), "d".to_string(), "k".to_string(), Vec::new())
}
#[test]
fn release_rejects_a_non_holder() {
let registry = MemoryRegistry::seeded(vec![a_pact()], 1000);
registry
.claim(&["d"], Timestamp::from_millis(0))
.expect("claim should not error")
.expect("a pact should be claimable");
let stranger = Retainer::new(Uuid::new_v4());
assert_eq!(
registry.release(&stranger, Timestamp::from_millis(0)),
Err(NotHeld),
"release by a non-holder must be rejected, like fulfill and breach"
);
}
#[test]
fn a_settled_pact_cannot_be_released() {
let registry = MemoryRegistry::seeded(vec![a_pact()], 1000);
let claim = registry
.claim(&["d"], Timestamp::from_millis(0))
.expect("claim should not error")
.expect("a pact should be claimable");
registry
.fulfill(&claim.retainer)
.expect("fulfill should settle");
assert_eq!(
registry.release(&claim.retainer, Timestamp::from_millis(0)),
Err(NotHeld),
"a concluded obligation has no claim to relinquish"
);
}
#[test]
fn apply_rejects_a_stranger_even_with_an_any_state_transition() {
let registry = MemoryRegistry::seeded(vec![a_pact()], 1000);
let claim = registry
.claim(&["d"], Timestamp::from_millis(0))
.expect("claim should not error")
.expect("a pact should be claimable");
let stranger = Retainer::new(Uuid::new_v4());
let accept_any = |_state: &State| Ok::<State, lifecycle::NotCurrentHolder>(State::Settled);
assert_eq!(
registry.apply(&stranger, &accept_any),
Err(NotHeld),
"a retainer that holds no record cannot apply, even an any-state transition"
);
registry
.fulfill(&claim.retainer)
.expect("the held state was untouched, so the holder still settles");
}
#[test]
fn apply_admits_the_true_holder() {
let registry = MemoryRegistry::seeded(vec![a_pact()], 1000);
let claim = registry
.claim(&["d"], Timestamp::from_millis(0))
.expect("claim should not error")
.expect("a pact should be claimable");
registry
.heartbeat(&claim.retainer, Timestamp::from_millis(500))
.expect("the holder's heartbeat extends the lease");
registry
.release(&claim.retainer, Timestamp::from_millis(0))
.expect("the holder releases");
assert_eq!(
registry.fulfill(&claim.retainer),
Err(NotHeld),
"release rotated authority, so the prior retainer no longer holds a record"
);
}
#[cfg(feature = "async")]
#[test]
fn passes_async_conformance() {
pacta_conformance::run_async(MemoryRegistryAsync::seeded);
}
#[cfg(feature = "async")]
#[test]
fn local_async_backend_passes_ready_future_conformance() {
pacta_conformance::run_async(LocalRegistryAsync::seeded);
}
#[cfg(feature = "async")]
#[test]
fn local_async_backend_and_driver_pass_runtime_compatible_conformance() {
pacta_conformance::run_async_with(LocalRegistryAsync::seeded, LocalDriver(PhantomData));
}
#[cfg(feature = "async")]
#[tokio::test]
async fn async_apply_rejects_a_stranger_even_with_an_any_state_transition() {
use pacta_contract::AsyncRegistry;
let registry = MemoryRegistryAsync::seeded(vec![a_pact()], 1000);
let claim = registry
.claim(&["d"], Timestamp::from_millis(0))
.await
.expect("claim should not error")
.expect("a pact should be claimable");
let stranger = Retainer::new(Uuid::new_v4());
let accept_any = |_state: &State| Ok::<State, lifecycle::NotCurrentHolder>(State::Settled);
assert_eq!(
registry.apply(&stranger, &accept_any).await,
Err(NotHeld),
"the async binding also locates by retainer"
);
registry
.fulfill(&claim.retainer)
.await
.expect("the held state was untouched, so the holder still settles");
}
#[cfg(feature = "async")]
#[test]
fn passes_async_contention() {
pacta_conformance::run_async_contention(MemoryRegistryAsync::seeded);
}
}