Skip to content

fix(supervisor): close adapter subscription and log-filter gaps #99

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

Merged
mfw78 merged 6 commits into from
Aug 6, 2026
Merged

fix(supervisor): close adapter subscription and log-filter gaps #99

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
bfc9a78
refactor(supervisor): promote Role to its own module and drop the dea…
mfw78 Aug 6, 2026
61a9009
refactor(supervisor): fold the provider and module load loops into one
mfw78 Aug 6, 2026
3964544
refactor(supervisor): extract role-keyed telemetry helpers
mfw78 Aug 6, 2026
d1d7eef
fix(supervisor): fail closed on adapter chain subscriptions and chain…
mfw78 Aug 6, 2026
9f2cbcb
refactor(supervisor): read the boot_single manifest role from the Rol…
mfw78 Aug 6, 2026
2d089e7
test(supervisor): pin the chain-log filter rebuild and the pre-compil…
mfw78 Aug 6, 2026
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
21 changes: 7 additions & 14 deletions crates/nexum-runtime/src/supervisor/dispatch.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ use tracing_core::Level;
use super::Supervisor;
use super::cursors::{commit_chain_log_cursor, persist_progress_marker};
use super::lifecycle::{revive_one, sweep};
use super::role::{Role, report_poison};
use crate::bindings::nexum;
use crate::host::component::RuntimeTypes;
use crate::host::extension::ExtensionEvent;
Expand Down Expand Up @@ -288,21 +289,13 @@ impl<T: RuntimeTypes> Supervisor<T> {
format!("run terminated abnormally: {}", trap.root_cause()),
));
if let Some(recent) = verdict.poisoned {
// A string field, not a `Display` one: the two record
// through different visitor methods and print differently.
let last_error = trap.to_string();
warn!(
module = %module.name,
recent_failures = recent,
window_secs = poison_policy.window.as_secs(),
last_error,
"module poisoned - quarantined; remove from engine.toml + restart to clear",
report_poison(
Role::Module,
&module.name,
recent,
poison_policy.window,
Some(trap.to_string()),
);
metrics::gauge!(
"nexum_runtime_module_poisoned",
"module" => module.name.clone(),
)
.set(1.0);
}
DispatchOutcome::Trapped
}
Expand Down
140 changes: 10 additions & 130 deletions crates/nexum-runtime/src/supervisor/lifecycle.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,10 @@ use std::collections::VecDeque;
use std::time::{Duration, Instant};

use anyhow::{Context, Error, Result, anyhow};
use tracing::{error, info, warn};

use super::Shared;
use super::load::{LoadedModule, LoadedProvider, run_init};
use super::role::{Role, report_restart_attempt, report_restart_outcome, report_trap};
use super::store::{self, build_linker, build_provider_linker};
use crate::bindings::EventModule;
use crate::digest::ContentDigest;
Expand Down Expand Up @@ -146,49 +146,6 @@ impl Health {
}
}

/// Keys the per-role metric names and the compile-time tracing field keys.
#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::IntoStaticStr)]
pub(super) enum Role {
Module,
Adapter,
}

impl Role {
const fn label(self) -> &'static str {
match self {
Self::Module => "module",
Self::Adapter => "adapter",
}
}

const fn errors_total(self) -> &'static str {
match self {
Self::Module => "nexum_runtime_module_errors_total",
Self::Adapter => "nexum_runtime_adapter_errors_total",
}
}

const fn restarts_total(self) -> &'static str {
match self {
Self::Module => "nexum_runtime_module_restarts_total",
Self::Adapter => "nexum_runtime_adapter_restarts_total",
}
}

const fn poisoned_gauge(self) -> &'static str {
match self {
Self::Module => "nexum_runtime_module_poisoned",
Self::Adapter => "nexum_runtime_adapter_poisoned",
}
}

/// A provider reinstall is a fresh instance, so its curve resets; a
/// module recovers in place and keeps climbing.
const fn resets_failure_count(self) -> bool {
matches!(self, Self::Adapter)
}
}

