Cyclium.EpisodeDrainer (Cyclium v0.4.4)

Copy Markdown View Source

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:

  1. UPDATE cyclium_episodes SET phase = 'interrupted' for every id still :running. The row stays running; Cyclium.Recovery.sweep/1 treats the phase as stale at once, and the expectation's recovery_policy decides.
  2. One UPDATE failing this node's work claims on those episodes, so the recovering node can re-acquire them without waiting for the lease.
  3. One INSERT of an :observation step per episode (%{"event" => "interrupted"}) — the marker a step feed renders as "interrupted, will resume shortly" — numbered from one MAX(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

Types

What the hand-off needs to know about one episode.

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

entry()

@type entry() :: %{
  id: String.t(),
  dedupe_key: String.t() | nil,
  actor_id: String.t(),
  conversation_id: String.t() | nil,
  workflow_instance_id: String.t() | nil,
  workflow_step_id: String.t() | nil
}

What the hand-off needs to know about one episode.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

drain_running()

@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.

entry(episode)

@spec entry(Cyclium.Schemas.Episode.t()) :: entry()

The hand-off entry for an episode.

interrupt(entries)

@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.

register(episode)

@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.

running()

@spec running() :: [{pid(), entry()}]

Every episode task running on this node, as {pid, entry}.

start_link(opts \\ [])