datum_agent/

registry.rs

use std::{
    collections::HashMap,
    future::Future,
    panic::{AssertUnwindSafe, catch_unwind},
    sync::{
        OnceLock,
        atomic::{AtomicU64, Ordering},
        mpsc,
    },
    time::{Duration, Instant, SystemTime},
};

use datum::{
    Materializer, NotUsed, RunnableGraph, Source, StreamError, StreamInstrumentationRegistry,
    StreamResult, Topic, TopicOverflow, TopicTryPublishError,
};
use ractor::{Actor, ActorProcessingErr, ActorRef};
use tokio::runtime::{Builder as TokioRuntimeBuilder, Runtime as TokioRuntime};

use crate::{
    AgentConfig, AgentError, AgentResult, ClusterJobMetadata, DesiredJobState, JobContext,
    JobControl, JobDrainBehavior, JobEvent, JobEventKind, JobExitReason, JobId, JobMat, JobSpec,
    JobState, JobStatus,
    job::{RestartCause, new_generation_kill_switch},
};

type Reply<T> = mpsc::Sender<AgentResult<T>>;

/// Ractor-backed local job registry.
pub struct JobRegistry;

impl JobRegistry {
    pub fn start(config: AgentConfig) -> AgentResult<JobRegistryHandle> {
        let event_buffer = config.event_buffer.max(1);
        let events = Topic::new(event_buffer, TopicOverflow::Sliding)?;
        let args = RegistryArgs {
            instrumentation: config.instrumentation.clone(),
            config,
            events: events.clone(),
            materializer: Materializer::new(),
        };
        let (actor, _join) =
            block_on_agent_runtime(RegistryActor::spawn(None, RegistryActor, args)).and_then(
                |result| {
                    result.map_err(|error| {
                        AgentError::Actor(format!("job registry failed to spawn: {error}"))
                    })
                },
            )?;
        Ok(JobRegistryHandle { actor, events })
    }
}

/// Cloneable in-process handle to the registry actor.
#[derive(Clone)]
pub struct JobRegistryHandle {
    actor: ActorRef<RegistryMessage>,
    events: Topic<JobEvent>,
}

impl JobRegistryHandle {
    pub fn submit(&self, spec: JobSpec) -> AgentResult<JobStatus> {
        self.request(|reply| RegistryMessage::Submit {
            spec: Box::new(spec),
            reply,
        })
    }

    pub fn start(&self, name: impl Into<String>) -> AgentResult<JobStatus> {
        let name = name.into();
        self.request(|reply| RegistryMessage::Start { name, reply })
    }

    pub fn restart(&self, name: impl Into<String>) -> AgentResult<JobStatus> {
        let name = name.into();
        self.request(|reply| RegistryMessage::Restart { name, reply })
    }

    pub fn drain(&self, name: impl Into<String>) -> AgentResult<JobStatus> {
        let name = name.into();
        self.request(|reply| RegistryMessage::Drain { name, reply })
    }

    pub fn stop(&self, name: impl Into<String>) -> AgentResult<JobStatus> {
        let name = name.into();
        self.request(|reply| RegistryMessage::Stop { name, reply })
    }

    pub fn status(&self, name: impl Into<String>) -> AgentResult<JobStatus> {
        let name = name.into();
        self.request(|reply| RegistryMessage::Status { name, reply })
    }

    pub fn list(&self) -> AgentResult<Vec<JobStatus>> {
        self.request(|reply| RegistryMessage::List { reply })
    }

    pub(crate) fn update_cluster_metadata(
        &self,
        name: impl Into<String>,
        metadata: ClusterJobMetadata,
    ) -> AgentResult<JobStatus> {
        self.request(|reply| RegistryMessage::UpdateClusterMetadata {
            name: name.into(),
            metadata: Box::new(metadata),
            reply,
        })
    }

    #[must_use]
    pub fn events(&self) -> Source<JobEvent> {
        self.events.subscribe()
    }

    pub fn shutdown(&self) -> AgentResult<()> {
        self.request(RegistryMessage::Shutdown)
    }

