Struct HeartbeatTracker 

pub struct HeartbeatTracker { /* private fields */ }

Per-task liveness tracker for remote-worker streams.

It is also the server’s ONLY store of worker progress notes, and that store is VOLATILE: notes live here and nowhere else, so a restart loses every note ever reported. Self::notes_held_since is the evidence a reader needs to tell “this worker has said nothing” from “this process was not running when the attempt began” — see crate::worker::attempt_progress.

Implementations

impl HeartbeatTracker

pub fn expired_workers( &self, registry: &ConnectedWorkerRegistry, now: Instant, ) -> Result<Vec<WorkerId>, ServerError>

Return the workers this sweep will DEREGISTER: those whose silence is corroborated by a push leg that is no longer open.

One clock nominates; two positive facts acquit

This read used to consult two silence clocks — the connection clock and the PER-TASK beat clock — and NOTHING about whether the server still held a channel to the worker. Both inputs are facts about ABSENCE, so neither could tell a worker that is gone from one that is busy, and a worker running at its full concurrency looked identical to a dead one. It was then deregistered while connected and holding live work, its in-flight activities were surfaced as transport losses, and the queue census — with the worker gone from the pool — reported that no worker was connected about a worker on an open stream.

Two things changed. The per-task clock no longer nominates anybody: is_expired is a predicate about ONE activity and was being used to condemn a whole process, so a single lapsed beat took down every other activity that worker was holding.

What now resolves a lapsed per-task beat — and what does not

Stated exactly, because the honest answer is narrower than “something else handles it” and an earlier draft of this doc promised a mechanism “below” that does not exist.

Three things still end a stuck activity, none of them this sweep:

• the CONNECTION clock, when the worker itself also goes quiet — this function nominates it, and fail_expired_workers then fails every activity it was holding;
• the engine seam’s authored per-attempt bound, measured from the lease (aion::LeaseSignal), where an operator authored one;
• the outbox row’s own attempt budget, backoff and dead-letter, once the dispatch call returns.

What is left uncovered, and is a real gap rather than a tidy hand-off: an activity whose beat lapses while its worker KEEPS answering. The worker stays in service, correctly — it is demonstrably alive — and no mechanism fails that single activity. is_expired now reaches production through nothing at all: its only reader is HeartbeatTracker::is_live, which has no production caller. Before this change the whole-worker reap was what (brutally, and wrongly) covered that case.

This is recorded rather than fixed here because inventing a per-activity deadline is a semantics decision about how long work may run without reporting, and that belongs to whoever owns the activity timeout — not to a sweep that was being asked to do it as a side effect. It is tracked as a server-lane issue; the number is deliberately not written here, because an issue reference in a doc comment rots the moment the issue moves.

And the surviving clock alone no longer condemns anybody either. Silence is the SUSPICION; the verdict needs a second, positive fact to survive, and the worker keeps its registration unless BOTH of these hold:

• The push leg is still open. [WorkerHandle::is_connected] asks the transport, not the noise floor: gRPC answers from its stream sender, liminal from its supervisor’s connection table. A worker with no leg left has nothing that could carry a beat, and reaping it is the honest reading of both facts together.
• The server has PROVED it can reach that leg. An open channel is not an answering worker — the server can hold a perfectly healthy looking mpsc to a wedged process indefinitely, which is the whole reason the liveness ping exists. So a connected worker is spared only while its dispatch path is proved reachable.

Why BOTH, and what each one alone would cost

Connection alone is not enough: on gRPC a sender closes only when the stream task ends, and that is already the teardown sweep’s business. A reaper gated on connection alone would therefore never fire for a gRPC worker at all, and a WEDGED process — stream open, nothing moving, in-flight work that will never complete — would hold its activities for ever. That is the failure #176 was written to end, and it must not be re-opened by fixing a different one. The connection clock is what catches it: a wedged process performs neither of the two acts that refresh it, so it is nominated within one window with its reachability unproved. That is why dropping the per-task arm costs #176 nothing.

Reachability alone is not enough either: it is a measurement with a cadence, and a worker mid-probation or one probe behind has not yet proved anything about a link that is fine.

The reachability conjunct is currently DEFENSIVE, and here is why

