#![allow(clippy::missing_const_for_fn)]
use std::sync::Arc;
use chrono::{DateTime, LocalResult, NaiveDateTime, Offset, TimeZone, Utc};
use rand::{Rng, SeedableRng};
use rand_chacha::ChaCha8Rng;
use uuid::Uuid;
use crate::entropy::{Entropy, SeededEntropy, uuid_v4_from_bytes, uuid_v7_from_parts};
use crate::time::TickingClock;
#[cfg(feature = "sqlite")]
#[doc(hidden)]
pub mod substrate;
pub mod chaos;
#[cfg(feature = "mail")]
pub use chaos::MailFault;
pub use chaos::{Chaos, ChaosEvent, ChaosHook};
pub mod fault;
pub use fault::{
FaultEffect, FaultLedger, FaultOutcome, FaultPlan, FinalState, FiredFault, PlannedFault,
ReportedError,
};
pub mod llm;
pub use llm::{LlmCall, LlmClient, LlmError, LlmRequest, LlmResponse, SeededLlm, SeededLlmBuilder};
pub mod crash;
pub mod assert;
#[cfg(feature = "http-client")]
pub mod net;
#[cfg(feature = "http-client")]
pub use net::{NetEvent, NetFault, SimNet};
#[doc(hidden)]
pub mod scenario;
mod shuffle;
pub mod replica;
pub use replica::{Replica, ReplicaClock};
#[cfg(feature = "sqlite")]
pub mod dblink;
#[cfg(feature = "sqlite")]
pub use dblink::{DbFaultEvent, DbFaultKind, DbLink};
pub(crate) mod gate;
pub use gate::{SimRuntime, runtime};
pub mod trace;
pub use trace::{Trace, TraceDiff};
#[cfg(feature = "sqlite")]
#[doc(hidden)]
pub mod fleet;
pub use assert::{
SometimesRegistry, assert_all_sometimes_satisfied, reset_sometimes_registry,
sometimes_snapshot, sometimes_unsatisfied,
};
pub use crash::{CrashOutcome, CrashPoint, CrashSchedule, crash_at};
#[cfg(feature = "sim-testing")]
pub mod op;
#[cfg(feature = "sim-testing")]
pub mod sweep;
#[cfg(feature = "sim-testing")]
pub use sweep::{SweepFailure, SweepOutcome, sweep_proptest};
const SIM_EPOCH_UNIX_SECS: i64 = 1_577_836_800;
#[non_exhaustive]
pub struct Sim {
pub seed: u64,
rng: SimRng,
clock: SimClock,
chaos: Chaos,
chaos_state: Option<Arc<chaos::ChaosState>>,
app: SimApp,
strict_budget: Option<std::time::Duration>,
#[cfg(feature = "http-client")]
net: Option<net::SimNet>,
shuffle_calls: std::sync::atomic::AtomicU64,
ambient: Arc<AmbientSimClock>,
_ambient_guard: crate::time::AmbientGuard,
stack_guard: SimStackGuard,
mounts: u64,
replicas: Vec<ReplicaSlot>,
#[cfg(feature = "sqlite")]
db_links: std::sync::Mutex<std::collections::BTreeMap<String, DbLink>>,
gate_seat: gate::SimSeat,
}
fn replica_host_key(name: &str) -> String {
#[cfg(feature = "http-client")]
{
net::canonical_host(name)
}
#[cfg(not(feature = "http-client"))]
{
name.to_ascii_lowercase()
}
}
struct ReplicaSlot {
name: String,
clock: replica::NodeClock,
client: Option<crate::test::TestClient>,
#[cfg(feature = "http-client")]
net: Option<net::SimNet>,
}
impl Sim {
#[must_use]
pub fn from_seed(seed: u64) -> Self {
assert::reset_sometimes_registry();
let clock = SimClock::new(TickingClock::starting_at(sim_epoch()));
let ambient_clock = Arc::new(AmbientSimClock {
wall: std::sync::RwLock::new(Arc::new(clock.ticking())),
own_advanced: std::sync::Mutex::new(std::time::Duration::ZERO),
auto_advanced: std::sync::Mutex::new(std::time::Duration::ZERO),
alive: std::sync::atomic::AtomicBool::new(true),
});
let gate_seat = gate::SimSeat::default();
gate_seat.take();
let ambient_guard = crate::time::install_ambient(ambient_clock.clone());
let sim_stack = SimStackGuard::enter(Arc::clone(&ambient_clock));
Self {
seed,
rng: SimRng::new(seed),
clock,
ambient: ambient_clock,
_ambient_guard: ambient_guard,
stack_guard: sim_stack,
chaos: Chaos::default(),
chaos_state: None,
app: SimApp::default(),
strict_budget: None,
#[cfg(feature = "http-client")]
net: None,
shuffle_calls: std::sync::atomic::AtomicU64::new(0),
mounts: 0,
replicas: Vec::new(),
#[cfg(feature = "sqlite")]
db_links: std::sync::Mutex::default(),
gate_seat,
}
}
pub fn anchor(&self) {
if tokio::runtime::Handle::try_current().is_ok() {
lock_sim_stack(&self.stack_guard.home, SimStack::settle);
self.gate_seat.take();
}
}
pub fn chaos(&mut self, chaos: Chaos) -> &mut Self {
self.chaos = chaos;
self
}
#[cfg(feature = "http-client")]
pub fn net(&mut self, net: SimNet) -> &mut Self {
self.net = Some(net);
self
}
#[must_use]
pub const fn seed(&self) -> u64 {
self.seed
}
#[must_use]
pub fn rng(&mut self) -> &mut SimRng {
&mut self.rng
}
#[must_use]
pub fn seeded_entropy(&self) -> Arc<dyn Entropy> {
SeededEntropy::shared(self.seed)
}
pub fn build(&mut self, app: crate::test::TestApp) -> &crate::test::TestClient {
self.mount(app)
}
fn mount(&mut self, app: crate::test::TestApp) -> &crate::test::TestClient {
let mount_seed = mount_entropy_seed(self.seed, self.mounts);
let app = app.with_default_entropy(SeededEntropy::shared(mount_seed));
self.mounts += 1;
#[cfg(feature = "http-client")]
let app = match &self.net {
Some(net) => {
net.reseed(mount_seed);
app.with_sim_net(net.clone())
}
None => app,
};
*self
.ambient
.wall
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner) =
chaos::wall_clock(&self.chaos, self.seed, self.clock.ticking());
let app = if self.chaos.is_active() {
let state = chaos::ChaosState::new(self.seed, &self.chaos);
self.chaos_state = Some(Arc::clone(&state));
chaos::install(app, &self.chaos, self.seed, self.clock.ticking(), state)
} else {
app.with_clock(self.clock.ticking())
};
let client = app.build();
self.app.client = Some(client);
self.app.client()
}
pub fn kill(&mut self) {
if let Some(client) = self.app.client.take() {
abort_app_tasks(&client);
drop(client);
}
self.chaos_state = None;
}
pub fn restart(&mut self, app: crate::test::TestApp) -> &crate::test::TestClient {
self.mount(app)
}
pub fn crash_and_restart(&mut self, app: crate::test::TestApp) -> &crate::test::TestClient {
self.kill();
self.restart(app)
}
pub fn mount_replica(
&mut self,
replica: impl Into<Replica>,
app: crate::test::TestApp,
) -> &crate::test::TestClient {
let replica = replica.into();
let key = replica_host_key(replica.name());
if let Some(slot) = self
.replicas
.iter()
.find(|slot| replica_host_key(&slot.name) == key)
{
panic!(
"replica `{}` is already mounted as `{}`; use `restart_replica` to mount it again",
replica.name(),
slot.name
);
}
let clock = replica::NodeClock::new(
Arc::clone(&self.ambient) as Arc<dyn crate::time::ClockSource>,
sim_epoch(),
replica.resolve(self.seed),
);
self.replicas.push(ReplicaSlot {
name: replica.name().to_owned(),
clock,
client: None,
#[cfg(feature = "http-client")]
net: None,
});
let index = self.replicas.len() - 1;
self.mount_replica_at(index, app)
}
pub fn restart_replica(
&mut self,
name: &str,
app: crate::test::TestApp,
) -> &crate::test::TestClient {
self.kill_replica(name);
let index = self.expect_replica_index(name);
self.mount_replica_at(index, app)
}
pub fn kill_replica(&mut self, name: &str) {
let index = self.expect_replica_index(name);
let Some(client) = self.replicas[index].client.take() else {
return;
};
#[cfg(feature = "http-client")]
if let Some(net) = self.replicas[index].net.take() {
net.remove_host(name);
}
abort_app_tasks(&client);
drop(client);
}
#[must_use]
pub fn replica(&self, name: &str) -> &crate::test::TestClient {
self.try_replica(name)
.unwrap_or_else(|| panic!("no live replica `{name}`: mount it with `mount_replica`"))
}
#[must_use]
pub fn try_replica(&self, name: &str) -> Option<&crate::test::TestClient> {
self.replica_index(name)
.and_then(|index| self.replicas[index].client.as_ref())
}
#[must_use]
pub fn replica_names(&self) -> Vec<&str> {
self.replicas
.iter()
.filter(|slot| slot.client.is_some())
.map(|slot| slot.name.as_str())
.collect()
}
#[must_use]
pub fn replica_clock(&self, name: &str) -> Option<ReplicaClock> {
self.replica_index(name)
.map(|index| self.replicas[index].clock.settings())
}
pub fn step_replica_clock(&self, name: &str, by: chrono::TimeDelta) {
let index = self.expect_replica_index(name);
self.replicas[index].clock.step(by);
}
pub async fn run_for(&self, duration: std::time::Duration) {
let guard_start = self.wall_clock_guard_start();
let start = tokio::time::Instant::now();
tokio::time::sleep(duration).await;
let moved = tokio::time::Instant::now().saturating_duration_since(start);
self.clock.advance(moved);
self.enforce_wall_clock_budget(guard_start);
}
#[cfg(feature = "sqlite")]
#[must_use]
pub fn db_link(&self, name: &str) -> DbLink {
self.db_links
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.entry(name.to_owned())
.or_insert_with(|| {
DbLink::new(name, derive_seed(self.seed, &format!("sim-db-link:{name}")))
})
.clone()
}
fn live_clients(&self) -> impl Iterator<Item = &crate::test::TestClient> {
self.app
.try_client()
.into_iter()
.chain(self.replicas.iter().filter_map(|slot| slot.client.as_ref()))
}
fn replica_index(&self, name: &str) -> Option<usize> {
self.replicas.iter().position(|slot| slot.name == name)
}
fn expect_replica_index(&self, name: &str) -> usize {
self.replica_index(name)
.unwrap_or_else(|| panic!("no replica `{name}` was mounted"))
}
fn mount_replica_at(
&mut self,
index: usize,
app: crate::test::TestApp,
) -> &crate::test::TestClient {
let mount_seed = mount_entropy_seed(self.seed, self.mounts);
let first_mount = self.mounts == 0;
let app = app.with_default_entropy(SeededEntropy::shared(mount_seed));
self.mounts += 1;
#[cfg(feature = "http-client")]
let app = match &self.net {
Some(net) => {
if first_mount {
net.reseed(mount_seed);
}
app.with_sim_net(net.clone())
}
None => app,
};
#[cfg(not(feature = "http-client"))]
let _ = first_mount;
let slot = &mut self.replicas[index];
#[cfg(feature = "http-client")]
let app = if self.net.is_some() {
app.trust_host(&slot.name)
} else {
app
};
let client = app.with_clock(slot.clock.clone()).build();
#[cfg(feature = "http-client")]
if let Some(net) = &self.net {
net.insert_host(&slot.name, client.router());
slot.net = Some(net.clone());
}
slot.client.insert(client)
}
pub async fn interleave<F: std::future::Future>(&self, ops: Vec<F>) -> Vec<F::Output> {
shuffle::Interleave::new(ops, self.next_shuffle_stream()).await
}
pub fn spawn<F>(&self, op: F) -> tokio::task::JoinHandle<F::Output>
where
F: std::future::Future + Send + 'static,
F::Output: Send + 'static,
{
tokio::spawn(shuffle::Shuffled::new(op, self.next_shuffle_stream()))
}
fn next_shuffle_stream(&self) -> Arc<dyn Entropy> {
let call = self
.shuffle_calls
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
shuffle::stream(self.seed, call)
}
#[must_use]
pub fn crash_schedule(&self) -> CrashSchedule {
CrashSchedule::derive(self.seed, crash::DEFAULT_CRASH_SCHEDULE_LEN)
}
#[must_use]
pub fn crash_point(&self) -> Option<CrashPoint> {
self.crash_schedule().first().cloned()
}
#[doc(hidden)]
#[must_use]
pub fn __chaos_events(&self) -> Vec<ChaosEvent> {
self.chaos_state
.as_ref()
.map(|state| state.events())
.unwrap_or_default()
}
#[must_use]
pub fn client(&self) -> &crate::test::TestClient {
self.app.client()
}
#[must_use]
pub const fn try_client(&self) -> Option<&crate::test::TestClient> {
self.app.try_client()
}
pub fn strict_wall_clock(&mut self) -> &mut Self {
self.strict_budget = Some(strict_budget_from_env_or(DEFAULT_STRICT_WALL_CLOCK_BUDGET));
self
}
pub fn strict_wall_clock_budget(&mut self, budget: std::time::Duration) -> &mut Self {
self.strict_budget = Some(strict_budget_from_env_or(budget));
self
}
fn wall_clock_guard_start(&self) -> Option<std::time::Instant> {
self.strict_budget.map(|_| std::time::Instant::now())
}
fn enforce_wall_clock_budget(&self, start: Option<std::time::Instant>) {
if let (Some(budget), Some(start)) = (self.strict_budget, start) {
let elapsed = start.elapsed();
assert!(
elapsed <= budget,
"Sim::strict_wall_clock real-time leak guard tripped: {elapsed:?} of real \
wall-clock time elapsed inside a paused-sim step, exceeding the {budget:?} \
budget. This means real time leaked into the virtual timeline — usually a real \
`std::thread::sleep`, blocking I/O, or `spawn_blocking` that escaped tokio's \
paused timer. If this is CI scheduling jitter rather than a genuine leak, raise \
the budget via the AUTUMN_SIM_STRICT_WALL_CLOCK_BUDGET_MS environment variable."
);
}
}
pub async fn advance(&self, duration: std::time::Duration) {
let guard_start = self.wall_clock_guard_start();
let _open = gate::Open::new();
tokio::task::yield_now().await;
self.clock.advance(duration);
note_own_advance(&self.ambient, duration);
tokio::time::advance(duration).await;
self.enforce_wall_clock_budget(guard_start);
}
pub fn advance_to<Tz>(
&self,
target: &DateTime<Tz>,
) -> impl std::future::Future<Output = ()> + '_
where
Tz: TimeZone,
{
let target_utc = target.with_timezone(&Utc);
self.advance_to_utc(target_utc)
}
pub fn advance_to_local<Tz>(
&self,
local: NaiveDateTime,
tz: &Tz,
) -> impl std::future::Future<Output = ()> + '_
where
Tz: TimeZone,
{
self.advance_to_utc(resolve_local_to_utc(local, tz))
}
async fn advance_to_utc(&self, target: DateTime<Utc>) {
match plan_advance_to(self.clock.now(), target) {
AdvancePlan::NoOp => {}
AdvancePlan::Advance(delta) => self.advance(delta).await,
}
}
pub async fn run_to_idle(&self) {
if let Err(stall) = self.try_run_to_idle().await {
panic!("{stall}");
}
}
pub async fn try_run_to_idle(&self) -> Result<(), SimStall> {
let guard_start = self.wall_clock_guard_start();
let _open = gate::Open::new();
#[cfg(feature = "db")]
let commit_hook_pools: Vec<_> = self
.live_clients()
.filter_map(|client| crate::db::DbState::pool(client.state()).cloned())
.collect();
let mut quiet_rounds = 0;
let mut rounds = 0;
while rounds < MAX_DRAIN_STEPS
|| (quiet_rounds < STALL_WINDOW && rounds < MAX_DRAIN_STEPS * 2)
{
rounds += 1;
let before = drain_fingerprint();
tokio::task::yield_now().await;
tokio::time::advance(std::time::Duration::ZERO).await;
#[cfg(feature = "db")]
for pool in &commit_hook_pools {
let hooks_drained =
crate::test::drain_ready_repository_commit_hooks(pool, MAX_DRAIN_STEPS).await;
if hooks_drained > 0 {
note_drain_progress();
}
}
#[cfg(feature = "db")]
for state in self
.live_clients()
.map(crate::test::TestClient::state)
.filter(|state| state.extension::<crate::outbox::OutboxRelay>().is_some())
{
match crate::outbox::drain(state, MAX_DRAIN_STEPS).await {
Ok(0) => {}
Ok(_) => note_drain_progress(),
Err(error) => tracing::warn!(%error, "sim outbox drain failed"),
}
}
if drain_fingerprint() == before {
quiet_rounds += 1;
} else {
quiet_rounds = 0;
}
}
self.enforce_wall_clock_budget(guard_start);
if quiet_rounds < STALL_WINDOW {
return Err(SimStall {
seed: self.seed,
steps: rounds,
});
}
Ok(())
}
}
impl Drop for Sim {
fn drop(&mut self) {
#[cfg(feature = "http-client")]
for slot in &mut self.replicas {
if let Some(net) = slot.net.take() {
net.remove_host(&slot.name);
}
}
self.gate_seat.leave();
}
}
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SimStall {
pub seed: u64,
pub steps: usize,
}
impl std::fmt::Display for SimStall {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"sim drain stall: work still ran after {steps} drain rounds (seed=0x{seed:x}). \
A job, hook or task keeps making new work, so the app is never idle. \
Replay with AUTUMN_SIM_SEED=0x{seed:x}.",
steps = self.steps,
seed = self.seed,
)
}
}
impl std::error::Error for SimStall {}
thread_local! {
static DRAIN_PROGRESS: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}
pub(crate) fn note_drain_progress() {
DRAIN_PROGRESS.with(|progress| progress.set(progress.get().wrapping_add(1)));
}
fn drain_fingerprint() -> (u64, usize) {
let tasks = tokio::runtime::Handle::try_current()
.map_or(0, |handle| handle.metrics().num_alive_tasks());
(DRAIN_PROGRESS.with(std::cell::Cell::get), tasks)
}
fn sim_epoch() -> DateTime<Utc> {
Utc.timestamp_opt(SIM_EPOCH_UNIX_SECS, 0)
.single()
.unwrap_or_else(|| Utc.timestamp_nanos(0))
}
#[cfg(feature = "sqlite")]
fn derive_seed(seed: u64, purpose: &str) -> u64 {
SeededEntropy::new(seed)
.derive_uuid(purpose)
.as_u64_pair()
.0
}
#[derive(Default)]
pub(crate) struct AppTasks {
handles: std::sync::Mutex<Vec<tokio::task::AbortHandle>>,
}
impl AppTasks {
fn track(&self, handle: tokio::task::AbortHandle) {
let mut handles = self
.handles
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
handles.retain(|handle| !handle.is_finished());
handles.push(handle);
}
fn abort_all(&self) {
let handles = std::mem::take(
&mut *self
.handles
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner),
);
for handle in handles {
handle.abort();
}
}
}
pub(crate) fn spawn_app_task<F>(
state: &crate::state::AppState,
future: F,
) -> tokio::task::JoinHandle<F::Output>
where
F: std::future::Future + Send + 'static,
F::Output: Send + 'static,
{
let handle = tokio::spawn(future);
if let Some(tasks) = state.extension::<AppTasks>() {
tasks.track(handle.abort_handle());
}
handle
}
fn abort_app_tasks(client: &crate::test::TestClient) {
if let Some(tasks) = client.state().extension::<AppTasks>() {
tasks.abort_all();
}
}
fn mount_entropy_seed(seed: u64, mount: u64) -> u64 {
if mount == 0 {
return seed;
}
let derived = SeededEntropy::new(seed).derive_uuid(format!("sim-mount-{mount}"));
let bytes: [u8; 8] = derived.as_bytes()[..8]
.try_into()
.expect("a uuid has 16 bytes");
u64::from_le_bytes(bytes)
}
const MAX_LIVENESS_BUDGET: std::time::Duration = std::time::Duration::from_millis(68_719_476_734);
fn parse_liveness_budget(raw: Option<&str>) -> Option<std::time::Duration> {
let secs = raw.map(str::trim)?.parse::<u64>().ok()?;
if secs == 0 {
return None;
}
Some(std::time::Duration::from_secs(secs).min(MAX_LIVENESS_BUDGET))
}
#[doc(hidden)]
pub async fn __with_liveness_watchdog<F: std::future::Future>(seed: u64, body: F) -> F::Output {
anchor_current_sim();
let raw = std::env::var("AUTUMN_SIM_LIVENESS_BUDGET_SECS").ok();
__with_liveness_budget(seed, parse_liveness_budget(raw.as_deref()), body).await
}
#[doc(hidden)]
pub async fn __with_liveness_budget<F: std::future::Future>(
seed: u64,
budget: Option<std::time::Duration>,
body: F,
) -> F::Output {
let Some(budget) = budget else {
return body.await;
};
tokio::time::timeout(budget, body)
.await
.unwrap_or_else(|_elapsed| {
panic!(
"sim liveness: the test body did not finish within {budget:?} of virtual time \
(seed=0x{seed:x}). Every task was parked with no timer to wake one, which is \
a deadlock, or the body waited past the budget. Raise \
AUTUMN_SIM_LIVENESS_BUDGET_SECS for a legitimately long run."
)
})
}
const MAX_DRAIN_STEPS: usize = 1024;
const STALL_WINDOW: usize = 64;
const DEFAULT_STRICT_WALL_CLOCK_BUDGET: std::time::Duration = std::time::Duration::from_secs(2);
fn strict_budget_from_env_or(default: std::time::Duration) -> std::time::Duration {
std::env::var("AUTUMN_SIM_STRICT_WALL_CLOCK_BUDGET_MS")
.ok()
.and_then(|raw| parse_strict_budget_ms(&raw))
.unwrap_or(default)
}
fn parse_strict_budget_ms(raw: &str) -> Option<std::time::Duration> {
let trimmed = raw.trim();
if trimmed.is_empty() {
return None;
}
trimmed
.parse::<u64>()
.ok()
.map(std::time::Duration::from_millis)
}
#[derive(Debug, PartialEq, Eq)]
enum AdvancePlan {
NoOp,
Advance(std::time::Duration),
}
fn plan_advance_to(now: DateTime<Utc>, target: DateTime<Utc>) -> AdvancePlan {
let delta = target - now;
match delta.cmp(&chrono::Duration::zero()) {
std::cmp::Ordering::Equal => AdvancePlan::NoOp,
std::cmp::Ordering::Less => panic!(
"Sim::advance_to target {target} is strictly before the current sim instant {now}; \
virtual time is forward-only (advancing to a past instant is a test bug)"
),
std::cmp::Ordering::Greater => AdvancePlan::Advance(
delta
.to_std()
.expect("a strictly-positive chrono delta always converts to std::time::Duration"),
),
}
}
fn resolve_local_to_utc<Tz>(local: NaiveDateTime, tz: &Tz) -> DateTime<Utc>
where
Tz: TimeZone,
{
match tz.from_local_datetime(&local) {
LocalResult::Single(dt) => dt.with_timezone(&Utc),
LocalResult::Ambiguous(earlier, _later) => earlier.with_timezone(&Utc),
LocalResult::None => {
let offset_secs =
i64::from(tz.offset_from_utc_datetime(&local).fix().local_minus_utc());
(local - chrono::Duration::seconds(offset_secs)).and_utc()
}
}
}
pub struct SimRng {
seed: u64,
inner: ChaCha8Rng,
}
impl SimRng {
pub(crate) fn new(seed: u64) -> Self {
Self {
seed,
inner: ChaCha8Rng::seed_from_u64(seed),
}
}
#[must_use]
pub fn derive_uuid(&self, purpose_tag: impl AsRef<[u8]>) -> Uuid {
crate::entropy::derive_uuid_from(self.seed, purpose_tag.as_ref())
}
pub fn next_u64(&mut self) -> u64 {
self.inner.next_u64()
}
pub fn fill_bytes(&mut self, dest: &mut [u8]) {
self.inner.fill_bytes(dest);
}
#[must_use]
pub fn uuid_v4(&mut self) -> Uuid {
let mut bytes = [0u8; 16];
self.inner.fill_bytes(&mut bytes);
uuid_v4_from_bytes(bytes)
}
#[must_use]
pub fn uuid_v7(&mut self, unix_millis: u64) -> Uuid {
let mut rand_bytes = [0u8; 10];
self.inner.fill_bytes(&mut rand_bytes);
uuid_v7_from_parts(unix_millis, rand_bytes)
}
#[cfg(test)]
pub(crate) fn inner_mut(&mut self) -> &mut ChaCha8Rng {
&mut self.inner
}
}
pub(crate) struct SimClock {
inner: TickingClock,
}
impl SimClock {
pub(crate) fn new(inner: TickingClock) -> Self {
Self { inner }
}
pub(crate) fn advance(&self, duration: std::time::Duration) {
self.inner.advance(duration);
}
pub(crate) fn ticking(&self) -> TickingClock {
self.inner.clone()
}
pub(crate) fn now(&self) -> DateTime<Utc> {
crate::time::ClockSource::now(&self.inner)
}
}
struct AmbientSimClock {
wall: std::sync::RwLock<Arc<dyn crate::time::ClockSource>>,
own_advanced: std::sync::Mutex<std::time::Duration>,
auto_advanced: std::sync::Mutex<std::time::Duration>,
alive: std::sync::atomic::AtomicBool,
}
impl AmbientSimClock {
fn add(slot: &std::sync::Mutex<std::time::Duration>, duration: std::time::Duration) {
let mut total = slot
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
*total = total.saturating_add(duration);
}
fn read(slot: &std::sync::Mutex<std::time::Duration>) -> std::time::Duration {
*slot
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn is_alive(&self) -> bool {
self.alive.load(std::sync::atomic::Ordering::Acquire)
}
}
#[derive(Default)]
struct SimStack {
clocks: Vec<Arc<AmbientSimClock>>,
checkpoint: Option<tokio::time::Instant>,
warned_unanchored: bool,
pending_advance: std::time::Duration,
runtime: Option<tokio::runtime::Handle>,
}
impl SimStack {
fn settle_lazily(&mut self) {
if self.checkpoint.is_none()
&& !self.warned_unanchored
&& self.clocks.iter().any(|clock| clock.is_alive())
&& tokio::runtime::Handle::try_current().is_ok()
{
self.warned_unanchored = true;
tracing::warn!(
"a Sim built outside its runtime was not anchored; its elapsed \
time starts now. Call `Sim::anchor` first in the runtime, or use \
#[sim_test]"
);
}
self.settle();
}
fn settle(&mut self) {
if let Ok(current) = tokio::runtime::Handle::try_current() {
self.runtime = Some(current);
}
let Some(runtime) = self.runtime.clone() else {
self.clocks.retain(|clock| clock.is_alive());
return;
};
let now = {
let _enter = runtime.enter();
tokio::time::Instant::now()
};
if let Some(checkpoint) = self.checkpoint {
let moved = now.saturating_duration_since(checkpoint);
let explicit = moved.min(self.pending_advance);
self.pending_advance = self.pending_advance.saturating_sub(explicit);
if let Some(top) = self.clocks.last() {
AmbientSimClock::add(&top.auto_advanced, moved.saturating_sub(explicit));
}
}
self.clocks.retain(|clock| clock.is_alive());
if self.clocks.is_empty() {
self.checkpoint = None;
self.pending_advance = std::time::Duration::ZERO;
self.runtime = None;
} else {
self.checkpoint = Some(now);
}
}
}
type SharedSimStack = Arc<std::sync::Mutex<SimStack>>;
thread_local! {
static SIM_STACK: SharedSimStack = SharedSimStack::default();
}
fn lock_sim_stack<T>(stack: &SharedSimStack, f: impl FnOnce(&mut SimStack) -> T) -> T {
f(&mut stack
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner))
}
fn with_sim_stack<T>(f: impl FnOnce(&mut SimStack) -> T) -> Option<T> {
SIM_STACK.try_with(|stack| lock_sim_stack(stack, f)).ok()
}
fn anchor_current_sim() {
if tokio::runtime::Handle::try_current().is_err() {
return;
}
with_sim_stack(SimStack::settle);
}
struct SimStackGuard {
own: Arc<AmbientSimClock>,
home: SharedSimStack,
}
impl SimStackGuard {
fn enter(own: Arc<AmbientSimClock>) -> Self {
let home = SIM_STACK.with(Arc::clone);
lock_sim_stack(&home, |stack| {
stack.settle();
stack.clocks.push(Arc::clone(&own));
if stack.checkpoint.is_none() && tokio::runtime::Handle::try_current().is_ok() {
stack.checkpoint = Some(tokio::time::Instant::now());
}
});
Self { own, home }
}
}
impl Drop for SimStackGuard {
fn drop(&mut self) {
lock_sim_stack(&self.home, |stack| {
stack.settle();
self.own
.alive
.store(false, std::sync::atomic::Ordering::Release);
stack.settle();
});
}
}
fn note_own_advance(sim: &AmbientSimClock, duration: std::time::Duration) {
with_sim_stack(|stack| {
stack.settle_lazily();
stack.pending_advance = stack.pending_advance.saturating_add(duration);
});
AmbientSimClock::add(&sim.own_advanced, duration);
}
impl crate::time::ClockSource for AmbientSimClock {
fn now(&self) -> DateTime<Utc> {
self.wall
.read()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.now()
}
fn monotonic(&self) -> crate::time::MonotonicInstant {
if tokio::runtime::Handle::try_current().is_ok() {
with_sim_stack(SimStack::settle_lazily);
}
crate::time::MonotonicInstant::from_origin_elapsed(
Self::read(&self.own_advanced).saturating_add(Self::read(&self.auto_advanced)),
)
}
}
#[derive(Default)]
pub(crate) struct SimApp {
client: Option<crate::test::TestClient>,
}
impl SimApp {
#[must_use]
pub fn client(&self) -> &crate::test::TestClient {
self.try_client()
.expect("no app mounted: call `sim.build(TestApp::new()...)` before `client()`")
}
#[must_use]
pub const fn try_client(&self) -> Option<&crate::test::TestClient> {
self.client.as_ref()
}
}
#[doc(hidden)]
#[must_use]
pub fn __seed_from_env() -> u64 {
std::env::var("AUTUMN_SIM_SEED").map_or(0, |raw| parse_seed(&raw))
}
fn parse_seed(raw: &str) -> u64 {
let trimmed = raw.trim();
trimmed
.strip_prefix("0x")
.or_else(|| trimmed.strip_prefix("0X"))
.map_or_else(
|| trimmed.parse::<u64>().unwrap_or(0),
|hex| u64::from_str_radix(hex, 16).unwrap_or(0),
)
}
#[doc(hidden)]
#[must_use]
pub fn __replay_line(seed: u64, pkg: &str, test: &str) -> String {
format!("AUTUMN_SIM_SEED=0x{seed:x} cargo test -p {pkg} {test}")
}
#[cfg(test)]
mod tests {
use super::{
__replay_line, AdvancePlan, DEFAULT_STRICT_WALL_CLOCK_BUDGET, MAX_LIVENESS_BUDGET, Sim,
mount_entropy_seed, parse_liveness_budget, parse_seed, parse_strict_budget_ms,
plan_advance_to, resolve_local_to_utc, strict_budget_from_env_or,
};
use chrono::{NaiveDate, TimeZone, Utc};
use rand::Rng;
#[tokio::test(start_paused = true)]
async fn blocking_work_reads_auto_advanced_sim_time() {
let sim = Sim::from_seed(11);
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
sim.advance(std::time::Duration::from_secs(2)).await;
let here = (
crate::time::ambient_monotonic(),
crate::time::ambient_instant(),
crate::time::ambient_now(),
);
assert_eq!(here.0.since_origin(), std::time::Duration::from_secs(7));
let there = crate::time::spawn_blocking(|| {
(
crate::time::ambient_monotonic(),
crate::time::ambient_instant(),
crate::time::ambient_now(),
)
})
.await
.unwrap();
assert_eq!(there, here, "a blocking worker reads the caller's sim time");
}
#[tokio::test(start_paused = true)]
async fn a_thread_with_no_runtime_reads_auto_advanced_sim_time() {
let sim = Sim::from_seed(12);
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
let here = crate::time::ambient_monotonic();
assert_eq!(here.since_origin(), std::time::Duration::from_secs(5));
let clock: std::sync::Arc<dyn crate::time::ClockSource> = sim.ambient.clone();
let there = std::thread::spawn(move || clock.monotonic())
.join()
.unwrap();
assert_eq!(there, here);
}
#[test]
fn liveness_budget_is_off_unless_armed() {
assert_eq!(parse_liveness_budget(None), None);
assert_eq!(parse_liveness_budget(Some("")), None);
assert_eq!(parse_liveness_budget(Some("0")), None);
assert_eq!(parse_liveness_budget(Some("soon")), None);
assert_eq!(parse_liveness_budget(Some("-5")), None);
}
#[test]
fn liveness_budget_takes_seconds_and_clamps_to_the_tokio_maximum() {
assert_eq!(
parse_liveness_budget(Some(" 90 ")),
Some(std::time::Duration::from_secs(90))
);
assert_eq!(
parse_liveness_budget(Some("18446744073709551615")),
Some(MAX_LIVENESS_BUDGET)
);
}
#[test]
fn mount_entropy_seed_is_the_sim_seed_first_then_derived() {
assert_eq!(mount_entropy_seed(7, 0), 7);
let restart = mount_entropy_seed(7, 1);
assert_ne!(restart, 7, "a restart draws a new stream");
assert_eq!(restart, mount_entropy_seed(7, 1), "and it is deterministic");
assert_ne!(restart, mount_entropy_seed(7, 2));
assert_ne!(restart, mount_entropy_seed(8, 1));
}
#[test]
fn replay_line_zero_seed_is_exact() {
assert_eq!(
__replay_line(0, "autumn-web", "my_test"),
"AUTUMN_SIM_SEED=0x0 cargo test -p autumn-web my_test"
);
}
#[test]
fn replay_line_formats_seed_as_hex() {
let line = __replay_line(0x9f3a, "autumn-web", "my_test");
assert!(
line.contains("0x9f3a"),
"seed must be rendered in hex: {line}"
);
assert_eq!(
line,
"AUTUMN_SIM_SEED=0x9f3a cargo test -p autumn-web my_test"
);
}
#[test]
fn parse_seed_covers_hex_decimal_and_garbage() {
assert_eq!(parse_seed("0"), 0);
assert_eq!(parse_seed("0x9f3a"), 0x9f3a);
assert_eq!(parse_seed("0X9F3A"), 0x9f3a);
assert_eq!(parse_seed("42"), 42);
assert_eq!(parse_seed("garbage"), 0);
assert_eq!(parse_seed(""), 0);
assert_eq!(parse_seed(" 0x10 "), 0x10);
}
#[test]
fn from_seed_exposes_the_seed() {
assert_eq!(Sim::from_seed(0).seed, 0);
assert_eq!(Sim::from_seed(7).seed(), 7);
}
#[test]
fn same_seed_produces_identical_first_draw() {
let mut a = Sim::from_seed(7);
let mut b = Sim::from_seed(7);
let da = a.rng().inner_mut().next_u64();
let db = b.rng().inner_mut().next_u64();
assert_eq!(da, db, "same seed must yield the same first RNG draw");
}
#[test]
fn plan_advance_to_equal_target_is_noop() {
let now = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
assert_eq!(plan_advance_to(now, now), AdvancePlan::NoOp);
}
#[test]
fn plan_advance_to_future_target_is_exact_delta() {
let now = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
let target = Utc.with_ymd_and_hms(2020, 1, 1, 1, 0, 0).unwrap();
assert_eq!(
plan_advance_to(now, target),
AdvancePlan::Advance(std::time::Duration::from_secs(3600))
);
}
#[test]
#[should_panic(expected = "forward-only")]
fn plan_advance_to_past_target_panics() {
let now = Utc.with_ymd_and_hms(2020, 1, 1, 1, 0, 0).unwrap();
let target = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
let _ = plan_advance_to(now, target);
}
#[test]
fn parse_strict_budget_ms_covers_valid_and_garbage() {
use std::time::Duration;
assert_eq!(
parse_strict_budget_ms("100"),
Some(Duration::from_millis(100))
);
assert_eq!(
parse_strict_budget_ms(" 250 "),
Some(Duration::from_millis(250))
);
assert_eq!(parse_strict_budget_ms("0"), Some(Duration::ZERO));
assert_eq!(parse_strict_budget_ms(""), None);
assert_eq!(parse_strict_budget_ms(" "), None);
assert_eq!(parse_strict_budget_ms("garbage"), None);
assert_eq!(parse_strict_budget_ms("-5"), None);
assert_eq!(parse_strict_budget_ms("1.5"), None);
}
#[test]
fn strict_budget_from_env_falls_back_to_default_when_unset() {
let default = std::time::Duration::from_millis(7);
assert_eq!(strict_budget_from_env_or(default), default);
assert_eq!(
strict_budget_from_env_or(DEFAULT_STRICT_WALL_CLOCK_BUDGET),
DEFAULT_STRICT_WALL_CLOCK_BUDGET
);
}
#[test]
fn strict_wall_clock_builders_set_the_budget() {
let mut sim = Sim::from_seed(0);
assert!(sim.strict_budget.is_none(), "guard is off by default");
sim.strict_wall_clock();
assert_eq!(sim.strict_budget, Some(DEFAULT_STRICT_WALL_CLOCK_BUDGET));
let mut custom = Sim::from_seed(0);
custom.strict_wall_clock_budget(std::time::Duration::from_millis(5));
assert_eq!(
custom.strict_budget,
Some(std::time::Duration::from_millis(5))
);
}
#[test]
fn resolve_local_unambiguous_maps_to_single_instant() {
let local = NaiveDate::from_ymd_opt(2020, 6, 1)
.unwrap()
.and_hms_opt(12, 0, 0)
.unwrap();
let got = resolve_local_to_utc(local, &Utc);
assert_eq!(got, Utc.with_ymd_and_hms(2020, 6, 1, 12, 0, 0).unwrap());
}
#[cfg(feature = "http-client")]
#[tokio::test(start_paused = true)]
async fn a_dropped_sim_frees_its_replica_apps() {
let marker = std::sync::Arc::new(());
let freed = std::sync::Arc::downgrade(&marker);
{
let mut sim = Sim::from_seed(21);
sim.net(super::net::SimNet::new());
let held = std::sync::Arc::clone(&marker);
let router = axum::Router::new().route(
"/held",
axum::routing::get(move || {
let holders = std::sync::Arc::strong_count(&held);
async move { holders.to_string() }
}),
);
sim.mount_replica("a", crate::test::TestApp::new().merge(router));
sim.mount_replica("b", crate::test::TestApp::new());
}
drop(marker);
assert!(
freed.upgrade().is_none(),
"the replica app is still alive after its sim dropped"
);
}
#[cfg(feature = "http-client")]
#[tokio::test(start_paused = true)]
async fn a_replica_is_cleaned_from_the_network_it_joined() {
let marker = std::sync::Arc::new(());
let freed = std::sync::Arc::downgrade(&marker);
let first = super::net::SimNet::new();
let held = std::sync::Arc::clone(&marker);
let router = axum::Router::new().route(
"/held",
axum::routing::get(move || {
let holders = std::sync::Arc::strong_count(&held);
async move { holders.to_string() }
}),
);
let second = super::net::SimNet::new().host("a", axum::Router::new());
{
let mut sim = Sim::from_seed(22);
sim.net(first.clone());
sim.mount_replica("a", crate::test::TestApp::new().merge(router));
sim.net(second.clone());
}
drop(marker);
assert!(
freed.upgrade().is_none(),
"the replica app is still alive in the network it joined"
);
assert!(first.service("a").is_none());
assert!(
second.service("a").is_some(),
"a host of another network stays"
);
}
}