    fn request<T, F>(&self, make: F) -> AgentResult<T>
    where
        T: Send + 'static,
        F: FnOnce(Reply<T>) -> RegistryMessage,
    {
        let (reply, receiver) = mpsc::channel();
        self.actor
            .send_message(make(reply))
            .map_err(|_| AgentError::RegistryClosed)?;
        receiver.recv().map_err(|_| AgentError::RegistryClosed)?
    }
}

struct RegistryArgs {
    instrumentation: StreamInstrumentationRegistry,
    config: AgentConfig,
    events: Topic<JobEvent>,
    materializer: Materializer,
}

struct RegistryActor;

impl Actor for RegistryActor {
    type Msg = RegistryMessage;
    type State = RegistryState;
    type Arguments = RegistryArgs;

    async fn pre_start(
        &self,
        _myself: ActorRef<Self::Msg>,
        args: Self::Arguments,
    ) -> Result<Self::State, ActorProcessingErr> {
        Ok(RegistryState {
            config: args.config,
            materializer: args.materializer,
            events: args.events,
            instrumentation: args.instrumentation,
            next_job_id: 1,
            next_event_sequence: 1,
            jobs: HashMap::new(),
        })
    }

    async fn post_start(
        &self,
        myself: ActorRef<Self::Msg>,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        let interval = state.config.poll_interval.max(Duration::from_millis(1));
        spawn_completion_poller(myself, interval);
        Ok(())
    }

    async fn post_stop(
        &self,
        _myself: ActorRef<Self::Msg>,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        abort_all_jobs(state);
        let materializer = state.materializer.clone();
        let events = state.events.clone();
        materializer.shutdown();
        let _ = events.close();
        Ok(())
    }

    async fn handle(
        &self,
        myself: ActorRef<Self::Msg>,
        message: Self::Msg,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        match message {
            RegistryMessage::Submit { spec, reply } => {
                send_reply(reply, submit_job(state, *spec));
            }
            RegistryMessage::Start { name, reply } => {
                send_reply(reply, start_job(state, &myself, &name, StartKind::Initial));
            }
            RegistryMessage::Restart { name, reply } => {
                send_reply(reply, restart_job(state, &myself, &name));
            }
            RegistryMessage::Drain { name, reply } => {
                send_reply(reply, drain_job(state, &name));
            }
            RegistryMessage::Stop { name, reply } => {
                send_reply(reply, stop_job(state, &name, JobExitReason::Stopped));
            }
            RegistryMessage::Status { name, reply } => {
                send_reply(reply, status_by_name(state, &name));
            }
            RegistryMessage::List { reply } => {
                send_reply(reply, Ok(list_statuses(state)));
            }
            RegistryMessage::UpdateClusterMetadata {
                name,
                metadata,
                reply,
            } => {
                send_reply(reply, update_cluster_metadata(state, &name, *metadata));
            }
            RegistryMessage::Tick => {
                poll_jobs(state, &myself);
            }
            RegistryMessage::RestartReady {
                name,
                expected_generation,
            } => {
                restart_ready(state, &myself, &name, expected_generation);
            }
            RegistryMessage::Shutdown(reply) => {
                abort_all_jobs(state);
                let _ = reply.send(Ok(()));
                myself.stop(Some("datum-agent shutdown".to_owned()));
            }
        }
        Ok(())
    }
}

enum RegistryMessage {
    Submit {
        spec: Box<JobSpec>,
        reply: Reply<JobStatus>,
    },
    Start {
        name: String,
        reply: Reply<JobStatus>,
    },
    Restart {
        name: String,
        reply: Reply<JobStatus>,
    },
    Drain {
        name: String,
        reply: Reply<JobStatus>,
    },
    Stop {
        name: String,
        reply: Reply<JobStatus>,
    },
    Status {
        name: String,
        reply: Reply<JobStatus>,
    },
    List {
        reply: Reply<Vec<JobStatus>>,
    },
    UpdateClusterMetadata {
        name: String,
        metadata: Box<ClusterJobMetadata>,
        reply: Reply<JobStatus>,
    },
    Tick,
    RestartReady {
        name: String,
        expected_generation: u64,
    },
    Shutdown(Reply<()>),
}

