Skip to main content

lifecycle

Lifecycle signalling for DAG execution.

A single LifecycleNotifier signals a DAG's lifecycle transitions — accepted (start), completed, failed — to whichever destinations are wired for the run:

  • the cache Run row (marked complete / failed), when cache + run_id are supplied; and
  • the task initiator's _WorkerMailbox (proto JOB_ACCEPT / TASK_COMPLETE / TASK_ABORT), when a mailbox is supplied — used for worker_only interactive DAGs triggered by a gRPC JOB_REQUEST, where the initiator (the envelope sender) has no modeller process listening.

Construct with the destinations available for a run:

  • cache + run_id → the cache Run row is closed out;
  • mailbox → lifecycle messages are sent to the initiator;
  • neither → no-op (tests, or a run with nothing to signal);
  • both → both destinations are updated.

Each destination self-guards on being present and swallows + logs its own delivery / IO errors, so one destination failing never skips the other and a signalling hiccup never re-raises into the DAG.

Classes​

LifecycleNotifier​

class LifecycleNotifier(    *,    cache: CacheProtocol | None = None,    run_id: str | None = None,    mailbox: _LifecycleMailbox | None = None,):

Signal DAG lifecycle transitions to whichever destinations are wired.

Construct with the destinations available for a run (cache + run_id closes the cache Run row; mailbox sends proto messages to the initiator). A LifecycleNotifier() with no arguments is the no-op notifier.

Add a new lifecycle signal by adding one on_* method that touches the relevant destination(s); add a new destination by adding one optional constructor argument and branching on it inside the hooks.

Arguments

  • cache: The cache holding the Run table. Combined with run_id, enables closing out the run row.
  • run_id: The run UUID returned by mark_run_started.
  • mailbox: A _WorkerMailbox addressed to the task initiator. Enables proto JOB_ACCEPT / TASK_COMPLETE / TASK_ABORT replies.

Initialise with whichever destinations are available for the run.

Methods​

on_accept​

async def on_accept(self) ‑> None:

Signal task acceptance (proto JOB_ACCEPT) at DAG start.

Delivery failures are logged as warnings but never re-raised.

on_batches_complete​

async def on_batches_complete(self, completion_state: str = 'BATCHES_ONLY') ‑> None:

Signal all batches are complete (proto BATCHES_COMPLETE).

Message-only: no cache Run-row write. Delivery failures are logged as warnings but never re-raised.

Arguments

  • completion_state: The batches-complete body. Defaults to "BATCHES_ONLY" so a stray call never sends the "TASK_COMPLETE" variant — the terminal TASK_COMPLETE is owned by on_success.

on_configuring_task​

async def on_configuring_task(self) ‑> None:

Signal the worker is configuring the task (proto CONFIGURING_TASK).

Message-only: no cache Run-row write. Delivery failures are logged as warnings but never re-raised.

on_current_batch_id​

async def on_current_batch_id(self, batch_id: int) ‑> None:

Signal progression to a batch (proto CURRENT_BATCH_ID).

Message-only: no cache Run-row write. Delivery failures are logged as warnings but never re-raised.

Arguments

  • batch_id: The zero-based id of the batch now being processed.

on_failure​

async def on_failure(    self,    exc: BaseException,    *,    reason: Reason | None = None,    user_readable_message: str | None = None,) ‑> None:

Signal failure — mark the run failed and send TASK_ABORT.

Arguments

  • exc: The exception that caused the DAG to fail.
  • reason: Machine-readable TASK_ABORT reason. Defaults to Reason.WORKER_ERROR when unset — pass a more specific reason when the failure context is known (e.g. NO_NEW_DATA).
  • user_readable_message: Human-readable abort message. Defaults to str(exc) when unset.

on_num_batches​

async def on_num_batches(self, num_batches: int) ‑> None:

Signal the number of batches to expect (proto NUMBER_OF_BATCHES).

Message-only: no cache Run-row write. Delivery failures are logged as warnings but never re-raised.

Arguments

  • num_batches: The number of batches the initiator should expect.

on_preparing_data​

async def on_preparing_data(self) ‑> None:

Signal the worker is preparing data (proto PREPARING_DATA).

Message-only: no cache Run-row write. Delivery failures are logged as warnings but never re-raised.

on_preparing_data_batches​

async def on_preparing_data_batches(self) ‑> None:

Signal the worker is preparing data batches (PREPARING_DATA_BATCHES).

Message-only: no cache Run-row write. Delivery failures are logged as warnings but never re-raised.

on_success​

async def on_success(self) ‑> None:

Signal successful completion — close the run row, send TASK_COMPLETE.

Delivery of the TASK_COMPLETE reply is logged as a warning on failure but never re-raised.

on_task_start​

async def on_task_start(self) ‑> None:

Signal the task is about to run (proto TASK_START).

Message-only: no cache Run-row write. Delivery failures are logged as warnings but never re-raised.