Hands this node's in-flight episodes to recovery, in bulk, when the node shuts down.
A graceful stop (a deploy's SIGTERM) gives the VM a few seconds. Left to
themselves, every running Cyclium.EpisodeTask would notice its shutdown
EXIT at its next step boundary and write its own hand-off — a status
update, a claim release, a journal step — a dozen episodes contending for a
connection pool that is itself draining, with the ones still mid-step
(usually an LLM call) getting no chance at all before the task supervisor
kills them.
So the hand-off is done once, here, for all of them. This process sits in
Cyclium.Supervisor after Cyclium.EpisodeSupervisor, which means it is
terminated before the episode tasks are (a supervisor stops children in
reverse start order). Its terminate/2 reads the running episodes off
Cyclium.EpisodeProcessRegistry — every task registers itself once it holds
its claim — and issues three statements for the lot:
UPDATE cyclium_episodes SET phase = 'interrupted'for every id still:running. The row stays running;Cyclium.Recovery.sweep/1treats the phase as stale at once, and the expectation'srecovery_policydecides.- One
UPDATEfailing this node's work claims on those episodes, so the recovering node can re-acquire them without waiting for the lease. - One
INSERTof an:observationstep per episode (%{"event" => "interrupted"}) — the marker a step feed renders as "interrupted, will resume shortly" — numbered from oneMAX(step_no)query over all of them.
Then it broadcasts episode.step_journaled and episode.interrupted per
episode (in-memory, no I/O) and sends each task :cyclium_drained, which the
runner's loop takes at its next step boundary as "already handed off — stop,
write nothing". A task killed mid-step never reaches that boundary, and does
not need to: its row was already marked in step 1.
Each statement is best-effort: a failure is logged and the next one still runs, so a pool timeout on the journal insert cannot cost the phase stamp that recovery depends on.
interrupt/1 is the same hand-off for an explicit list, used by the runner
for a single episode when it receives a shutdown EXIT with no drainer having
run first (a host that embeds the runner without Cyclium.Supervisor).
Summary
Functions
Returns a specification to start this module under a supervisor.
Hand every running episode on this node to recovery and tell its task to
stop. Returns {:ok, count}. Public so a host with its own shutdown hook
can call it earlier than the supervisor would.
The hand-off entry for an episode.
The hand-off for a given list of episodes: phase "interrupted", claims
released, one journal marker each, then the broadcasts. Three statements
however long the list. Does not notify the tasks — drain_running/0 does.
Register the calling episode task as running episode, so a drain can find
it. Called by Cyclium.EpisodeTask once it holds its claim; the entry goes
away with the process. A no-op when the registry is not running.
Every episode task running on this node, as {pid, entry}.
Types
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec drain_running() :: {:ok, non_neg_integer()}
Hand every running episode on this node to recovery and tell its task to
stop. Returns {:ok, count}. Public so a host with its own shutdown hook
can call it earlier than the supervisor would.
@spec entry(Cyclium.Schemas.Episode.t()) :: entry()
The hand-off entry for an episode.
@spec interrupt([entry()]) :: {:ok, non_neg_integer()}
The hand-off for a given list of episodes: phase "interrupted", claims
released, one journal marker each, then the broadcasts. Three statements
however long the list. Does not notify the tasks — drain_running/0 does.
@spec register(Cyclium.Schemas.Episode.t()) :: :ok
Register the calling episode task as running episode, so a drain can find
it. Called by Cyclium.EpisodeTask once it holds its claim; the entry goes
away with the process. A no-op when the registry is not running.
Every episode task running on this node, as {pid, entry}.