Skip to content

test(supervisor): give the poison window and restart backoff a clock seam #434

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lgahdl wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

test(supervisor): give the poison window and restart backoff a clock seam #434

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions crates/nexum-runtime/src/bootstrap.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ pub async fn run<T: RuntimeTypes>(
wasm,
manifest,
clocks: None,
supervisor_clock: None,
};
let ctx = LaunchContext {
tasks: TaskManager::new(),
Expand Down
41 changes: 40 additions & 1 deletion crates/nexum-runtime/src/builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
use std::future::{Future, IntoFuture};
use std::marker::PhantomData;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use nexum_tasks::{DrainOutcome, TaskExit, TaskHandle, TaskManager, TaskSet};
Expand All @@ -33,8 +34,8 @@ use crate::host::extension::Extension;
use crate::host::logs::LogPipeline;
use crate::preset::Runtime;
use crate::runtime::event_loop;
pub use crate::supervisor::WasiClockOverride;
use crate::supervisor::{self, Supervisor};
pub use crate::supervisor::{SupervisorClock, WasiClockOverride};

/// Ambient inputs the imperative launcher reads: the task manager every
/// runtime task spawns through, and the loaded config.
Expand Down Expand Up @@ -137,6 +138,10 @@ pub struct AssembledRuntime<'a, T: RuntimeTypes> {
pub manifest: Option<&'a Path>,
/// Per-store WASI clock override; `None` leaves the ambient host clocks.
pub clocks: Option<WasiClockOverride>,
/// Clock the supervisor's poison window and restart backoff read
/// `now()` from; `None` uses the real host clock. Distinct from
/// `clocks`, which only virtualizes guest-visible WASI time.
pub supervisor_clock: Option<Arc<dyn SupervisorClock>>,
}

/// An assembled runtime launchable from a [`LaunchContext`].
Expand All @@ -154,6 +159,7 @@ impl<T: RuntimeTypes> LaunchRuntime for AssembledRuntime<'_, T> {
wasm,
manifest,
clocks,
supervisor_clock,
} = self;
let LaunchContext {
tasks,
Expand Down Expand Up @@ -196,6 +202,7 @@ impl<T: RuntimeTypes> LaunchRuntime for AssembledRuntime<'_, T> {
&engine_cfg.limits,
&extensions,
clocks,
supervisor_clock,
)
.await?
} else if !engine_cfg.modules.is_empty() {
Expand All @@ -206,6 +213,7 @@ impl<T: RuntimeTypes> LaunchRuntime for AssembledRuntime<'_, T> {
&components,
&extensions,
clocks,
supervisor_clock,
)
.await?
} else {
Expand Down Expand Up @@ -347,6 +355,7 @@ impl<'a> RuntimeBuilder<'a> {
wasm: None,
manifest: None,
clocks: None,
supervisor_clock: None,
_t: PhantomData,
}
}
Expand All @@ -361,6 +370,7 @@ impl<'a> RuntimeBuilder<'a> {
wasm: None,
manifest: None,
clocks: None,
supervisor_clock: None,
_r: PhantomData,
}
}
Expand All @@ -375,6 +385,7 @@ pub struct PresetBuilder<'a, R: Runtime> {
wasm: Option<PathBuf>,
manifest: Option<PathBuf>,
clocks: Option<WasiClockOverride>,
supervisor_clock: Option<Arc<dyn SupervisorClock>>,
_r: PhantomData<fn() -> R>,
}

Expand Down Expand Up @@ -405,6 +416,16 @@ impl<'a, R: Runtime> PresetBuilder<'a, R> {
self
}

/// Override the clock the supervisor's poison window and restart
/// backoff read `now()` from. Distinct from
/// [`with_wasi_clocks`](Self::with_wasi_clocks), which only
/// virtualizes guest-visible WASI time. Omitting it uses the real
/// host clock, which is behaviour-neutral.
pub fn with_supervisor_clock(mut self, clock: Arc<dyn SupervisorClock>) -> Self {
self.supervisor_clock = Some(clock);
self
}

/// Open the preset's backends and launch. Builds the [`Components`] bundle
/// from the preset's component builders, installs the preset's add-ons,
/// then drives [`LaunchRuntime::launch`] with a fresh [`TaskManager`].
Expand All @@ -431,6 +452,7 @@ impl<'a, R: Runtime> PresetBuilder<'a, R> {
wasm: self.wasm.as_deref(),
manifest: self.manifest.as_deref(),
clocks: self.clocks,
supervisor_clock: self.supervisor_clock,
};
let ctx = LaunchContext {
tasks,
Expand All @@ -448,6 +470,7 @@ pub struct TypedBuilder<'a, T: RuntimeTypes> {
wasm: Option<PathBuf>,
manifest: Option<PathBuf>,
clocks: Option<WasiClockOverride>,
supervisor_clock: Option<Arc<dyn SupervisorClock>>,
_t: PhantomData<fn() -> T>,
}