struct RegistryState {
    config: AgentConfig,
    materializer: Materializer,
    events: Topic<JobEvent>,
    instrumentation: StreamInstrumentationRegistry,
    next_job_id: u64,
    next_event_sequence: u64,
    jobs: HashMap<String, JobEntry>,
}

struct JobEntry {
    id: JobId,
    spec: JobSpec,
    state: JobState,
    desired_state: DesiredJobState,
    generation: u64,
    run: Option<JobRun>,
    backoff: RestartBackoffState,
    starts_total: u64,
    restarts_total: u64,
    last_start_at: Option<SystemTime>,
    last_exit_at: Option<SystemTime>,
    last_exit_reason: Option<JobExitReason>,
    backoff_until: Option<Instant>,
    drain_deadline: Option<Instant>,
    drain_supported: bool,
}

impl JobEntry {
    fn new(id: JobId, spec: JobSpec) -> Self {
        Self {
            id,
            spec,
            state: JobState::Submitted,
            desired_state: DesiredJobState::Stopped,
            generation: 0,
            run: None,
            backoff: RestartBackoffState::default(),
            starts_total: 0,
            restarts_total: 0,
            last_start_at: None,
            last_exit_at: None,
            last_exit_reason: None,
            backoff_until: None,
            drain_deadline: None,
            drain_supported: true,
        }
    }
}

struct JobRun {
    generation: u64,
    control: JobControl,
    completion: datum::StreamCompletion<NotUsed>,
}

#[derive(Default)]
struct RestartBackoffState {
    window_start: Option<Instant>,
    restarts_in_window: usize,
}

impl RestartBackoffState {
    fn reset(&mut self) {
        self.window_start = None;
        self.restarts_in_window = 0;
    }

    fn next_delay(&mut self, settings: &datum::RestartSettings) -> Option<Duration> {
        let now = Instant::now();
        match self.window_start {
            Some(window_start)
                if now.duration_since(window_start) <= settings.max_restarts_within() => {}
            _ => {
                self.window_start = Some(now);
                self.restarts_in_window = 0;
            }
        }

        if self.restarts_in_window >= settings.max_restarts() {
            return None;
        }

        let exponent = (self.restarts_in_window.min(31) as i32).max(0);
        let base = settings
            .min_backoff()
            .mul_f64(2_f64.powi(exponent))
            .min(settings.max_backoff());
        self.restarts_in_window += 1;

        if settings.random_factor() == 0.0 || base.is_zero() {
            Some(base)
        } else {
            let jitter = base.mul_f64(next_random_fraction() * settings.random_factor());
            Some(base.saturating_add(jitter))
        }
    }
}

#[derive(Clone, Copy)]
enum StartKind {
    Initial,
    AutomaticRestart { previous_generation: u64 },
    ManualRestart,
}

fn submit_job(state: &mut RegistryState, spec: JobSpec) -> AgentResult<JobStatus> {
    if spec.name.trim().is_empty() {
        return Err(AgentError::InvalidJobName);
    }
    if state.jobs.contains_key(&spec.name) {
        return Err(AgentError::JobAlreadyExists(spec.name));
    }

    let name = spec.name.clone();
    let id = JobId(state.next_job_id);
    state.next_job_id += 1;
    state.jobs.insert(name.clone(), JobEntry::new(id, spec));
    emit_event(state, &name, 0, JobEventKind::Submitted);
    status_by_name(state, &name)
}

fn update_cluster_metadata(
    state: &mut RegistryState,
    name: &str,
    metadata: ClusterJobMetadata,
) -> AgentResult<JobStatus> {
    let entry = state
        .jobs
        .get_mut(name)
        .ok_or_else(|| AgentError::JobNotFound(name.to_owned()))?;
    entry.spec.cluster = Some(metadata);
    status_by_name(state, name)
}