Say this rather than let a future reader assume it is load-bearing. Every call that proves reachability — record_dispatch_reachability — also refreshes the connection clock, and so do record_heartbeat and record_liveness. Since the only nomination is a lapsed connection clock, a worker that reaches this gate has by construction not been proved recently either: the two clocks are advanced by the same calls, so “overdue but provably reachable” is not a state this tracker can be in today. The conjunct therefore cannot currently spare anybody the connection arm would have condemned.

It stays because it is the correct predicate and it is free. If a future change proves reachability without touching the connection clock — a side-channel probe, a transport that answers without traffic — this gate is already the right one, and its absence would be a silent re-opening of the reap-a-busy-worker hole rather than a visible one.

Together they say what the sweep actually needs: this worker’s leg is open AND the server has recently pushed something down it that came back, so its silence is BUSY, not gone. That is exactly the state a worker at its full concurrency is now in — the receive loop no longer parks on admission, so a saturated worker answers pings while it works — and it is exactly the state the old code could not see.

What a connected, unreachable worker gets first

Dispatch INELIGIBILITY, published by the liveness probe on this same evidence and several probe cadences before this sweep can fire: the worker stays registered, keeps its in-flight work, and is sent nothing new. This method neither reads nor re-derives that verdict — one reaper, one eligibility verdict, and each reaches its own conclusion from the evidence it owns.

A read failure on either input yields the reaping answer for that worker, which is the conservative direction: it can only fall back to the behaviour this method had before it could ask at all, never invent a reap the silence clocks did not already ask for.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

impl HeartbeatTracker

pub fn new(heartbeat_window: Duration) -> Self

Build a tracker using the operator-supplied heartbeat window.

pub const fn notes_held_since(&self) -> DateTime<Utc>

When this tracker began holding progress notes.

An attempt dispatched before this instant left its notes in a process that no longer exists, so their absence here says nothing about what the worker reported.

pub fn register_connection( &self, worker_id: WorkerId, now: Instant, ) -> Result<(), ServerError>

Start the connection-level lease for a newly registered worker, and open its dispatch probation.

The connection lease starts fresh — the worker’s process is plainly alive, it just registered. Dispatch reachability does NOT: a new connection is a channel to prove reachability on, not proof of it.

This deliberately reverses an earlier reading of mine, that the registration handshake is “itself a completed server-to-worker round trip”. The ack is SENT by the server; nothing reports that it was RECEIVED. Treating a send as a delivery is the same inference this lane exists to stop making, and left unfixed it meant a worker the server could never reach would re-seed itself on every redial and cycle in and out of eligibility forever instead of settling out.

See [DISPATCH_PROBATION_PINGS].

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn record_connection_activity( &self, worker_id: WorkerId, now: Instant, ) -> Result<bool, ServerError>

Advance a worker’s connection lease after receiving any frame.

Records ONLY that the worker’s process is alive. It deliberately does NOT advance dispatch reachability: an inbound frame — a heartbeat, a pump beat, a completion — proves the worker-to-server direction and says nothing about whether the server can push to it. Use Self::record_dispatch_reachability for the fact that gates dispatch.

Returns false if the worker has already been removed from lease tracking; a frame racing deregistration must not resurrect it.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn record_dispatch_reachability( &self, worker_id: WorkerId, now: Instant, ) -> Result<bool, ServerError>

Record proof that the server can reach this worker’s dispatch path — an ANSWERED liveness ping, and nothing else.

Advances both facts, because an answered ping proves both: the worker received a server push (reachability) and replied to it (alive). It also serves one ping of the dispatch probation; eligibility returns once [DISPATCH_PROBATION_PINGS] consecutive answers have landed.

Returns false if the worker has already been removed from lease tracking; a pong racing a reap must not resurrect it.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn record_dispatch_unreachable( &self, worker_id: WorkerId, ) -> Result<bool, ServerError>

Record that a liveness ping went UNANSWERED: the probation restarts.

This is what makes the probation consecutive rather than cumulative. A link answering one probe in three would otherwise accumulate its way to eligibility and keep it, which is the flapping this design removes.

Returns false if the worker has already been removed from lease tracking.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn is_dispatch_reachable( &self, worker_id: WorkerId, now: Instant, ) -> Result<bool, ServerError>

Whether the server has proved, within the heartbeat window, that it can reach this worker’s dispatch path — probation served AND the proof still fresh.