/// Run identity mints and commits only inside a successful [`Sweepable::revive`];
/// a failed attempt never advances the run sequence.
pub(super) trait Sweepable<T: RuntimeTypes> {
Expand Down Expand Up @@ -358,47 +315,7 @@ pub(super) async fn sweep<T: RuntimeTypes, S: Sweepable<T>>(
let policy = item.poison_policy(engine_default);
if let Some(died_at) = item.detect_death() {
let verdict = item.health_mut().record_trap(died_at, now, policy);
match S::ROLE {
Role::Module => warn!(
module = %item.name(),
failure_count = verdict.failure_count,
backoff_ms = verdict.backoff.as_millis() as u64,
"module trapped - marked dead; will restart after backoff",
),
Role::Adapter => warn!(
adapter = %item.name(),
failure_count = verdict.failure_count,
backoff_ms = verdict.backoff.as_millis() as u64,
"adapter trapped - marked dead; will restart after backoff",
),
}
metrics::counter!(
S::ROLE.errors_total(),
S::ROLE.label() => item.name().clone(),
"error_kind" => "trap",
)
.increment(1);
if let Some(recent) = verdict.poisoned {
match S::ROLE {
Role::Module => warn!(
module = %item.name(),
recent_failures = recent,
window_secs = policy.window.as_secs(),
"module poisoned - quarantined; remove from engine.toml + restart to clear",
),
Role::Adapter => warn!(
adapter = %item.name(),
recent_failures = recent,
window_secs = policy.window.as_secs(),
"adapter poisoned - quarantined; remove from engine.toml + restart to clear",
),
}
metrics::gauge!(
S::ROLE.poisoned_gauge(),
S::ROLE.label() => item.name().clone(),
)
.set(1.0);
}
report_trap(S::ROLE, item.name(), &verdict, policy);
}
if item.health().due_restart(now) {
revive_one(shared, item, now).await;
Expand All @@ -411,59 +328,22 @@ pub(super) async fn revive_one<T: RuntimeTypes, S: Sweepable<T>>(
item: &mut S,
now: Instant,
) {
let failure_count = item.health().failure_count();
// Revives reuse the cached component, so the boot-time digest holds.
match S::ROLE {
Role::Module => info!(
module = %item.name(),
failure_count,
digest = %item.digest(),
"restart attempt",
),
Role::Adapter => info!(
adapter = %item.name(),
failure_count,
digest = %item.digest(),
"adapter restart attempt",
),
}
metrics::counter!(
S::ROLE.restarts_total(),
S::ROLE.label() => item.name().clone(),
)
.increment(1);
report_restart_attempt(
S::ROLE,
item.name(),
item.health().failure_count(),
item.digest(),
);
match item.revive(shared).await {
Ok(()) => {
item.health_mut()
.restart_succeeded(S::ROLE.resets_failure_count());
match S::ROLE {
Role::Module => info!(module = %item.name(), "restart succeeded"),
Role::Adapter => info!(adapter = %item.name(), "adapter restart succeeded"),
}
report_restart_outcome(S::ROLE, item.name(), Ok(()));
}
Err(e) => {
let deferral = item.health_mut().defer_restart(now);
match S::ROLE {
Role::Module => error!(
module = %item.name(),
failure_count = deferral.failure_count,
backoff_ms = deferral.backoff.as_millis() as u64,
error = %e,
"restart failed - will retry after backoff",
),
Role::Adapter => {
// A string field, not a `Display` one: the two record
// through different visitor methods and print differently.
let error = format!("{e:#}");
error!(
adapter = %item.name(),
failure_count = deferral.failure_count,
backoff_ms = deferral.backoff.as_millis() as u64,
error,
"adapter restart failed - will retry after backoff",
);
}
}
report_restart_outcome(S::ROLE, item.name(), Err((deferral, e)));
}
}
}
Expand Down
7 changes: 3 additions & 4 deletions crates/nexum-runtime/src/supervisor/load.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ use super::admission::{
use super::artifact::read_verified_component;
use super::dispatch::with_dispatch_deadline;
use super::lifecycle::Health;
use super::prepass::{MODULE_FALLBACK_NAME, PROVIDER_FALLBACK_NAME, manifest_namespace};
use super::prepass::manifest_namespace;
use super::store::{
self, HostStore, ResolvedLimits, StoreSpec, build_provider_linker, resolve_module_limits,
};
Expand Down Expand Up @@ -151,8 +151,7 @@ pub(super) async fn module<T: RuntimeTypes>(
require_component_digest: bool,
provider_manifests: &[ProviderManifest],
) -> Result<LoadedModule<T>> {
let module_namespace: ModuleId =
manifest_namespace(&loaded_manifest, MODULE_FALLBACK_NAME).into();
let module_namespace: ModuleId = manifest_namespace(&loaded_manifest).into();
let registry = capability_registry(&shared.extensions);
let sections = &loaded_manifest.manifest.extensions;
let ((), component, digest) = admit_and_verify(
Expand Down Expand Up @@ -274,7 +273,7 @@ pub(super) async fn provider<T: RuntimeTypes>(
limits_cfg: &ModuleLimits,
require_component_digest: bool,
) -> Result<LoadedProvider> {
let namespace: ModuleId = manifest_namespace(&loaded_manifest, PROVIDER_FALLBACK_NAME).into();
let namespace: ModuleId = manifest_namespace(&loaded_manifest).into();
// A core-only declaration fails at manifest load; an undeclared gated
// import fails after compile; the linker withholds the core interfaces.
let registry = CapabilityRegistry::provider();
Expand Down
Loading
Loading