fn start_job(
    state: &mut RegistryState,
    myself: &ActorRef<RegistryMessage>,
    name: &str,
    kind: StartKind,
) -> AgentResult<JobStatus> {
    let (spec, job_id, generation) = {
        let entry = state
            .jobs
            .get_mut(name)
            .ok_or_else(|| AgentError::JobNotFound(name.to_owned()))?;

        if matches!(
            entry.state,
            JobState::Starting | JobState::Running | JobState::Draining | JobState::BackingOff
        ) && !matches!(kind, StartKind::AutomaticRestart { .. })
        {
            return Err(AgentError::JobAlreadyRunning(name.to_owned()));
        }

        entry.state = JobState::Starting;
        entry.desired_state = DesiredJobState::Running;
        entry.generation += 1;
        entry.backoff_until = None;
        entry.drain_deadline = None;
        entry.run = None;
        (entry.spec.clone(), entry.id, entry.generation)
    };

    let kill_switch = new_generation_kill_switch(&spec.name, generation);
    let context = JobContext::new(
        spec.name.clone(),
        job_id,
        generation,
        kill_switch,
        state.instrumentation.clone(),
    );
    let graph = invoke_factory(&spec, context);
    let materializer = state
        .materializer
        .with_name_prefix(format!("datum-agent-{}", spec.name));
    let materialized = graph.and_then(|graph| materialize_graph(&graph, &materializer));

    match materialized {
        Ok(mat) => {
            let (completion, control) = mat.into_parts();
            let mut restart_event = None;
            {
                let entry = state
                    .jobs
                    .get_mut(name)
                    .ok_or_else(|| AgentError::JobNotFound(name.to_owned()))?;
                entry.run = Some(JobRun {
                    generation,
                    control: control.clone(),
                    completion,
                });
                entry.state = JobState::Running;
                entry.desired_state = DesiredJobState::Running;
                entry.starts_total += 1;
                entry.last_start_at = Some(SystemTime::now());
                entry.drain_supported = control.drain_supported();
                if let StartKind::AutomaticRestart {
                    previous_generation,
                } = kind
                {
                    entry.restarts_total += 1;
                    restart_event = Some(JobEventKind::Restarted {
                        previous_generation,
                    });
                }
            }
            if let Some(kind) = restart_event {
                emit_event(state, name, generation, kind);
            }
            emit_event(state, name, generation, JobEventKind::Started);
            status_by_name(state, name)
        }
        Err(error) => {
            handle_start_failure(state, myself, name, generation, error.clone());
            Err(AgentError::Stream(error))
        }
    }
}

fn restart_job(
    state: &mut RegistryState,
    myself: &ActorRef<RegistryMessage>,
    name: &str,
) -> AgentResult<JobStatus> {
    if !state.jobs.contains_key(name) {
        return Err(AgentError::JobNotFound(name.to_owned()));
    }
    let _ = stop_job(state, name, JobExitReason::Stopped)?;
    if let Some(entry) = state.jobs.get_mut(name) {
        entry.backoff.reset();
    }
    start_job(state, myself, name, StartKind::ManualRestart)
}

fn drain_job(state: &mut RegistryState, name: &str) -> AgentResult<JobStatus> {
    let generation = {
        let entry = state
            .jobs
            .get_mut(name)
            .ok_or_else(|| AgentError::JobNotFound(name.to_owned()))?;

        let Some(run) = entry.run.as_ref() else {
            entry.state = JobState::Drained;
            entry.desired_state = DesiredJobState::Stopped;
            entry.last_exit_at = Some(SystemTime::now());
            entry.last_exit_reason = Some(JobExitReason::Drained);
            return status_by_name(state, name);
        };

        let generation = run.generation;
        let drain_supported = run.control.drain_supported();
        let control = run.control.clone();
        let timeout = match entry.spec.drain_behavior {
            JobDrainBehavior::Graceful { timeout } if drain_supported => timeout,
            JobDrainBehavior::Graceful { .. } | JobDrainBehavior::CancelOnly => {
                return Err(AgentError::DrainUnsupported(name.to_owned()));
            }
        };

        if !control.shutdown() {
            return Err(AgentError::DrainUnsupported(name.to_owned()));
        }

        entry.state = JobState::Draining;
        entry.desired_state = DesiredJobState::Draining;
        let deadline = Instant::now() + timeout;
        entry.drain_deadline = Some(deadline);
        generation
    };

    emit_event(state, name, generation, JobEventKind::Draining);
    status_by_name(state, name)
}

