Agent Jobs, Lifecycle & DCP

datum-agent is the operational shell around a Datum application: it turns ad-hoc materialized streams into named, supervised jobs with graceful drain and restart policies, and exposes them to tooling over DCP (the Datum Control Protocol). It is the foundation the datum CLI, the datum-tui cluster console, and cluster placement all talk to.

cargo add datum-agent

The registry is a Ractor actor, but it does not carry stream elements (the two-plane rule): it owns specs, desired state, generation numbers, completion polling, restart timers, and lifecycle events. The job itself remains an ordinary Datum stream.

The job model

A job is a named blueprint factory plus a restart policy. Factories receive a JobContext and return a RunnableGraph<JobMat> — blueprints only; the registry materializes on start and on every automatic restart:

rust

use datum::prelude::*;
use datum_agent::{Agent, JobMat, JobSpec};

// Start an embedded agent and get its job registry.
let agent = Agent::start().expect("agent starts");
let registry = agent.registry();

// A job is a NAMED, SUPERVISED stream: the factory rebuilds the blueprint on
// every (re)start, and the agent wires in drain + instrumentation for you.
let spec = JobSpec::new("tick-counter", |context| {
    let control = context.control();
    Ok(
        Source::tick(Duration::ZERO, Duration::from_millis(10), 1_u64)
            .instrumented(
                format!("{}:{}", context.name(), context.generation()),
                context.instrumentation_registry(),
            )
            .via_mat(context.drain_flow(), Keep::right)
            .to_mat(Sink::ignore(), move |_switch, completion| {
                JobMat::new(completion, control.clone())
            }),
    )
});

registry.submit(spec).expect("job submitted");
registry.start("tick-counter").expect("job started");

let status = registry.status("tick-counter").expect("status");
assert_eq!(status.name, "tick-counter");

// Graceful drain: stop intake, let in-flight elements finish, then stop.
registry.drain("tick-counter").expect("drain requested");
registry.shutdown().expect("agent shuts down");

Inside the factory:

context.drain_flow() — splice it into the pipeline to support graceful drain (intake stops, in-flight elements finish). Jobs that do not return a JobControl with a kill switch are cancel-only and report drain_supported = false.
context.instrumentation_registry() + .instrumented(name, …) — opt-in per-job counters (elements, state, restarts, uptime) with zero cost when unused; this is what datum ps and the TUI read. On the 1M-element fold microbenchmark, enabling .instrumented() measured +56–79% overhead (about 1.5–2.0 ns per element). That delta includes both the relaxed per-element atomic and the loss of the inline-micro drain hint because instrumented() applies hints.without_inline_micro(). Workloads doing at least 100 ns of real work per element measured well under 2% overhead.
context.graph_metrics_bridge() — records existing GraphDSL FusedNodeMetrics reports as stable per-stage rows (stall/depth at MetricsLevel::Stalls, processing time/p99 at Timing or Stalls). Linear DSL jobs have no equivalent measurements and keep those fields unavailable.
context.generation() — increments on every restart, keeping instrument names unique.

Registry handle surface: submit, start, status, list, drain, stop, restart, shutdown, and events(). drain(name) triggers the job kill switch's graceful shutdown and waits for the stored StreamCompletion to settle; stop(name) aborts hard. Restart policies rematerialize the whole graph with backoff, advance generation, and emit lifecycle events. A panicking job never takes the registry down.

Lifecycle events

registry.events() is an ordinary Source<JobEvent> backed by Datum's own Topic<JobEvent> with TopicOverflow::Sliding — a slow subscriber can never block registry control messages. Event sequence numbers let DCP/CLI/TUI clients detect missed history and resync via list/status.

DCP — the Datum Control Protocol

DCP is a versioned, prost-encoded protocol served over the datum-net carriers:

Remote: QUIC + mTLS (rustls certificates) — the default for anything non-local.
Local dev: plaintext TCP on loopback only — the server refuses to bind a plaintext listener to a non-loopback address.

The message surface mirrors the registry (ListJobs, StartJob, DrainJob, StopJob, RestartJob, JobStatus, GetConfig/PutConfig) plus two server-streamed subscriptions: SubscribeEvents (the lifecycle feed) and SubscribeMetrics (instrumentation samples, coalesced per tick). Version negotiation happens in Hello; unknown major versions are rejected, unknown fields ignored.

StartJob uses a registered-factory model: the daemon registers named JobSpec factories at startup, and clients start instances by factory name + parameters (closures don't cross the wire — dynamic blueprint upload is deliberately deferred to cluster placement at v0.10).

A typed async client lives in datum_agent::dcp::client — it is what the datum CLI uses, and the API to reach for if you build your own tooling.

The daemon

The datum-agent binary wraps Agent::start + the DCP server with config from file/env; shutdown drains all jobs, then closes the listeners. See datum-agent --help.

From one agent to a cluster

The datum CLI and datum-tui cluster console ride on DCP, and at v0.10 the same protocol carries node-to-node control. ClusterAgent (in this crate) bundles a job registry, a DCP server, datum-cluster membership, and long-lived per-peer node sessions into one cluster-aware node — so the coordinator can place jobs across the cluster (datum submit --cluster) and datum-cluster-sharding can route entity messages between nodes. See the datum-cluster guide for membership, downing, and placement, and the M10 roadmap (roadmap) for the sequencing.

Agent Jobs, Lifecycle & DCP ​

The job model ​

Lifecycle events ​

DCP — the Datum Control Protocol ​