symfonic.kernel.background

background

R7 — the run's background-work registry (RCX-8, BP-8).

The construct this module retires is the module-level _background_tasks set: a process-wide container that outlives every run that adds to it, so a task detached by run A is still reachable — and still running — long after A returned. Ownership there is nobody's, which is why nothing ever waits for it and nothing ever reports it.

Here, a task cannot exist without an owner, a purpose, and a deadline that is no wider than the run's; it cannot be created after the run has closed; and teardown either awaits it inside a bounded grace window or cancels it and says so. There is one registry per :class:~symfonic.kernel.context.RequestContext and no registry anywhere else.

BackgroundRegistry

BackgroundRegistry(run_id: str, *, run_deadline_seconds: float | None = None)

Every task one invocation spawned, owned from creation to teardown.

Source code in src/symfonic/kernel/background.py

def __init__(self, run_id: str, *, run_deadline_seconds: float | None = None) -> None:
    self.run_id = run_id
    self._run_deadline = run_deadline_seconds
    self._tasks: list[asyncio.Task[Any]] = []
    self._entries: list[BackgroundEntry] = []
    self._closed = False

entries property

entries: tuple[BackgroundEntry, ...]

The owner/purpose/deadline of every task still held.

drain async

drain(*, grace_seconds: float) -> DrainReport

Await what finishes inside the grace window, cancel and await the rest.

Closing first is what makes the window a bound: a task that spawned another task on its way out would otherwise refill the registry behind the drain, and the loop would be as long as the work chose to make it.

Source code in src/symfonic/kernel/background.py

async def drain(self, *, grace_seconds: float) -> DrainReport:
    """Await what finishes inside the grace window, cancel and await the rest.

    Closing *first* is what makes the window a bound: a task that spawned
    another task on its way out would otherwise refill the registry behind
    the drain, and the loop would be as long as the work chose to make it.
    """
    self._closed = True
    owned = tuple(zip(self._tasks, self._entries, strict=True))
    self._tasks.clear()
    self._entries.clear()
    if not owned:
        return DrainReport()

    remaining = [task for task, _ in owned if not task.done()]
    if remaining and grace_seconds > 0:
        _, unfinished = await asyncio.wait(remaining, timeout=grace_seconds)
        remaining = list(unfinished)

    for task in remaining:
        task.cancel()
    if remaining:
        await asyncio.gather(*remaining, return_exceptions=True)
    return DrainReport(
        awaited=len(owned) - len(remaining),
        cancelled=len(remaining),
        failures=_failures(owned),
    )

spawn

spawn(work: Coroutine[Any, Any, Any], *, owner: str, purpose: str, deadline_seconds: float | None = None) -> asyncio.Task[Any]

Register and start one unit of run-owned work.

Every rejection closes work first. A coroutine that is refused and then left unawaited would surface as a RuntimeWarning from whatever code happened to run next, attributing this module's refusal to an innocent bystander.

Source code in src/symfonic/kernel/background.py

def spawn(
    self,
    work: Coroutine[Any, Any, Any],
    *,
    owner: str,
    purpose: str,
    deadline_seconds: float | None = None,
) -> asyncio.Task[Any]:
    """Register and start one unit of run-owned work.

    Every rejection closes ``work`` first. A coroutine that is refused and
    then left unawaited would surface as a ``RuntimeWarning`` from whatever
    code happened to run next, attributing this module's refusal to an
    innocent bystander.
    """
    deadline = deadline_seconds if deadline_seconds is not None else self._run_deadline
    problem = self._reject(owner, purpose, deadline_seconds)
    if problem is not None:
        _discard(work)
        raise ContractViolationError(problem)

    entry = BackgroundEntry(owner=owner, purpose=purpose, deadline_seconds=deadline)
    task = asyncio.create_task(
        _bounded(work, deadline, owner, purpose),
        name=f"{owner}:{purpose}:{self.run_id}",
    )
    self._tasks.append(task)
    self._entries.append(entry)
    return task

DrainReport dataclass

DrainReport(awaited: int = 0, cancelled: int = 0, failures: tuple[BackgroundFailure, ...] = ())

What draining the registry cost: work that finished, work cut, work that failed.

failures is not a subset of cancelled: a task can finish well inside the grace window and still have raised. Reporting the two separately is what lets teardown say "nothing was forced, but something owned by this run broke" — a sentence the old awaited/cancelled pair could not form.