Expand All @@ -474,6 +497,16 @@ impl<'a, T: RuntimeTypes> TypedBuilder<'a, T> {
self
}

/// Override the clock the supervisor's poison window and restart
/// backoff read `now()` from. Distinct from
/// [`with_wasi_clocks`](Self::with_wasi_clocks), which only
/// virtualizes guest-visible WASI time. Omitting it uses the real
/// host clock, which is behaviour-neutral.
pub fn with_supervisor_clock(mut self, clock: Arc<dyn SupervisorClock>) -> Self {
self.supervisor_clock = Some(clock);
self
}

/// Bind the component builders that open the backends at launch.
pub fn with_components<C, S, E>(
self,
Expand All @@ -485,6 +518,7 @@ impl<'a, T: RuntimeTypes> TypedBuilder<'a, T> {
wasm: self.wasm,
manifest: self.manifest,
clocks: self.clocks,
supervisor_clock: self.supervisor_clock,
components,
_t: PhantomData,
}
Expand All @@ -498,6 +532,7 @@ pub struct ComponentsStage<'a, T: RuntimeTypes, C, S, E> {
wasm: Option<PathBuf>,
manifest: Option<PathBuf>,
clocks: Option<WasiClockOverride>,
supervisor_clock: Option<Arc<dyn SupervisorClock>>,
components: ComponentsBuilder<C, S, E>,
_t: PhantomData<fn() -> T>,
}
Expand All @@ -511,6 +546,7 @@ impl<'a, T: RuntimeTypes, C, S, E> ComponentsStage<'a, T, C, S, E> {
wasm: self.wasm,
manifest: self.manifest,
clocks: self.clocks,
supervisor_clock: self.supervisor_clock,
components: self.components,
add_ons,
}
Expand All @@ -525,6 +561,7 @@ pub struct ReadyBuilder<'a, T: RuntimeTypes, C, S, E> {
wasm: Option<PathBuf>,
manifest: Option<PathBuf>,
clocks: Option<WasiClockOverride>,
supervisor_clock: Option<Arc<dyn SupervisorClock>>,
components: ComponentsBuilder<C, S, E>,
add_ons: &'a [&'a dyn RuntimeAddOn],
}
Expand Down Expand Up @@ -557,6 +594,7 @@ where
wasm: self.wasm.as_deref(),
manifest: self.manifest.as_deref(),
clocks: self.clocks,
supervisor_clock: self.supervisor_clock,
};
let ctx = LaunchContext {
tasks,
Expand Down Expand Up @@ -704,6 +742,7 @@ every_n_blocks = "1"
wasm: None,
manifest: None,
clocks: None,
supervisor_clock: None,
};
let ctx = LaunchContext {
tasks,
Expand Down
47 changes: 40 additions & 7 deletions crates/nexum-runtime/src/supervisor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,10 @@ pub struct Supervisor<T: RuntimeTypes> {
/// including the ones rebuilt on restart. `None` leaves the ambient
/// host clocks.
clocks: Option<WasiClockOverride>,
/// Clock the poison window and restart backoff read `now()` from.
/// Real host clock in production; a harness swaps in a manually
/// driven fake for deterministic timing tests.
clock: Arc<dyn SupervisorClock>,
}

/// Core-only lattice for the runtime's own tests: the reference core
Expand Down Expand Up @@ -167,6 +171,28 @@ fn resolve_module_limits(res: &ResourceSection, cfg: &ModuleLimits) -> ResolvedL
}
}

/// Supervisor-internal clock seam for poison-window accounting and
/// restart-backoff scheduling. Distinct from [`WasiClockOverride`],
/// which only virtualizes guest-visible WASI time: this drives the
/// supervisor's own bookkeeping (`next_attempt`, `failure_timestamps`).
/// Defaults to the real host clock; a harness overrides it with a
/// manually-driven fake so backoff/poison-window tests assert
/// deterministically instead of sleeping real wall-clock time.
pub trait SupervisorClock: Send + Sync {
/// Current instant, per this clock's notion of time.
fn now(&self) -> std::time::Instant;
}

/// Production default: the real host [`Instant`](std::time::Instant).
#[derive(Default)]
struct SystemClock;

impl SupervisorClock for SystemClock {
fn now(&self) -> std::time::Instant {
std::time::Instant::now()
}
}