fn stop_job(
    state: &mut RegistryState,
    name: &str,
    reason: JobExitReason,
) -> AgentResult<JobStatus> {
    let generation = {
        let entry = state
            .jobs
            .get_mut(name)
            .ok_or_else(|| AgentError::JobNotFound(name.to_owned()))?;

        let generation = entry.generation;
        if let Some(run) = entry.run.take() {
            let _ = run.control.abort(StreamError::Cancelled);
            drop(run.completion);
        }
        entry.state = JobState::Stopped;
        entry.desired_state = DesiredJobState::Stopped;
        entry.backoff_until = None;
        entry.drain_deadline = None;
        entry.last_exit_at = Some(SystemTime::now());
        entry.last_exit_reason = Some(reason.clone());
        generation
    };

    emit_event(state, name, generation, JobEventKind::Stopped { reason });
    status_by_name(state, name)
}

fn status_by_name(state: &RegistryState, name: &str) -> AgentResult<JobStatus> {
    state
        .jobs
        .get(name)
        .map(|entry| snapshot(state, entry))
        .ok_or_else(|| AgentError::JobNotFound(name.to_owned()))
}

fn list_statuses(state: &RegistryState) -> Vec<JobStatus> {
    let mut statuses = state
        .jobs
        .values()
        .map(|entry| snapshot(state, entry))
        .collect::<Vec<_>>();
    statuses.sort_by(|left, right| left.name.cmp(&right.name));
    statuses
}

fn poll_jobs(state: &mut RegistryState, myself: &ActorRef<RegistryMessage>) {
    let names = state.jobs.keys().cloned().collect::<Vec<_>>();
    for name in names {
        let completed = state.jobs.get_mut(&name).and_then(|entry| {
            entry.run.as_mut().and_then(|run| {
                run.completion
                    .try_wait()
                    .map(|result| (run.generation, result))
            })
        });

        if let Some((generation, result)) = completed {
            finish_generation(state, myself, &name, generation, result);
        }
    }

    let now = Instant::now();
    let timed_out = state
        .jobs
        .iter()
        .filter_map(|(name, entry)| match (entry.state, entry.drain_deadline) {
            (JobState::Draining, Some(deadline)) if deadline <= now => Some(name.clone()),
            _ => None,
        })
        .collect::<Vec<_>>();

    for name in timed_out {
        let _ = stop_job(state, &name, JobExitReason::DrainTimedOut);
    }
}

fn finish_generation(
    state: &mut RegistryState,
    myself: &ActorRef<RegistryMessage>,
    name: &str,
    generation: u64,
    result: StreamResult<NotUsed>,
) {
    let mut restart = None;
    let event;
    {
        let Some(entry) = state.jobs.get_mut(name) else {
            return;
        };
        if entry.run.as_ref().map(|run| run.generation) != Some(generation) {
            return;
        }
        let _run = entry.run.take();
        entry.drain_deadline = None;
        entry.last_exit_at = Some(SystemTime::now());

        match (entry.desired_state, result) {
            (DesiredJobState::Draining, Ok(_)) => {
                entry.state = JobState::Drained;
                entry.desired_state = DesiredJobState::Stopped;
                entry.last_exit_reason = Some(JobExitReason::Drained);
                event = JobEventKind::Drained;
            }
            (DesiredJobState::Draining | DesiredJobState::Stopped, Err(error)) => {
                let reason = if matches!(error, StreamError::Cancelled) {
                    JobExitReason::Stopped
                } else {
                    JobExitReason::Failed(error)
                };
                entry.state = JobState::Stopped;
                entry.desired_state = DesiredJobState::Stopped;
                entry.last_exit_reason = Some(reason.clone());
                event = JobEventKind::Stopped { reason };
            }
            (DesiredJobState::Stopped, Ok(_)) => {
                entry.state = JobState::Stopped;
                entry.last_exit_reason = Some(JobExitReason::Stopped);
                event = JobEventKind::Stopped {
                    reason: JobExitReason::Stopped,
                };
            }
            (DesiredJobState::Running, Ok(_)) => {
                entry.last_exit_reason = Some(JobExitReason::Completed);
                entry.state = JobState::Completed;
                event = JobEventKind::Completed;
                if let Some(settings) = entry
                    .spec
                    .restart_policy
                    .settings_for(RestartCause::Completion)
                {
                    restart = Some((settings, RestartCause::Completion));
                }
            }
            (DesiredJobState::Running, Err(error)) => {
                let reason = JobExitReason::Failed(error);
                entry.last_exit_reason = Some(reason.clone());
                entry.state = JobState::Failed;
                event = JobEventKind::Failed { reason };
                if let Some(settings) = entry
                    .spec
                    .restart_policy
                    .settings_for(RestartCause::Failure)
                {
                    restart = Some((settings, RestartCause::Failure));
                }
            }
        }
    }

    emit_event(state, name, generation, event);
    if let Some((settings, cause)) = restart {
        schedule_restart(state, myself, name, generation, settings, cause);
    }
}

