use alloc::{string::String, vec::Vec};
use core::num::{NonZeroU32, NonZeroU64};
use crate::{TimerCancellationRefusal, TimerOut, storage::warp_table::Capacity};
use super::*;
struct Worker;
impl Pipe for Worker {
type In<'message> = Vec<u8>;
type Out<'message> = ();
}
fn armed_work(
ledger: &mut ContinuationLedger<u8>,
return_to: u8,
input: Vec<u8>,
timeout_after: TimeoutAfter,
) -> Result<ContinuationWork<'static, Worker>, &'static str> {
let ContinuationAdmissionOutcome::Admitted { admission } =
ledger.admit::<Worker>(return_to, Owed::new(input, timeout_after))
else {
return Err("work was refused");
};
let arm = TimerOut::Armed {
token: admission.timer_token(),
}
.into_arm_outcome()
.map_err(|_| "arm fixture was not an arm outcome")?;
let ContinuationArmOutcome::Armed { work } = admission
.settle_timer_arm(arm)
.map_err(|_| "arm fixture returned the wrong token")?
else {
return Err("work was not armed");
};
Ok(work)
}
fn cancelled_work(
work: ContinuationWork<'static, Worker>,
) -> Result<CancelledContinuationWork<'static, Worker>, &'static str> {
let cancelled = TimerOut::Cancelled {
token: work.timer_token(),
}
.into_cancellation_outcome()
.map_err(|_| "cancellation fixture was not a cancellation outcome")?;
let ContinuationWorkCancellationOutcome::Cancelled { work } = work
.settle_timer_cancellation(cancelled)
.map_err(|_| "cancellation fixture returned the wrong token")?
else {
return Err("work was stale");
};
Ok(work)
}
#[test]
fn retraction_recovers_the_original_allocation_timeout_and_address_once() -> Result<(), &'static str>
{
let mut ledger = ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN))
.map_err(|_| "ledger unavailable")?;
let input = Vec::from(b"unstarted work".as_slice());
let original_address = input.as_ptr();
let original_capacity = input.capacity();
let timeout = TimeoutAfter::seconds(NonZeroU64::new(7).ok_or("zero timeout")?);
let work = armed_work(&mut ledger, 19, input, timeout)?;
let token = work.timer_token();
assert_eq!(work.timeout_after(), timeout);
assert_eq!(work.timer_input(), TimerIn::Cancel { token });
let work = cancelled_work(work)?;
assert_eq!(work.id().value(), token.value());
assert_eq!(work.owed().work().as_ptr(), original_address);
let ContinuationRetractionOutcome::Retracted {
return_to,
rejected,
} = ledger.retract(work)
else {
return Err("cancelled work was stale");
};
assert_eq!(return_to, 19);
assert_eq!(rejected.work().as_slice(), b"unstarted work");
assert_eq!(rejected.work().as_ptr(), original_address);
assert_eq!(rejected.work().capacity(), original_capacity);
assert_eq!(rejected.timeout_after(), timeout);
assert!(ledger.is_empty());
assert!(
matches!(ledger.expire(token), ContinuationExpirationOutcome::Stale { token: old } if old == token)
);
let ContinuationAdmissionOutcome::Admitted { admission } = ledger.admit::<Worker>(29, rejected)
else {
return Err("retraction did not release capacity");
};
assert_ne!(admission.timer_token(), token);
assert_eq!(admission.timeout_after(), timeout);
Ok(())
}
#[test]
fn refused_and_mismatched_cancellation_preserve_unstarted_work_without_releasing_the_ledger()
-> Result<(), &'static str> {
let mut ledger = ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN))
.map_err(|_| "ledger unavailable")?;
let input = Vec::from(b"not cancelled".as_slice());
let original_address = input.as_ptr();
let timeout = TimeoutAfter::milliseconds(NonZeroU64::MIN);
let work = armed_work(&mut ledger, 19, input, timeout)?;
let token = work.timer_token();
let wrong_token = TimerToken::from(NonZeroU64::new(2).ok_or("zero wrong token")?);
let wrong_ack = TimerOut::Cancelled { token: wrong_token }
.into_cancellation_outcome()
.map_err(|_| "invalid cancellation fixture")?;
let Err(mismatch) = work.settle_timer_cancellation(wrong_ack) else {
return Err("wrong token was accepted");
};
let (work, decision) = mismatch.into_parts();
assert_eq!(
decision,
TimerCancellationDecision::Cancelled { token: wrong_token }
);
assert_eq!(work.timer_token(), token);
assert_eq!(work.input().as_ptr(), original_address);
assert_eq!(work.timeout_after(), timeout);
let refusal = TimerOut::CancellationRefused {
token,
reason: TimerCancellationRefusal::NotArmed,
}
.into_cancellation_outcome()
.map_err(|_| "invalid cancellation refusal")?;
let ContinuationWorkCancellationOutcome::Stale { work } = work
.settle_timer_cancellation(refusal)
.map_err(|_| "refusal returned the wrong token")?
else {
return Err("refused cancellation certified work");
};
assert_eq!(work.input().as_ptr(), original_address);
assert_eq!(work.timeout_after(), timeout);
assert_eq!(ledger.len(), 1);
assert!(matches!(
ledger.retract(cancelled_work(work)?),
ContinuationRetractionOutcome::Retracted { return_to: 19, .. }
));
assert!(ledger.is_empty());
Ok(())
}
#[test]
fn an_expired_retraction_preserves_ownership_and_cannot_remove_a_reused_slot()
-> Result<(), &'static str> {
let mut ledger = ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN))
.map_err(|_| "ledger unavailable")?;
let input = Vec::from(b"expired work".as_slice());
let original_address = input.as_ptr();
let timeout = TimeoutAfter::nanoseconds(NonZeroU64::MIN);
let work = armed_work(&mut ledger, 19, input, timeout)?;
let token = work.timer_token();
assert!(matches!(
ledger.expire(token),
ContinuationExpirationOutcome::Expired { return_to: 19, .. }
));
let current = armed_work(&mut ledger, 29, Vec::new(), timeout)?;
let ContinuationRetractionOutcome::Stale { work } = ledger.retract(cancelled_work(work)?)
else {
return Err("expired retraction removed current work");
};
assert_eq!(work.id().value(), token.value());
assert_eq!(work.owed().work().as_ptr(), original_address);
assert_eq!(work.owed().timeout_after(), timeout);
assert_eq!(ledger.len(), 1);
assert!(matches!(
ledger.retract(cancelled_work(current)?),
ContinuationRetractionOutcome::Retracted { return_to: 29, .. }
));
assert!(ledger.is_empty());
Ok(())
}
#[test]
fn retraction_preserves_borrowed_work_through_synchronous_routing() -> Result<(), &'static str> {
struct Borrowed;
impl Pipe for Borrowed {
type In<'message> = &'message str;
type Out<'message> = ();
}
let input = String::from("borrowed work");
let mut ledger = ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN))
.map_err(|_| "ledger unavailable")?;
let timeout = TimeoutAfter::milliseconds(NonZeroU64::MIN);
let ContinuationAdmissionOutcome::Admitted { admission } =
ledger.admit::<Borrowed>(19u8, Owed::new(input.as_str(), timeout))
else {
return Err("borrowed work was refused");
};
let token = admission.timer_token();
let armed = TimerOut::Armed { token }
.into_arm_outcome()
.map_err(|_| "invalid arm fixture")?;
let ContinuationArmOutcome::Armed { work } = admission
.settle_timer_arm(armed)
.map_err(|_| "wrong arm token")?
else {
return Err("borrowed work was not armed");
};
let cancelled = TimerOut::Cancelled { token }
.into_cancellation_outcome()
.map_err(|_| "invalid cancellation fixture")?;
let ContinuationWorkCancellationOutcome::Cancelled { work } = work
.settle_timer_cancellation(cancelled)
.map_err(|_| "wrong cancellation token")?
else {
return Err("borrowed work was stale");
};
let ContinuationRetractionOutcome::Retracted {
return_to,
rejected,
} = ledger.retract(work)
else {
return Err("borrowed retraction was stale");
};
assert_eq!(return_to, 19);
assert_eq!(rejected.work().as_ptr(), input.as_ptr());
assert_eq!(rejected.timeout_after(), timeout);
assert!(ledger.is_empty());
Ok(())
}