API reference¶

class Terminal(EverystepError):
    """Raised from a step to stop the workflow for a known reason.

    The step in flight finishes and is recorded; no new step starts and the
    engine will not retry the workflow. It ends in the `stopped` status with
    the reason and payload recorded on it, so the caller can take over (for
    example, re-scheduling with a different resource). Not a failure: it is
    never reported to Sentry.
    """

    def __init__(self, reason, payload=None):
        self.reason = reason
        self.payload = payload
        super().__init__(reason, payload)

    def __str__(self):
        return str(self.reason)

class Context:
    """Execution state for one workflow run, or one branch of one fork.

    Steps are identified by a dotpath like "3" or "3.1.2". Unnamed steps
    take the next position in their scope; a step called with `everystep_id`
    takes that name as its segment instead, which makes its id stable
    against edits elsewhere in the body. Names must be unique within a
    scope (one body, or one branch). The shared `outcomes` dict maps step
    ids to `{name, status, result, error}` outcomes. It is seeded with all
    outcomes recorded before the current claim and updated as each step
    completes, so inside a step you can read any previously-completed
    step's outcome from the current context. Treat it as read-only.
    """

    def __init__(self, workflow_id, outcomes, persistent, prefix="", draining=None, workflow_name=None):
        self.workflow_id = workflow_id
        self.workflow_name = workflow_name
        self.outcomes = outcomes
        self.persistent = persistent
        self.prefix = prefix
        # Event set by the runner on stop; checked before each new step.
        self.draining = draining
        self.counter = itertools.count(1)
        self._used_ids = set()

    def next_id(self, name=None):
        # Always consume a position, so naming (or un-naming) a step never
        # shifts the ids of the other unnamed steps in the scope.
        tick = next(self.counter)
        if name is None:
            segment = str(tick)
        else:
            _validate_name(name)
            segment = name
        step_id = f"{self.prefix}{segment}"
        if len(step_id) > _MAX_ID_LENGTH:
            raise EverystepError(f"everystep_id {name!r} is too long: step id would exceed {_MAX_ID_LENGTH} characters")
        if step_id in self._used_ids:
            raise EverystepError(
                f"everystep_id {segment!r} is already used in this scope (step id "
                f"{step_id!r}); everystep_ids must be unique per body or branch"
            )
        self._used_ids.add(step_id)
        return step_id

    def branch(self, fork_id, index):
        return Context(
            workflow_id=self.workflow_id,
            outcomes=self.outcomes,
            persistent=self.persistent,
            prefix=f"{fork_id}.{index}.",
            draining=self.draining,
            workflow_name=self.workflow_name,
        )

class EverystepError(Exception):
    """Base class for everystep errors."""

class WorkflowCodeError(EverystepError):
    """A workflow body diverged from its previously recorded step identities."""

class EffectUncertain(EverystepError):
    """A step marked unsafe to repeat may have performed its effect.

    The step's outcome was never recorded (the worker died in the effect
    window, or another claimant holds it), so the engine cannot tell whether
    the effect happened. It refuses to execute the step again and ends the
    run in the `blocked` status — a holding state, not a failure: it is
    never reported to Sentry. The run is resolved by a human with the
    `everystep_resolve_step` management command.
    """

class StepFailure(EverystepError):
    """A recorded step failure whose original exception type is unavailable."""

class SimulatedCrash(EverystepError):
    """Raised from a test fault handler to simulate a process death mid-step.

    The runner re-raises it without writing any further state, leaving the
    workflow exactly as it would be if the worker had died.
    """

class DrainOrphan(EverystepError):
    """Raised at a step boundary when the runner is draining after a stop signal.

    The step in flight at the signal finishes and is recorded; no new step
    starts. The worker requeues the workflow so any runner can claim it and
    resume it from the recorded steps.
    """

