Struct DispatchState 

pub struct DispatchState { /* private fields */ }

Implementations

impl DispatchState

pub async fn hand_session( &self, task_id: &str, io: AdapterIo, capabilities: Vec<Capability>, ) -> Result<()>

Bind a plugin transport to one staged session and hand it the payload.

Both hand-over paths run through here: a plugin that mounted first, and a session staged first. The report that marks the session ready leaves before the assignment, which is the causal order §6 line 285 fixes.

pub async fn hand_staged(&self, session_id: &str) -> Result<bool>

Route one staged session to the thing that serves it.

A self-driven backend owns its agent and takes the payload immediately, without an adapter socket. Otherwise the session’s own connection comes first: a plugin the client spawned mounts with this session’s id in ONLYNE_SESSION_ID, and a plugin that reconnected mounts with it again, so its assignment rides that socket alone. A plugin parked for the role takes the next staged session, once. A plugin-driven session with neither waits for its mount. Answers whether the payload had somewhere to go.

pub async fn inject_note(&self, envelope: &Envelope) -> bool

Hand one note to the session already serving this role’s work.

A note carries no task, so it owns no session: §3’s note is a message to an agent that is already running, and the plan refuses one whose role is offline (note_queue off). A role with no running agent has nothing to answer it, which is what the caller reports. Answers whether an agent took the note.

impl DispatchState

pub fn enqueue_outbound(&self, envelope: &Envelope) -> Result<String>

Queue an outbound envelope before its first write and answer its op_id.

The queue keys every row by an op_id, and the proto requires that key only for the non-note kinds, so a note that arrives without one gets a fresh client-minted id here: the row is keyed and what it replays is the whole stamped envelope. A non-note keeps the id it brought, so a re-delivered task still dedups on its original one.

pub fn accept_new(&self) -> Arc<AtomicBool>

The flag the runloop and the dispatcher share.

pub fn link_up(&self) -> bool

Whether the role holds a ready server link.

pub fn set_link_up(&self, up: bool)

Record that the server link came up or went down.

pub fn cluster_ref(&self) -> String

Aggregate name this role supervises, empty for a plain role.

pub fn set_topology(&self, cluster: &str)

Record the server’s topology name, read from welcome.cluster.

Each spawned session carries it as ONLYNE_CLUSTER, which is how a host backend (herdr) addresses the tree it splits panes into. The runloop calls this on every welcome, so a server that reloads under a new name is followed by the sessions spawned after that point.

pub fn topology(&self) -> String

The topology name recorded from welcome, empty before the first welcome.

pub fn set_cluster_ref(&self, aggregate: impl Into<String>)

Record the aggregate name once, so every report keeps the same value across a reconnect.

pub async fn request(&self, op: ClientOp) -> Result<ResBody, NetError>

Ask the server one question over the live link.

Err means the link is down, never a refusal: a refusal arrives as an Ok body carrying ok: false, which is what the local CLI shows.

pub fn attach_outbox(&self, outbox: Arc<dyn Outbox>)

Install the live link as the outbound path.

pub fn detach_outbox(&self)

Remove the outbound path; lifecycle frames then queue as intents.

pub fn enqueue_op(&self, op: &ClientOp) -> Result<String>

Queue one client op in the durable intent table and answer its op_id.

impl DispatchState

pub fn reclaim_exited_resources(&self) -> Vec<String>

Retire tracked resources whose stored lifecycle has reached Exited.

The periodic readiness tick calls this after completed work becomes an idle slot. Task-free sessions with an attached transport stay bound to their host resource, and task-free sessions whose agent has left release it.

The session ids come back because the retirement wrote each one of those rows and the server mirrors only what this client reports: the resource close and, for a completed session, the agent’s exit both moved the row this tick found, and a publish cannot run under this lock. The caller is handed what to publish, the same answer DispatchState::retire_dropped_ghosts gives the sweep above.

pub fn retire_dropped_ghosts( &self, now: Instant, grace_secs: u64, ) -> Vec<Retired>

Retire the sessions whose plugin connection dropped and never came back, or whose connection stayed up while they went quiet, and answer which ones left and why.