fn handle_start_failure(
    state: &mut RegistryState,
    myself: &ActorRef<RegistryMessage>,
    name: &str,
    generation: u64,
    error: StreamError,
) {
    let mut restart = None;
    {
        let Some(entry) = state.jobs.get_mut(name) else {
            return;
        };
        let reason = JobExitReason::Failed(error);
        entry.state = JobState::Failed;
        entry.run = None;
        entry.last_exit_at = Some(SystemTime::now());
        entry.last_exit_reason = Some(reason.clone());
        if let Some(settings) = entry
            .spec
            .restart_policy
            .settings_for(RestartCause::Failure)
        {
            restart = Some(settings);
        }
    }

    let reason = state
        .jobs
        .get(name)
        .and_then(|entry| entry.last_exit_reason.clone())
        .unwrap_or(JobExitReason::Failed(StreamError::AbruptTermination));
    emit_event(state, name, generation, JobEventKind::Failed { reason });
    if let Some(settings) = restart {
        schedule_restart(
            state,
            myself,
            name,
            generation,
            settings,
            RestartCause::Failure,
        );
    }
}

fn schedule_restart(
    state: &mut RegistryState,
    myself: &ActorRef<RegistryMessage>,
    name: &str,
    generation: u64,
    settings: datum::RestartSettings,
    _cause: RestartCause,
) {
    let delay = {
        let Some(entry) = state.jobs.get_mut(name) else {
            return;
        };
        if entry.desired_state != DesiredJobState::Running {
            return;
        }
        let Some(delay) = entry.backoff.next_delay(&settings) else {
            entry.backoff_until = None;
            return;
        };
        entry.state = JobState::BackingOff;
        entry.backoff_until = Some(Instant::now() + delay);
        delay
    };

    emit_event(
        state,
        name,
        generation,
        JobEventKind::RestartScheduled { delay },
    );

    let actor = myself.clone();
    let name = name.to_owned();
    tokio::spawn(async move {
        tokio::time::sleep(delay).await;
        let _ = actor.send_message(RegistryMessage::RestartReady {
            name,
            expected_generation: generation,
        });
    });
}

fn restart_ready(
    state: &mut RegistryState,
    myself: &ActorRef<RegistryMessage>,
    name: &str,
    expected_generation: u64,
) {
    let ready = state.jobs.get(name).is_some_and(|entry| {
        entry.generation == expected_generation
            && entry.state == JobState::BackingOff
            && entry.desired_state == DesiredJobState::Running
    });
    if ready {
        let _ = start_job(
            state,
            myself,
            name,
            StartKind::AutomaticRestart {
                previous_generation: expected_generation,
            },
        );
    }
}

fn abort_all_jobs(state: &mut RegistryState) {
    let names = state.jobs.keys().cloned().collect::<Vec<_>>();
    for name in names {
        let _ = stop_job(state, &name, JobExitReason::Stopped);
    }
}