class Worker:
    """Claim due workflows and execute them on a thread pool.

    See the running section of the documentation for the claim loop, the
    name contract, and the SIGTERM drain behavior.
    """

    def __init__(
        self, pool_size=4, poll=0.2, name=None, drain=30, metrics_port=0, metrics_bind="0.0.0.0"
    ):
        self.pool_size = pool_size
        self.poll = poll
        self.name = name or socket.gethostname()
        self.drain = drain
        self.metrics_port = metrics_port
        self.metrics_bind = metrics_bind
        self._stop = threading.Event()
        self._draining = threading.Event()
        self._active = {}
        self._executor = ThreadPoolExecutor(max_workers=pool_size, thread_name_prefix="everystep-w")
        self._metrics_server = None

    def stop(self):
        self._stop.set()
        self._draining.set()

    def run(self):
        _install_signal_handlers(self)
        try:
            metrics.worker_started(self.name, self.pool_size)
            self._start_metrics_server()
            self._catchup()
            while not self._stop.is_set():
                self._reap()
                capacity = self.pool_size - len(self._active)
                if capacity > 0:
                    claimed = claim_new(capacity, self.name)
                    for workflow in claimed:
                        self._active[self._executor.submit(self._execute, workflow)] = workflow
                    metrics.record_claims(self.name, len(claimed))
                metrics.set_inflight(self.name, len(self._active))
                self._stop.wait(self.poll)
        finally:
            leftovers = self._drain()
            if leftovers:
                self._abandon(leftovers)
                if threading.current_thread() is threading.main_thread():
                    # Standalone worker: exit now rather than let the
                    # interpreter join the abandoned pool threads at
                    # shutdown. An embedded worker just returns; the host
                    # process owns its own lifecycle.
                    os._exit(0)
            metrics.set_inflight(self.name, 0)
            if self._metrics_server is not None:
                self._metrics_server.server_close()
            connections.close_all()

    def _start_metrics_server(self):
        if not self.metrics_port:
            return
        self._metrics_server = metrics.start_http_server(self.metrics_port, self.metrics_bind)

    def _drain(self):
        """Stop queued work and wait up to `self.drain` seconds for the
        in-flight workflows to finish. Returns the futures still running
        when the deadline is hit. With drain of 0, waits indefinitely and
        returns nothing."""
        if not self.drain or not self._active:
            self._executor.shutdown(wait=True)
            return []
        self._executor.shutdown(wait=False, cancel_futures=True)
        _, not_done = wait(list(self._active), timeout=self.drain)
        return list(not_done)

    def _abandon(self, leftovers):
        """Requeue the runs still in flight when the drain deadline expired,
        so nothing is left claimed by this worker."""
        for future in leftovers:
            self._requeue(self._active[future].id)
        metrics.record_requeues(self.name, len(leftovers))
        logger.warning(
            "everystep worker: drain deadline of %ss expired with %d workflow(s) still "
            "in flight; they were requeued and will be picked up by the next available "
            "runner",
            self.drain, len(leftovers),
        )

    def _requeue(self, workflow_id):
        """Put a run back in the queue: scheduled and unclaimed, so any
        runner can claim it and resume it from the recorded steps."""
        updated = Workflow.objects.filter(
            id=workflow_id,
            status=Workflow.Status.RUNNING,
            claimed_by=self.name,
        ).update(status=Workflow.Status.SCHEDULED, claimed_by=None)
        if not updated:
            logger.warning(
                "everystep worker: could not requeue workflow %s: no longer claimed by %r",
                workflow_id, self.name,
            )

    def _catchup(self):
        """Re-claim the runs a crashed process with this name left behind.

        Runs once at startup, before the poll loop, so it cannot re-select
        workflows this process is already executing in its pool. After a
        clean shutdown it finds nothing: in-flight runs are requeued at
        shutdown.
        """
        workflows = resume_own(self.pool_size, self.name)
        for workflow in workflows:
            self._active[self._executor.submit(self._execute, workflow)] = workflow
        metrics.record_claims(self.name, len(workflows))

    def _reap(self):
        pending = {}
        for future, workflow in self._active.items():
            if future.done():
                exc = future.exception()
                if exc is not None:
                    logger.exception("everystep worker: unexpected worker failure: %s", exc)
            else:
                pending[future] = workflow
        self._active = pending

    def _execute(self, workflow):
        started = time.monotonic()
        try:
            execute(workflow.id, draining=self._draining)
        except Workflow.DoesNotExist:
            logger.warning("everystep worker: workflow %s no longer exists", workflow.id)
        except DrainOrphan:
            # Drained at a step boundary with everything so far recorded:
            # give the run back to the queue for any runner to resume.
            self._requeue(workflow.id)
        except Exception as exc:
            logger.exception("everystep worker: workflow %s crashed outside the runner", workflow.id)
            report_workflow_failure(exc, workflow_id=workflow.id)
            try:
                # Guarded on claimed_by: if this run was requeued while a
                # step was still in flight (drain deadline) and since
                # claimed by another runner, its late failure must not fail
                # that runner's live run.
                updated = Workflow.objects.filter(
                    id=workflow.id,
                    status=Workflow.Status.RUNNING,
                    claimed_by=self.name,
                ).update(
                    status=Workflow.Status.FAILED,
                    error=serde.encode_exception(exc),
                    completed_at=timezone.now(),
                )
                if updated and not isinstance(exc, (SimulatedCrash, Terminal)):
                    metrics.record_workflow_terminal(
                        workflow.name, "failed", time.monotonic() - started
                    )
            except Exception:
                logger.exception("everystep worker: could not mark workflow %s failed", workflow.id)
        finally:
            # Pool threads are long-lived and Django connections are
            # thread-local, so release this thread's connection to avoid
            # leaking one per executed workflow.
            connections.close_all()