An untracked worker is not reachable: absence of proof is not proof.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn unreachable_workers( &self, now: Instant, ) -> Result<Vec<ExcludedWorker>, ServerError>

Every tracked worker the server has NOT been able to reach within the heartbeat window, regardless of how alive its process looks, each paired with WHY it is excluded.

The reason travels with the membership rather than being recomputed by the caller, so the set that gates dispatch and the reason an operator is told can never describe different states.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn unregister_connection( &self, worker_id: WorkerId, ) -> Result<(), ServerError>

End connection-lease tracking when a transport closes normally.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn track_task( &self, worker_id: WorkerId, task: InFlightActivity, now: Instant, registry: &ConnectedWorkerRegistry, reservation: Option<DispatchReservation>, ) -> Result<(), ServerError>

Track a newly accepted in-flight activity for heartbeat expiry.

The reservation, and why it is taken by value

reservation is the capacity slot the caller claimed at selection. Passing it here is how the slot is HANDED OVER rather than re-counted: this call commits it and skips its own increment, so the worker’s in_flight goes 1 → 1 instead of 1 → 2 → 1.

That intermediate value was not harmless. The tracker incremented under one registry lock and the reservation released under another, and a concurrent leg landing between them was refused a slot the worker demonstrably had — measured at in_flight = 5 on a worker advertising 4, on a fan sized exactly to its pool.

Taken BY VALUE so the transfer cannot be half-done: this call either commits the reservation or drops it, and a dropped one still releases. None is for callers that never reserved — the in-process façades, and any path tracking work it did not select.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn complete_task( &self, worker_id: WorkerId, workflow_id: &WorkflowId, activity_id: &ActivityId, registry: &ConnectedWorkerRegistry, ) -> Result<bool, ServerError>

Stop tracking a completed activity and wake drain waiters if this was the last task.

Returns whether the task was still tracked when this ran: true means THIS call retired the in-flight entry, false means another path (the expiry sweep, a disconnect teardown, shutdown, or a completed dispatch) already did. The liminal reply router uses that bool as its structural gate for synthesizing a lost-worker failure — the exact mirror of the gRPC sweep failing only still-tracked tasks.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn is_tracked( &self, worker_id: WorkerId, workflow_id: &WorkflowId, activity_id: &ActivityId, ) -> Result<bool, ServerError>

Whether the given in-flight task is still tracked (not yet completed, swept, or drained). The liminal reply router polls this to bound its wait: once the entry is gone the dispatch was resolved by another path, so the router exits instead of parking on the connection forever.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn record_liveness( &self, worker_id: WorkerId, workflow_id: &WorkflowId, activity_id: &ActivityId, now: Instant, ) -> Result<bool, ServerError>

Refresh the liveness stamp of an in-flight task from a transport-level liveness beat that carries no progress payload (the liminal worker’s automatic pump). Returns true when the task was tracked and refreshed, false when it is not in flight — a benign outcome for a beat racing a completion or covering an outbox dispatch the tracker never held.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub const fn heartbeat_window(&self) -> Duration

The operator-configured heartbeat window this tracker expires against. The bridge stamps it onto each liminal dispatch so the worker’s automatic liveness pump beats at the matching quarter-window cadence.

pub fn in_flight_count(&self) -> Result<usize, ServerError>

Number of currently tracked in-flight activities.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn in_flight_for_worker( &self, worker_id: WorkerId, ) -> Result<usize, ServerError>

Number of in-flight activities tracked for ONE worker.

Beside Self::in_flight_count rather than replacing it: that answer is deliberately global (the drain gate asks “is anything still running”), and widening it to mean per-worker would make every existing caller read a different question than the one it asked.

This is the tracker’s own count, derived from the task map it owns. The registry keeps a projection of the same fact for selection to read without taking this lock, and the two are written by one call (Self::track_task / Self::complete_task). This accessor is what lets a test hold them against each other instead of trusting that.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

pub fn record_heartbeat( &self, worker_id: WorkerId, heartbeat: ProtoHeartbeat, now: Instant, ) -> Result<HeartbeatUpdate, ServerError>

Record a worker heartbeat without completing the activity.