A connection that ends without a detach frame leaves its session tracked so an agent that restarts inside [client] reconnect_grace_secs finds the resource it was using. That promise has to expire: a process that is simply gone would otherwise hold a slot, a projected idle row, and a live host resource forever, and on a role with max_sessions = 1 it stops every later delivery. The window answers for the agent itself, so a session still bound to a task goes with it: the plugin connection that would have reported the ending is the one that dropped. The agent-gone feed is what says the process left — the session’s own tuple reaches Exited through AgentState::Gone rather than through a task result — and the reason the backend is handed is the one grace_close_reason reads off what the slot still owes.

What the slot owed is settled too: the task a bound session was serving ends failed here, because the agent that would have reported its ending is the one that left. A task with no verdict stays open for the server to re-offer and for open_tasks to keep reading, and no later caller exists to write one.

A slot this client holds read-only is not this sweep’s to end, agent gone or not: the session id it would feed is the task id, so the ghost’s death would take the live session’s mirror and its delivery row down with it. That retirement belongs to retire_revived, which runs when the completion that answers the held connection merges.

The window has a second way to open, and it is the one a socket cannot report: a plugin whose event loop is blocked keeps its connection and stops beating, so no socket ends and no clock this sweep could read before moved. What such a session leaves behind is a stamp going stale while its task stays bound and unsettled, and that is the reading this sweep takes now. It is the same window and the same verdict — one clock, one retirement, no second threshold beside [client] reconnect_grace_secs and no fault row of the kind stall_report_secs records and leaves behind.

The sessions’ own ids come back rather than a count, because a retirement still owes the server the session’s own ending: it is the only writer left for that task, and a mirror nobody tells keeps that session’s last reading — working, for one that had beaten — until the server’s own observer records a fault about it. The publish is sync_session’s, which is the report an ordinary ending travels on, and it cannot run under this lock — so the caller is handed what to publish instead of a second writer being invented here.

impl DispatchState

pub fn plugin_handoff(&self, io: &AdapterIo, args: HandoffArgs) -> ResBody

Take one plugin handoff frame and answer what the plugin is told.

The frame names the task the session is handing on and the role it goes to. The child is minted here, through the builder the report-driven path uses, so the family id and the family’s figures ride along and the depth grows by one hop. The envelope leaves on the queue the plugin send op writes to.

The answer names the child: {"task_id": "<uuid>", "hop": 3, "queued": true, "op_id": "<uuid>"} (onlyne_proto::HandoffArgs).

A frame is answered only for the connection serving the task it names. An unknown task and a foreign connection earn the same code and the same field, and their messages say which of the two refused the frame.

impl DispatchState

pub fn new( role: impl Into<String>, workspace: impl Into<PathBuf>, command: Vec<String>, max_sessions: u32, backend: Arc<dyn SessionBackend>, store: ClientStore, ) -> Self

pub fn session_count(&self) -> usize

pub fn role(&self) -> String

pub fn command(&self) -> Vec<String>

pub fn backend_name(&self, task_id: &str) -> Option<String>

pub fn outcome_feed(&self) -> Option<OutcomeFeed>

The terminal-fact stream of a backend that drives its own agent.

pub fn push_assign_ack(&self, ack: AssignAckArgs) -> bool

Queue a delivery ack when the plugin refuses an assignment.

An accepted assignment is not terminal for the server row: the normal completion path still settles that delivery. A refused assignment is a terminal local decision, so it uses the same durable ack queue as every other delivery settlement.

pub fn push_settled(&self, ack: AckArgs)

Queue an ack the client owes the server.

Record an ack the client owes the server.

The ack is durable: D11’s control plane is at-least-once, and a settled session whose ack is lost leaves the row in flight forever. The intent queue carries it across a link that is down, and the flusher is the sender.

pub fn owe_controlled_settle( &self, task_id: &str, word: ControlWord, now: Instant, )

Note that one task’s plugin has been told by this client’s own control command to report the ending of that task.

