extern crate std;
use core::{
hash::{Hash, Hasher},
num::{NonZeroU32, NonZeroU64},
ptr,
};
use crate::{TimerCancellationRefusal, TimerOut, storage::warp_table::Capacity};
use super::*;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct PlayerId(u8);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum Lookup {
Found(PlayerId),
Absent(PlayerId),
}
struct Database;
impl StateMachine for Database {}
impl Pipe for Database {
type In<'message> = PlayerId;
type Out<'message> = Lookup;
}
impl StepInputOf<Database> for PlayerId {
type Outcome = Lookup;
fn step(self, _database: &mut Database) -> Self::Outcome {
if self.0 == 0 {
Lookup::Absent(self)
} else {
Lookup::Found(self)
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ReturnTo {
Requester,
}
const fn timeout_after() -> TimeoutAfter {
TimeoutAfter::milliseconds(NonZeroU64::MIN)
}
fn ledger(capacity: u32) -> Result<ContinuationLedger<ReturnTo>, &'static str> {
let capacity = Capacity::new(NonZeroU32::new(capacity).ok_or("capacity was zero")?);
ContinuationLedger::try_new(capacity).map_err(|_| "ledger storage unavailable")
}
fn admit(
ledger: &mut ContinuationLedger<ReturnTo>,
player: u8,
) -> Result<ContinuationAdmission<'static, Database>, &'static str> {
match ledger.admit(
ReturnTo::Requester,
Owed::new(PlayerId(player), timeout_after()),
) {
ContinuationAdmissionOutcome::Admitted { admission } => Ok(admission),
ContinuationAdmissionOutcome::Refused { .. } => Err("admission was refused"),
}
}
struct WordHasher(u64);
impl Hasher for WordHasher {
fn finish(&self) -> u64 {
self.0
}
fn write(&mut self, bytes: &[u8]) {
for byte in bytes {
self.0 = self.0.rotate_left(8) ^ u64::from(*byte);
}
}
fn write_u64(&mut self, value: u64) {
self.0 = value;
}
}
fn hash_value(value: &impl Hash) -> u64 {
let mut hasher = WordHasher(0);
value.hash(&mut hasher);
hasher.finish()
}
fn arm(
admission: ContinuationAdmission<'static, Database>,
) -> Result<ContinuationWork<'static, Database>, &'static str> {
let token = admission.timer_token();
let timer_outcome = TimerOut::Armed { token }
.into_arm_outcome()
.map_err(|_| "armed output was rejected")?;
match admission
.settle_timer_arm(timer_outcome)
.map_err(|_| "timer returned the wrong token")?
{
ContinuationArmOutcome::Armed { work } => Ok(work),
ContinuationArmOutcome::Refused { .. } => Err("timer refused the deadline"),
}
}
fn cancel(
completion: ContinuationCompletion<'static, Database>,
) -> Result<CancelledContinuationCompletion<'static, Database>, &'static str> {
let token = completion.timer_token();
let timer_outcome = TimerOut::Cancelled { token }
.into_cancellation_outcome()
.map_err(|_| "cancelled output was rejected")?;
match completion
.settle_timer_cancellation(timer_outcome)
.map_err(|_| "timer returned the wrong token")?
{
ContinuationCancellationOutcome::Cancelled { completion } => Ok(completion),
ContinuationCancellationOutcome::Stale { .. } => Err("completion was stale"),
}
}
#[test]
fn ledger_owns_only_the_return_address_while_work_flows() -> Result<(), &'static str> {
let mut ledger = ledger(2)?;
let admission = admit(&mut ledger, 7)?;
assert_eq!(ledger.len(), 1);
assert_eq!(
admission.timer_input(),
crate::TimerIn::Arm {
token: admission.timer_token(),
after: timeout_after(),
}
);
let completion = arm(admission)?.step(&mut Database);
let completion = cancel(completion)?;
let ContinuationResolutionOutcome::Resolved { return_to, result } = ledger.resolve(completion)
else {
return Err("completion was stale");
};
assert_eq!(return_to, ReturnTo::Requester);
assert!(matches!(result, Ok(Lookup::Found(PlayerId(7)))));
assert!(ledger.is_empty());
Ok(())
}
#[test]
fn worker_branded_ids_and_ledger_counts_preserve_their_values() -> Result<(), &'static str> {
let mut ledger = ledger(2)?;
assert_eq!(ledger.len(), 0);
assert!(ledger.is_empty());
let first = admit(&mut ledger, 1)?;
let second = admit(&mut ledger, 2)?;
let first_id = first.id();
let second_id = second.id();
assert_eq!(ledger.len(), 2);
assert!(!ledger.is_empty());
assert_eq!(first_id, first_id);
assert_ne!(first_id, second_id);
assert_eq!(
first_id.partial_cmp(&second_id),
Some(core::cmp::Ordering::Less)
);
assert_eq!(first_id.cmp(&second_id), core::cmp::Ordering::Less);
assert_eq!(
std::format!("{first_id:?}"),
std::format!("{}", first_id.value())
);
assert_eq!(hash_value(&first_id), hash_value(&first_id.value()));
Ok(())
}
#[test]
fn timer_refusal_returns_the_complete_owed_work() -> Result<(), &'static str> {
let mut ledger = ledger(1)?;
let admission = admit(&mut ledger, 9)?;
let token = admission.timer_token();
let reason = crate::TimerArmRefusal::AtCapacity {
maximum_deadlines: 1,
};
let timer_outcome = TimerOut::ArmRefused {
token,
after: timeout_after(),
reason,
}
.into_arm_outcome()
.map_err(|_| "refusal output was rejected")?;
let ContinuationArmOutcome::Refused { refusal } = admission
.settle_timer_arm(timer_outcome)
.map_err(|_| "timer returned the wrong token")?
else {
return Err("refused work was released");
};
let ContinuationRefusalOutcome::Refused { return_to, result } = ledger.refuse(refusal) else {
return Err("timer refusal was stale");
};
assert_eq!(return_to, ReturnTo::Requester);
let Err(ContinuationFailure::Unavailable {
reason: ContinuationUnavailable::TimerRefused { reason: actual },
rejected,
}) = result
else {
return Err("timer refusal was not preserved");
};
assert_eq!(actual, reason);
assert_eq!(rejected.work(), &PlayerId(9));
assert_eq!(rejected.timeout_after(), timeout_after());
assert!(ledger.is_empty());
Ok(())
}
#[test]
fn expiry_wins_once_and_late_completion_is_stale() -> Result<(), &'static str> {
let mut ledger = ledger(1)?;
let work = arm(admit(&mut ledger, 11)?)?;
let token = work.id().timer_token();
let completion = work.step(&mut Database);
let ContinuationExpirationOutcome::Expired {
return_to,
expiration,
} = ledger.expire(token)
else {
return Err("deadline was stale");
};
assert_eq!(return_to, ReturnTo::Requester);
assert_eq!(expiration.token(), token);
assert!(matches!(
expiration.into_result::<Database>(),
Err(ContinuationFailure::Expired)
));
let refused = TimerOut::CancellationRefused {
token,
reason: TimerCancellationRefusal::NotArmed,
}
.into_cancellation_outcome()
.map_err(|_| "cancellation refusal was rejected")?;
let ContinuationCancellationOutcome::Stale { completion } = completion
.settle_timer_cancellation(refused)
.map_err(|_| "timer returned the wrong token")?
else {
return Err("late completion displaced expiry");
};
assert_eq!(completion.output(), &Lookup::Found(PlayerId(11)));
assert!(matches!(
ledger.expire(token),
ContinuationExpirationOutcome::Stale { .. }
));
Ok(())
}
#[test]
fn capacity_refusal_preserves_address_work_and_timeout() -> Result<(), &'static str> {
let mut ledger = ledger(1)?;
let _first = admit(&mut ledger, 1)?;
let ContinuationAdmissionOutcome::Refused { return_to, result } =
ledger.admit::<Database>(ReturnTo::Requester, Owed::new(PlayerId(2), timeout_after()))
else {
return Err("full ledger admitted a second continuation");
};
assert_eq!(return_to, ReturnTo::Requester);
let Err(ContinuationFailure::Unavailable {
reason: ContinuationUnavailable::AtCapacity { maximum_pending },
rejected,
}) = result
else {
return Err("capacity refusal was not preserved");
};
assert_eq!(maximum_pending, 1);
assert_eq!(rejected.work(), &PlayerId(2));
assert_eq!(rejected.timeout_after(), timeout_after());
Ok(())
}
struct BorrowedEcho;
impl Pipe for BorrowedEcho {
type In<'message> = &'message str;
type Out<'message> = &'message str;
}
#[test]
fn synchronous_borrowed_work_is_not_copied() -> Result<(), &'static str> {
let text = "borrowed";
let mut ledger = ledger(1)?;
let ContinuationAdmissionOutcome::Admitted { admission } =
ledger.admit::<BorrowedEcho>(ReturnTo::Requester, Owed::new(text, timeout_after()))
else {
return Err("borrowed work was refused");
};
let token = admission.timer_token();
let armed = TimerOut::Armed { token }
.into_arm_outcome()
.map_err(|_| "armed output was rejected")?;
let ContinuationArmOutcome::Armed { work } = admission
.settle_timer_arm(armed)
.map_err(|_| "timer returned the wrong token")?
else {
return Err("timer refused borrowed work");
};
assert!(ptr::eq(work.input().as_ptr(), text.as_ptr()));
let completion = work.fulfill_with(|input| input);
assert!(ptr::eq(completion.output().as_ptr(), text.as_ptr()));
Ok(())
}