fn snapshot(_state: &RegistryState, entry: &JobEntry) -> JobStatus {
    JobStatus {
        name: entry.spec.name.clone(),
        job_id: entry.id,
        state: entry.state,
        desired_state: entry.desired_state,
        generation: entry.generation,
        starts_total: entry.starts_total,
        restarts_total: entry.restarts_total,
        last_start_at: entry.last_start_at,
        last_exit_at: entry.last_exit_at,
        last_exit_reason: entry.last_exit_reason.clone(),
        backoff_until: entry.backoff_until,
        drain_deadline: entry.drain_deadline,
        drain_supported: entry.drain_supported,
        active_streams: Some(usize::from(entry.run.is_some())),
        instrumentation: None,
        cluster: entry.spec.cluster.clone(),
    }
}

fn emit_event(state: &mut RegistryState, name: &str, generation: u64, kind: JobEventKind) {
    let Some(entry) = state.jobs.get(name) else {
        return;
    };
    let event = JobEvent {
        sequence: state.next_event_sequence,
        timestamp: SystemTime::now(),
        name: entry.spec.name.clone(),
        job_id: entry.id,
        generation,
        kind,
    };
    state.next_event_sequence += 1;
    match state.events.try_publish(event) {
        Ok(()) | Err(TopicTryPublishError::Full(_)) | Err(TopicTryPublishError::Busy(_)) => {}
        Err(TopicTryPublishError::Closed(_)) => {}
    }
}

fn invoke_factory(spec: &JobSpec, context: JobContext) -> StreamResult<RunnableGraph<JobMat>> {
    catch_unwind(AssertUnwindSafe(|| (spec.factory)(context)))
        .unwrap_or_else(|_| Err(StreamError::Failed("job factory panicked".to_owned())))
}

fn materialize_graph(
    graph: &RunnableGraph<JobMat>,
    materializer: &Materializer,
) -> StreamResult<JobMat> {
    catch_unwind(AssertUnwindSafe(|| {
        graph.run_with_materializer(materializer)
    }))
    .unwrap_or_else(|_| {
        Err(StreamError::Failed(
            "job materialization panicked".to_owned(),
        ))
    })
}

fn send_reply<T>(reply: Reply<T>, result: AgentResult<T>) {
    let _ = reply.send(result);
}

fn spawn_completion_poller(actor: ActorRef<RegistryMessage>, interval: Duration) {
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(interval).await;
            if actor.send_message(RegistryMessage::Tick).is_err() {
                break;
            }
        }
    });
}

fn agent_runtime() -> AgentResult<&'static TokioRuntime> {
    static RUNTIME: OnceLock<Result<TokioRuntime, String>> = OnceLock::new();
    match RUNTIME.get_or_init(|| {
        TokioRuntimeBuilder::new_multi_thread()
            .thread_name("datum-agent-ractor")
            .enable_all()
            .build()
            .map_err(|error| format!("agent runtime failed to start: {error}"))
    }) {
        Ok(runtime) => Ok(runtime),
        Err(error) => Err(AgentError::Actor(error.clone())),
    }
}

fn block_on_agent_runtime<T, F>(future: F) -> AgentResult<T>
where
    T: Send + 'static,
    F: Future<Output = T> + Send + 'static,
{
    let runtime = agent_runtime()?;
    if tokio::runtime::Handle::try_current().is_ok() {
        std::thread::spawn(move || runtime.block_on(future))
            .join()
            .map_err(|_| AgentError::Actor("agent runtime task panicked".to_owned()))
    } else {
        Ok(runtime.block_on(future))
    }
}

fn next_random_fraction() -> f64 {
    static BACKOFF_RANDOM_SEED: AtomicU64 = AtomicU64::new(0x9e37_79b9_7f4a_7c15);
    let mut current = BACKOFF_RANDOM_SEED.load(Ordering::Relaxed);
    loop {
        let mut next = current;
        next ^= next << 13;
        next ^= next >> 7;
        next ^= next << 17;
        match BACKOFF_RANDOM_SEED.compare_exchange_weak(
            current,
            next,
            Ordering::Relaxed,
            Ordering::Relaxed,
        ) {
            Ok(_) => return ((next >> 11) as f64) / ((1_u64 << 53) as f64),
            Err(observed) => current = observed,
        }
    }
}