Every heartbeat refreshes the task’s liveness stamp. The progress payload is only overwritten when the heartbeat CARRIES one: the worker runtime’s automatic liveness beats are payload-free and interleave with explicit handler progress heartbeats, and a liveness beat must never erase the handler’s most recent progress report.

Errors

Returns a stable wire error for malformed heartbeats or unknown in-flight tasks.

pub fn is_live( &self, worker_id: WorkerId, workflow_id: &WorkflowId, activity_id: &ActivityId, now: Instant, ) -> Result<bool, ServerError>

Return whether an in-flight task is still within its configured heartbeat window.

Errors

Returns a stable wire error if the task is not tracked, or lock poison if state cannot be trusted.

pub fn fail_expired_workers( &self, registry: &ConnectedWorkerRegistry, sink: &impl ActivityCompletionSink, now: Instant, ) -> Result<Vec<LostWorkerReport>, ServerError>

Mark all currently expired workers lost and fail their in-flight tasks through the engine sink.

Errors

Returns registry, tracker, or sink errors without retrying or rescheduling activities.

pub fn fail_disconnected_worker( &self, worker_id: WorkerId, registry: &ConnectedWorkerRegistry, sink: &impl ActivityCompletionSink, ) -> Result<LostWorkerReport, ServerError>

Mark a disconnected worker lost and fail its in-flight tasks through the engine sink.

Errors

Returns registry, tracker, or sink errors without retrying or rescheduling activities.

pub fn fail_all_in_flight_workers( &self, registry: &ConnectedWorkerRegistry, sink: &impl ActivityCompletionSink, ) -> Result<Vec<LostWorkerReport>, ServerError>

Mark every currently in-flight worker lost and fail all remaining tasks through the sink.

Errors

Returns registry, tracker, or sink errors without retrying or rescheduling activities.

pub fn park_disconnected_worker( &self, worker_id: WorkerId, registry: &ConnectedWorkerRegistry, sink: &impl ActivityCompletionSink, ) -> Result<LostWorkerReport, ServerError>

Park a drain-disconnected worker’s in-flight tasks for restart recovery (#207): deregister the worker, remove its tracked tasks, and resolve each pending waiter through ActivityCompletionSink::park_activity.

The graceful-drain counterpart of Self::fail_disconnected_worker: same deregister-before-collect ordering (same closed dispatch/disconnect race), but NO completion is synthesized — the durable log keeps its dangling scheduled/started trail, byte-equivalent to a kill -9, and restart recovery re-dispatches it. Deregistered with the honest WorkerDeathReason::Disconnect: the transport genuinely dropped (the worker obeyed the drain request).

Errors

Returns registry, tracker, or sink errors without retrying or rescheduling activities.

pub fn park_all_in_flight_workers( &self, registry: &ConnectedWorkerRegistry, sink: &impl ActivityCompletionSink, ) -> Result<Vec<LostWorkerReport>, ServerError>

Park EVERY currently in-flight worker’s tasks for restart recovery (#207) — the drain-timeout backstop’s bulk counterpart of Self::fail_all_in_flight_workers.

Deregistered with the honest WorkerDeathReason::Timeout: the drain window genuinely expired on these workers. Wakes drain waiters after the sweep so wait_for_empty observes the emptied tracker.

Errors

Returns registry, tracker, or sink errors without retrying or rescheduling activities.

pub fn in_flight_for_workflow( &self, workflow_id: &WorkflowId, ) -> Result<Vec<TaskLiveness>, ServerError>

Every activity of workflow_id a connected worker is holding RIGHT NOW (#233).

A tracked entry means a live worker owns that activity: entries are added at dispatch (Self::track_task) and removed on completion, disconnect, or heartbeat expiry. That is what makes this — rather than the durable outbox row, which records no worker at all — the authority on where a cancellation has to be sent. The worker id here is present membership, not a recorded historical fact, so it cannot go stale the way a worker id written down at claim time would.

A within-attempt failover can briefly show a dying owner AND its adopter for one activity. Both are returned: a cancel is addressed to a (workflow, activity) key that either worker may or may not still hold, and a worker that does not hold it ignores the message, so asking both is correct and asking one would be a guess.

Errors

Returns ServerError::LockPoisoned if tracker state cannot be trusted.

Trait Implementations

impl Clone for HeartbeatTracker

fn clone(&self) -> HeartbeatTracker

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for HeartbeatTracker

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.