on_control runs this before the recycle frame leaves, which is the point where the client still knows the order: the plugin’s completion and the retirement that command triggers race over the session’s row, and a guard that read the row would answer the same operator action two ways. One note per task is kept, so a command issued twice waits for one answer.

word is what the operator said and now is when they said it, and both are the caller’s to name rather than this call’s to invent: the command is the authority on the ending it asked for, and the instant it reads is the one the watchdog’s bound runs from.

pub fn take_controlled_settle(&self, task_id: &str) -> bool

Whether one completion answers a command noted above, consuming the note.

The note is spent whichever way the settle it authorises goes: a refused verdict leaves no second answer owed, and an applied one has travelled the command’s own completion. A later report for the same task is the plugin speaking for itself again, and reads the ordinary door.

pub fn control_settles_due(&self, now: Instant) -> Vec<ControlNote>

The notes whose operator’s word has gone unanswered past the bound.

The reading a sweep takes before it acts, and it spends nothing: the settle below goes through take_controlled_settle, so a completion that answers a word between this read and that call takes the note first.

pub fn settle_unanswered_control(&self, note: &ControlNote) -> bool

Settle the work one operator’s word left open, with no report behind it.

The word asks a plugin for its own ending and the completion that answers it is a frame of the plugin’s, so a plugin that never sends one — it left with the command’s frame, or implements no recycle at all — leaves the task open and the delivery row this client was handed in flight, with no later caller to answer either. This is that caller.

The writes are the ones retire_dropped_ghosts makes for the task its owed session left: the verdict through the task’s own record, which refuses to overwrite one that landed first, and the still-held delivery row refused with the operator’s word, which is terminal for that row the way every refusal is. The publish is the caller’s, because it cannot run under this lock.

Answers false when this call is not the settle: the note is already spent by a completion that answered the word, or the task’s record carries a verdict already, and either way nothing here is written and nothing is for the caller to publish.

pub fn role_slice(&self) -> RoleSlice

The role slice the dispatcher currently runs.

pub fn live_task_ids(&self) -> HashSet<String>

Task ids currently occupying a live slot.

pub fn hello_live_tasks(&self) -> Vec<String>

Sorted live-slot task ids for a hello claim.

pub fn note_stall_assigned(&self, task_id: &str, now: Instant)

Start the stall clock for a newly assigned task.

pub fn note_stall_applied(&self, task_id: &str, now: Instant)

Refresh the stall clock after an Applied persist.

pub fn stall_due(&self, now: Instant, threshold_secs: u64) -> Vec<String>

Task ids whose freeze exceeds threshold_secs in this episode. Exited projections retire their remaining progress clocks.

pub fn mark_stalled(&self, task_id: &str)

Remember that this freeze episode has been reported.

pub fn stall_report(&self, task_id: &str) -> Option<Report>

Observation-only stall fault for an active task, carrying the stored watermark. A tuple and verdict that derive exited retire their progress clock before the send boundary.

pub fn has_mounted_adapter(&self) -> bool

Whether any adapter is currently mounted (named or parked).

pub fn reconfigure(&self, slice: RoleSlice)

Adopt a role slice: the one welcome carried, or the one a reload’s role row carries.

pub fn role_prose(&self) -> String

Role prose last cached from welcome.

pub fn holds_task(&self, task_id: &str) -> bool

Whether one task names a session this client holds, in memory or in its durable session rows.

pub fn session_generation(&self, task_id: &str) -> Option<u64>

Generation the reducer holds for one task, before any hand-off.

pub fn has_capacity(&self) -> bool

Whether a delivery has somewhere to run.

§5’s max_sessions caps concurrency, so a delivery that arrives at the cap waits on the server: the row stays in flight and the next pull offers it again once a session frees. Each task runs in its own session, so a slot whose task has finished still spends capacity until it retires.

pub fn task_completed_here(&self, task_id: &str) -> bool

Whether this role already finished one task with a terminal Done.

The task’s own record is the account: the settle writes the verdict the agent filed and refuses to overwrite it, so a record reading done means this role answered for this task id once already. A redelivery of that task is not new work — running it again would stage its payload on whichever session happens to be idle, so one chain’s task executes inside another conversation and the second answer collides with the verdict the first one settled.

