use super::*;
use crate::{TimerOut, storage::warp_table::Capacity};
use alloc::boxed::Box;
use core::num::{NonZeroU32, NonZeroU64};
struct Worker;
impl Pipe for Worker {
type In<'a> = Box<u64>;
type Out<'a> = Box<u64>;
}
fn begin(
ledger: &mut ContinuationLedger<u8>,
address: u8,
input: Box<u64>,
) -> Result<ContinuationWork<'static, Worker>, &'static str> {
let ContinuationAdmissionOutcome::Admitted { admission } = ledger.admit::<Worker>(
address,
Owed::new(input, TimeoutAfter::seconds(NonZeroU64::MIN)),
) else {
return Err("admission");
};
let armed = TimerOut::Armed {
token: admission.timer_token(),
}
.into_arm_outcome()
.map_err(|_| "arm outcome")?;
match admission
.settle_timer_arm(armed)
.map_err(|_| "arm mismatch")?
{
ContinuationArmOutcome::Armed { work } => Ok(work),
ContinuationArmOutcome::Refused { .. } => Err("refused"),
}
}
#[test]
fn deferred_reply_preserves_owned_allocation_and_return_address() -> Result<(), &'static str> {
let mut ledger =
ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN)).map_err(|_| "ledger")?;
let input = Box::new(41);
let pointer = core::ptr::from_ref(input.as_ref());
let work = begin(&mut ledger, 7, input)?;
let id = work.id();
let (mut input, reply) = work.begin();
assert_eq!(core::ptr::from_ref(input.as_ref()), pointer);
*input = 42;
assert_eq!(ledger.len(), 1);
let completion = reply.fulfill(input);
assert_eq!(completion.id(), id);
assert_eq!(core::ptr::from_ref(completion.output().as_ref()), pointer);
let cancelled = TimerOut::Cancelled {
token: completion.timer_token(),
}
.into_cancellation_outcome()
.map_err(|_| "cancel outcome")?;
let ContinuationCancellationOutcome::Cancelled { completion } = completion
.settle_timer_cancellation(cancelled)
.map_err(|_| "cancel mismatch")?
else {
return Err("stale cancellation");
};
let ContinuationResolutionOutcome::Resolved { return_to, result } = ledger.resolve(completion)
else {
return Err("stale completion");
};
assert_eq!(return_to, 7);
let value = result.map_err(|_| "failed completion")?;
assert_eq!(*value, 42);
assert_eq!(core::ptr::from_ref(value.as_ref()), pointer);
assert!(ledger.is_empty());
Ok(())
}
#[test]
fn expired_deferred_work_can_finish_without_resolving_a_reused_slot() -> Result<(), &'static str> {
let mut ledger =
ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN)).map_err(|_| "ledger")?;
let work = begin(&mut ledger, 7, Box::new(41))?;
let token = work.timer_token();
let (mut input, reply) = work.begin();
let ContinuationExpirationOutcome::Expired {
return_to,
expiration,
} = ledger.expire(token)
else {
return Err("expiry");
};
assert_eq!(return_to, 7);
assert!(matches!(
expiration.into_result::<Worker>(),
Err(ContinuationFailure::Expired)
));
let next = begin(&mut ledger, 9, Box::new(5))?;
assert_ne!(next.timer_token(), token);
*input = 42;
let completion = reply.fulfill(input);
assert_eq!(**completion.output(), 42);
let cancelled = TimerOut::Cancelled { token }
.into_cancellation_outcome()
.map_err(|_| "cancel outcome")?;
let ContinuationCancellationOutcome::Cancelled { completion } = completion
.settle_timer_cancellation(cancelled)
.map_err(|_| "cancel mismatch")?
else {
return Err("cancel refused");
};
assert!(matches!(
ledger.resolve(completion),
ContinuationResolutionOutcome::Stale { .. }
));
assert_eq!(ledger.len(), 1);
assert!(matches!(
ledger.expire(next.timer_token()),
ContinuationExpirationOutcome::Expired { return_to: 9, .. }
));
Ok(())
}
#[test]
fn deferred_reply_can_preserve_an_inline_borrow() -> Result<(), &'static str> {
struct Borrowed;
impl Pipe for Borrowed {
type In<'a> = &'a mut u64;
type Out<'a> = &'a u64;
}
let mut value = 41;
let mut ledger =
ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN)).map_err(|_| "ledger")?;
let ContinuationAdmissionOutcome::Admitted { admission } = ledger.admit::<Borrowed>(
(),
Owed::new(&mut value, TimeoutAfter::seconds(NonZeroU64::MIN)),
) else {
return Err("admission");
};
let armed = TimerOut::Armed {
token: admission.timer_token(),
}
.into_arm_outcome()
.map_err(|_| "arm outcome")?;
let ContinuationArmOutcome::Armed { work } = admission
.settle_timer_arm(armed)
.map_err(|_| "arm mismatch")?
else {
return Err("arm refused");
};
let (input, reply) = work.begin();
*input = 42;
let completion = reply.fulfill(input);
assert_eq!(**completion.output(), 42);
Ok(())
}
#[test]
fn mismatched_timer_acknowledgments_preserve_the_deferred_obligation() -> Result<(), &'static str> {
let mut ledger =
ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN)).map_err(|_| "ledger")?;
let ContinuationAdmissionOutcome::Admitted { admission } = ledger.admit::<Worker>(
7,
Owed::new(Box::new(42), TimeoutAfter::seconds(NonZeroU64::MIN)),
) else {
return Err("admission");
};
let token = admission.timer_token();
let wrong = TimerToken::from_nonzero(NonZeroU64::new(2).ok_or("token")?);
let armed = TimerOut::Armed { token: wrong }
.into_arm_outcome()
.map_err(|_| "arm")?;
let Err(mismatch) = admission.settle_timer_arm(armed) else {
return Err("accepted mismatched arm");
};
let (admission, decision) = mismatch.into_parts();
assert!(matches!(decision, TimerArmDecision::Armed { token } if token == wrong));
assert_eq!(admission.timer_token(), token);
assert_eq!(Clone::clone(&admission.id()), admission.id());
let armed = TimerOut::Armed { token }
.into_arm_outcome()
.map_err(|_| "arm")?;
let ContinuationArmOutcome::Armed { work } = admission
.settle_timer_arm(armed)
.map_err(|_| "arm mismatch")?
else {
return Err("refused");
};
let (input, reply) = work.begin();
let completion = reply.fulfill(input);
let cancel = TimerOut::Cancelled { token: wrong }
.into_cancellation_outcome()
.map_err(|_| "cancel")?;
let Err(mismatch) = completion.settle_timer_cancellation(cancel) else {
return Err("accepted mismatched cancel");
};
let (completion, decision) = mismatch.into_parts();
assert!(matches!(decision, TimerCancellationDecision::Cancelled { token } if token == wrong));
assert_eq!(completion.timer_token(), token);
assert_eq!(**completion.output(), 42);
let cancel = TimerOut::Cancelled { token }
.into_cancellation_outcome()
.map_err(|_| "cancel")?;
let ContinuationCancellationOutcome::Cancelled { completion } = completion
.settle_timer_cancellation(cancel)
.map_err(|_| "cancel mismatch")?
else {
return Err("stale cancellation");
};
assert!(matches!(
ledger.resolve(completion),
ContinuationResolutionOutcome::Resolved { return_to: 7, .. }
));
assert!(ledger.is_empty());
Ok(())
}
#[test]
fn refused_admission_retains_its_input_and_exact_timer_reason() -> Result<(), &'static str> {
let mut ledger =
ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN)).map_err(|_| "ledger")?;
let timeout = TimeoutAfter::seconds(NonZeroU64::MIN);
let ContinuationAdmissionOutcome::Admitted { admission } =
ledger.admit::<Worker>(7, Owed::new(Box::new(42), timeout))
else {
return Err("admission");
};
let id = admission.id();
let reason = crate::TimerArmRefusal::AtCapacity {
maximum_deadlines: 1,
};
let refused = TimerOut::ArmRefused {
token: admission.timer_token(),
after: timeout,
reason,
}
.into_arm_outcome()
.map_err(|_| "arm")?;
let ContinuationArmOutcome::Refused { refusal } = admission
.settle_timer_arm(refused)
.map_err(|_| "arm mismatch")?
else {
return Err("armed");
};
assert_eq!(refusal.id(), id);
assert_eq!(refusal.reason(), reason);
let owed = refusal.into_owed();
assert_eq!(**owed.work(), 42);
assert_eq!(owed.timeout_after(), timeout);
assert!(matches!(
ledger.expire(id.timer_token()),
ContinuationExpirationOutcome::Expired { return_to: 7, .. }
));
Ok(())
}
#[test]
fn immediate_completion_uses_the_same_token_and_empty_work_keeps_its_timeout()
-> Result<(), &'static str> {
let timeout = TimeoutAfter::seconds(NonZeroU64::MIN);
let wait = Owed::wait(timeout);
assert_eq!(*wait.work(), ());
assert_eq!(wait.timeout_after(), timeout);
let mut ledger =
ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN)).map_err(|_| "ledger")?;
let work = begin(&mut ledger, 7, Box::new(41))?;
let token = work.timer_token();
let completion = work.fulfill(Box::new(42));
assert_eq!(completion.timer_token(), token);
assert_eq!(**completion.output(), 42);
Ok(())
}
#[test]
fn consumed_envelopes_preserve_owned_payloads_and_do_not_settle_the_ledger()
-> Result<(), &'static str> {
let timeout = TimeoutAfter::seconds(NonZeroU64::MIN);
let payload = Box::new(42);
let address = core::ptr::from_ref(payload.as_ref());
let (payload, retained_timeout) = Owed::new(payload, timeout).into_parts();
assert_eq!(core::ptr::from_ref(payload.as_ref()), address);
assert_eq!(retained_timeout, timeout);
let mut ledger =
ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN)).map_err(|_| "ledger")?;
let work = begin(&mut ledger, 7, payload)?;
let id = work.id();
let completion = work.fulfill_with(|payload| payload);
let (returned_id, payload) = completion.into_parts();
assert_eq!(returned_id, id);
assert_eq!(core::ptr::from_ref(payload.as_ref()), address);
assert_eq!(*payload, 42);
assert_eq!(ledger.len(), 1);
assert!(matches!(
ledger.expire(id.timer_token()),
ContinuationExpirationOutcome::Expired { return_to: 7, .. }
));
Ok(())
}