struct LoadedModule<T: RuntimeTypes> {
name: String,
bindings: EventModule,
Expand Down Expand Up @@ -245,6 +271,7 @@ impl<T: RuntimeTypes> Supervisor<T> {
components: &Components<T>,
extensions: &[Extension<T>],
clocks: Option<WasiClockOverride>,
supervisor_clock: Option<Arc<dyn SupervisorClock>>,
) -> Result<Self> {
let registry = capability_registry(extensions);
let mut modules = Vec::with_capacity(engine_cfg.modules.len());
Expand All @@ -270,6 +297,7 @@ impl<T: RuntimeTypes> Supervisor<T> {
extensions: extensions.to_vec(),
poison_policy: engine_cfg.limits.poison(),
clocks,
clock: supervisor_clock.unwrap_or_else(|| Arc::new(SystemClock)),
})
}

Expand All @@ -289,6 +317,7 @@ impl<T: RuntimeTypes> Supervisor<T> {
limits: &ModuleLimits,
extensions: &[Extension<T>],
clocks: Option<WasiClockOverride>,
supervisor_clock: Option<Arc<dyn SupervisorClock>>,
) -> Result<Self> {
let registry = capability_registry(extensions);
let entry = ModuleEntry {
Expand All @@ -312,6 +341,7 @@ impl<T: RuntimeTypes> Supervisor<T> {
extensions: extensions.to_vec(),
poison_policy: limits.poison(),
clocks,
clock: supervisor_clock.unwrap_or_else(|| Arc::new(SystemClock)),
})
}

Expand Down Expand Up @@ -733,7 +763,7 @@ impl<T: RuntimeTypes> Supervisor<T> {
let chain_id = chain.id();
let block_number = block.number;
let event = nexum::host::types::Event::Block(block);
let now = std::time::Instant::now();
let now = self.clock.now();
// Hoist the local-store reference out so the per-module
// borrow checker is happy when we write the progress
// marker after a successful dispatch.
Expand Down Expand Up @@ -820,7 +850,7 @@ impl<T: RuntimeTypes> Supervisor<T> {
log: alloy_rpc_types_eth::Log,
cursor_key: Option<&str>,
) -> bool {
let now = std::time::Instant::now();
let now = self.clock.now();
let Some(idx) = self.modules.iter().position(|m| m.name == module_name) else {
warn!(module = %module_name, "no such module - dropping chain-log");
return false;
Expand Down Expand Up @@ -907,8 +937,10 @@ impl<T: RuntimeTypes> Supervisor<T> {
let chain_id = chain.id();
let poison_policy = self.poison_policy;
// Hoisted before the per-module borrow so the trap arm can
// synthesize a panic record without re-borrowing `self`.
// synthesize a panic record and schedule the backoff without
// re-borrowing `self`.
let router = self.components.logs.router();
let now = self.clock.now();
let module = &mut self.modules[idx];
// Dispatch-boundary rate limit: throttle before spending any
// fuel or entering the guest, so a flood of cheap-to-dispatch
Expand Down Expand Up @@ -1010,7 +1042,7 @@ impl<T: RuntimeTypes> Supervisor<T> {
let latency_ms = elapsed.as_millis() as u64;
module.failure_count = module.failure_count.saturating_add(1);
let backoff = crate::runtime::restart_policy::backoff_for(module.failure_count);
let next_attempt = std::time::Instant::now() + backoff;
let next_attempt = now + backoff;
error!(
module = %module.name,
chain_id,
Expand Down Expand Up @@ -1041,7 +1073,7 @@ impl<T: RuntimeTypes> Supervisor<T> {
Level::ERROR,
format!("run terminated abnormally: {}", trap.root_cause()),
));
record_failure_and_maybe_poison(module, poison_policy, &trap.to_string());
record_failure_and_maybe_poison(module, poison_policy, now, &trap.to_string());
DispatchOutcome::Trapped
}
}
Expand All @@ -1068,10 +1100,11 @@ impl<T: RuntimeTypes> Supervisor<T> {
Err(e) => {
// Re-instantiation failed: bump the backoff again so
// the next attempt is further out.
let now = self.clock.now();
let m = &mut self.modules[idx];
m.failure_count = m.failure_count.saturating_add(1);
let backoff = crate::runtime::restart_policy::backoff_for(m.failure_count);
m.next_attempt = Some(std::time::Instant::now() + backoff);
m.next_attempt = Some(now + backoff);
error!(
module = %name,
failure_count = m.failure_count,
Expand Down Expand Up @@ -1210,9 +1243,9 @@ enum DispatchOutcome {
fn record_failure_and_maybe_poison<T: RuntimeTypes>(
module: &mut LoadedModule<T>,
policy: crate::runtime::poison_policy::PoisonPolicy,
now: std::time::Instant,
last_error: &str,
) {
let now = std::time::Instant::now();
// Prune entries outside the window.
while let Some(&front) = module.failure_timestamps.front() {
if now.duration_since(front) > policy.window {
Expand Down
Loading
Loading