Only done counts. A session killed or crashed mid-flight leaves its task open, or settles it failed, and the server’s requeue, repair_retry, and control retry all re-offer that task on purpose, so those deliveries still run.

pub fn staged_without_transport(&self) -> Option<String>

The task of one session that holds a payload with no connection bound.

A work item that arrives before its always-running agent mounts waits in exactly this state, and the mount ends the wait. A session answers through the transport its first task claimed, so it stays served.

impl DispatchState

pub fn hold_frame(&self, io: &AdapterIo) -> FrameGuard<'_>

Hold io for as long as one of its inbound frames is being handled.

pub fn bind_adapter( &self, session_id: &str, io: AdapterIo, capabilities: Vec<Capability>, )

Bind one adapter connection to the session it named.

The name is the session id the client spawned the plugin with, which is enough on its own: a plugin that mounts before the client staged its session is remembered here and takes the payload the moment it is staged, and a plugin that mounts after finds its session waiting.

pub fn attach_msg_id(&self, task_id: &str, msg_id: &str)

Remember the delivery handle for one task.

The handle goes to the session serving the task, not to a read-only slot that came back for it, so the ack this earns answers the live delivery.

pub fn plugin_send(&self, io: &AdapterIo, envelope: &Envelope) -> Result<Value>

Take one plugin send frame and answer what the plugin is told.

A live connection’s envelope goes to the durable outbound queue exactly as it always has, and the answer keeps the shape the plugin reads. A frame from a connection this client holds read-only is held instead (§1 (c)): it leaves as part of the merged handoff its task’s completion routes, so the recipient sees one message per downstream role and can tell which session wrote which half of it.

pub fn park_transport(&self, io: AdapterIo, capabilities: Vec<Capability>)

Park one plugin connection as this role’s waiting agent.

Only a mount that names no session parks: it is a plugin that attached before any work existed, so it takes the next session this role stages (plan §6 line 285).

pub fn session_transport( &self, session_id: &str, ) -> Option<(AdapterIo, Vec<Capability>)>

The connection that serves one session, when its plugin is attached.

A plugin names the session it was spawned for, and the slot’s key is the other spelling worth trying.

pub async fn recycle_plugin( &self, task_id: &str, reason: &str, outcome: Option<Outcome>, )

Tell the plugin serving one session to tear itself down, when that plugin implements recycle. A plugin without the capability is skipped: the caller’s backend close stops the process either way.

pub async fn probe_plugin(&self, task_id: &str) -> bool

Ask the plugin serving one session for a fresh observation.

The plugin answers with a heartbeat report, which is the reducer’s evidence and the projection the operator reads. Answers whether a probe frame actually went out, and false says nothing was asked: a session no connection serves has no plugin to put the question to, and a notify that failed left the frame in this process. The caller must not record either as a probe that landed, because the answer a live plugin would have given never existed — the projection that says a plugin is gone is written when the session ends, never by this call.

pub fn release_connection( &self, session_id: Option<&str>, io: &AdapterIo, graceful_detach: bool, ) -> Vec<String>

Release the bindings served by one plugin connection.

A graceful detach retires each idle session because the agent that owned it has left. An attached transport preserves the idle resource because the same agent is still reachable. A connection ending through another path preserves the slot and resource for an agent reconnection and starts the reconnect clock on it, which is what bounds how long a session waits for an agent that is never coming back. Every released binding retires its task progress clock. A slot carrying work remains under lifecycle ownership, and it carries that same clock: a goodbye and a silent drop both leave no heartbeat coming for the task it owes, and the window is what ends a session whose agent never returns.

The session ids a goodbye retired come back, because that retirement wrote their rows: the resource close and, for a completed session, the agent’s exit. The server mirrors only what this client reports, and a publish cannot run under this lock, so the caller is handed what to publish — the same answer DispatchState::retire_dropped_ghosts gives its sweep.

Trait Implementations

impl Clone for DispatchState

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.