Struct Supervisor 

pub struct Supervisor { /* private fields */ }

Host-side child restarter (design notes §15-16).

A Supervisor is not a bytecode Flow. It is a dedicated OS thread plus a table of ChildSpecs. When a supervised flow becomes FlowState::Failed (or completes, under RestartPolicy::Always), the worker delivers the FlowOutcome here instead of letting the fault take anything else down. The supervisor then consults the child’s RestartPolicy and, if intensity allows, respawns it under a fresh FlowId — Pids are never reused (see super::process::FlowId).

Constructed from a RuntimeSpawner so it does not have to own the runtime’s worker JoinHandles.

Implementations

impl Supervisor

pub fn new(spawner: RuntimeSpawner) -> Result<Self, SpawnError>

pub fn with_config( spawner: RuntimeSpawner, config: SupervisorConfig, ) -> Result<Self, SpawnError>

Start the dedicated supervisor OS thread. Thread-spawn failure is SpawnError::ThreadSpawnFailed — same category-A surface as super::runtime::Runtime::new, not a panic.

Examples found in repository

examples/crash_and_restart.rs (lines 34-40)

fn main() {
    let rt = match Runtime::with_config(
        samples::boom(),
        RuntimeConfig {
            workers: 1,
            quantum: 1_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(boom) = rt.function_index("boom") else {
        eprintln!("missing boom");
        std::process::exit(1);
    };
    let sup = match Supervisor::with_config(
        rt.spawner(),
        SupervisorConfig {
            max_restarts: 2,
            max_period: Duration::from_secs(5),
        },
    ) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("supervisor: {e}");
            std::process::exit(1);
        }
    };
    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
    {
        eprintln!("start_child: {e}");
        std::process::exit(1);
    }

    let start = std::time::Instant::now();
    while !sup.intensity_exceeded() {
        if start.elapsed() > Duration::from_secs(2) {
            eprintln!("supervisor did not hit intensity in time");
            std::process::exit(1);
        }
        thread::sleep(Duration::from_millis(5));
    }

    let metrics = rt.metrics();
    println!("intensity exceeded after {} failures", metrics.processes_failed);
    println!("{metrics}");
    sup.shutdown();
    rt.shutdown();
}

pub fn start_child(&self, spec: ChildSpec) -> Result<FlowHandle, SpawnError>

Spawn spec and start supervising it. The returned handle is for this incarnation only — a restart allocates a new Pid and a new completion channel.

Examples found in repository

examples/crash_and_restart.rs (line 47)

fn main() {
    let rt = match Runtime::with_config(
        samples::boom(),
        RuntimeConfig {
            workers: 1,
            quantum: 1_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(boom) = rt.function_index("boom") else {
        eprintln!("missing boom");
        std::process::exit(1);
    };
    let sup = match Supervisor::with_config(
        rt.spawner(),
        SupervisorConfig {
            max_restarts: 2,
            max_period: Duration::from_secs(5),
        },
    ) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("supervisor: {e}");
            std::process::exit(1);
        }
    };
    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
    {
        eprintln!("start_child: {e}");
        std::process::exit(1);
    }

    let start = std::time::Instant::now();
    while !sup.intensity_exceeded() {
        if start.elapsed() > Duration::from_secs(2) {
            eprintln!("supervisor did not hit intensity in time");
            std::process::exit(1);
        }
        thread::sleep(Duration::from_millis(5));
    }

    let metrics = rt.metrics();
    println!("intensity exceeded after {} failures", metrics.processes_failed);
    println!("{metrics}");
    sup.shutdown();
    rt.shutdown();
}

pub fn live_children(&self) -> usize

pub fn intensity_exceeded(&self) -> bool

true once more than SupervisorConfig::max_restarts respawns landed inside the intensity window. Remaining children keep running; we just stop bringing them back (no safe abort of a mid-quantum flow).

Examples found in repository

examples/crash_and_restart.rs (line 54)

fn main() {
    let rt = match Runtime::with_config(
        samples::boom(),
        RuntimeConfig {
            workers: 1,
            quantum: 1_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(boom) = rt.function_index("boom") else {
        eprintln!("missing boom");
        std::process::exit(1);
    };
    let sup = match Supervisor::with_config(
        rt.spawner(),
        SupervisorConfig {
            max_restarts: 2,
            max_period: Duration::from_secs(5),
        },
    ) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("supervisor: {e}");
            std::process::exit(1);
        }
    };
    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
    {
        eprintln!("start_child: {e}");
        std::process::exit(1);
    }

    let start = std::time::Instant::now();
    while !sup.intensity_exceeded() {
        if start.elapsed() > Duration::from_secs(2) {
            eprintln!("supervisor did not hit intensity in time");
            std::process::exit(1);
        }
        thread::sleep(Duration::from_millis(5));
    }

    let metrics = rt.metrics();
    println!("intensity exceeded after {} failures", metrics.processes_failed);
    println!("{metrics}");
    sup.shutdown();
    rt.shutdown();
}

pub fn shutdown(self)

Stop the drive thread. Does not terminate live children — they belong to the runtime, not to us.

Examples found in repository

examples/crash_and_restart.rs (line 65)

fn main() {
    let rt = match Runtime::with_config(
        samples::boom(),
        RuntimeConfig {
            workers: 1,
            quantum: 1_000,
            mailbox: byteflow::MailboxConfig::DEFAULT,
        },
    ) {
        Ok(rt) => rt,
        Err(e) => {
            eprintln!("runtime: {e}");
            std::process::exit(1);
        }
    };
    let Some(boom) = rt.function_index("boom") else {
        eprintln!("missing boom");
        std::process::exit(1);
    };
    let sup = match Supervisor::with_config(
        rt.spawner(),
        SupervisorConfig {
            max_restarts: 2,
            max_period: Duration::from_secs(5),
        },
    ) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("supervisor: {e}");
            std::process::exit(1);
        }
    };
    if let Err(e) = sup.start_child(ChildSpec::new("boom", boom).restart(RestartPolicy::OnFailure))
    {
        eprintln!("start_child: {e}");
        std::process::exit(1);
    }

    let start = std::time::Instant::now();
    while !sup.intensity_exceeded() {
        if start.elapsed() > Duration::from_secs(2) {
            eprintln!("supervisor did not hit intensity in time");
            std::process::exit(1);
        }
        thread::sleep(Duration::from_millis(5));
    }

    let metrics = rt.metrics();
    println!("intensity exceeded after {} failures", metrics.processes_failed);
    println!("{metrics}");
    sup.shutdown();
    rt.shutdown();
}