symfonic.capabilities.memory.napping

napping

When a turn triggers consolidation, and what stops two from colliding.

The shipped engine counts turns on itself and fires quick_nap from three call sites with asyncio.create_task. That gives the cadence three copies, the task no owner, and the tenant no isolation: the counter is the agent's, so two tenants served by one agent share a cadence and either consolidates on the other's turns.

Here the cadence is a :class:~symfonic.capabilities.memory.schedule.ConsolidationSchedule, the counter is a :class:~symfonic.capabilities.memory.schedule.ScheduleCursor per scope, and the coordinator is the only thing that advances either.

Three rules, and each of them is a way two cycles could otherwise consolidate one batch twice:

• The decision is atomic. :meth:ConsolidationCoordinator.after_turn advances the cursor and reads the schedule without awaiting, so two turns finishing together cannot both see the same turn count as due. The first takes the cycle and resets the cursor; the second is one turn into the next cadence.
• Execution is serialised per scope, across processes. A cycle holds that scope's lease for its whole run, so a turn-driven quick nap, a scheduler-driven Deep Sleep and an operator's manual backfill never interleave over the same rows -- and those are three different processes, which an asyncio.Lock would not have coordinated at all. Renewed by a heartbeat while the cycle runs (:mod:.heartbeat), so a Deep Sleep whose model call outlasts the TTL keeps its scope and expiry keeps meaning "this worker died"; and fenced at every durable write (:mod:.fencing) for when it lapses anyway. See :mod:.leases: the in-process table is for a deployment that declares itself to be one, and the Postgres adapter is what the scaffold composes. The loser is told the scope is already running; it never reports having consolidated.
• Only a cycle that finished counts as done. A cycle that raised still advances the cursor -- it ran, it published nothing, and retrying it on the very next turn would turn one broken store into a nap every turn. A cycle that was cancelled does not: it did not finish, and the run that cancelled it is the run that will not be reporting it.

The coordinator owns the task, and outlives the turn. A nap is maintenance for a scope, not work belonging to one invocation, so the task is held by :class:~.owned_cycles.OwnedCycles -- referenced, drained by :meth:ConsolidationCoordinator.aclose, and its failure logged -- rather than on the run's registry. That module carries the reasoning.

The cursor lives in memory unless a deployment hands one in. A restart therefore restarts the turn count, which is what the shipped engine does too; ScheduleCursor.to_state exists for a deployment that would rather not.

ConsolidationCoordinator

ConsolidationCoordinator(runtime: ConsolidationRuntime, *, schedule: ConsolidationSchedule | None = None, cursors: MutableMapping[str, ScheduleCursor] | None = None, background: bool = True, cycles: Sequence[ConsolidationCycle] = (ConsolidationCycle.QUICK,), leases: Any = None, lease_ttl_seconds: float = 900.0, transaction: Any = None, terminal_sink: Any = None)

Bases: BackgroundCycles

Decides when a scope consolidates, and runs at most one cycle at a time.

Source code in src/symfonic/capabilities/memory/napping.py

def __init__(
    self,
    runtime: ConsolidationRuntime,
    *,
    schedule: ConsolidationSchedule | None = None,
    cursors: MutableMapping[str, ScheduleCursor] | None = None,
    background: bool = True,
    cycles: Sequence[ConsolidationCycle] = (ConsolidationCycle.QUICK,),
    leases: Any = None,
    #: How long a lease stands before it lapses. Long by default because
    #: a port with no ``renew`` has only the TTL between a slow cycle and
    #: losing its scope; on a renewing port -- both shipped adapters are --
    #: it can safely be far shorter. See :mod:`.heartbeat`.
    lease_ttl_seconds: float = 900.0,
    #: The one transaction domain this cycle commits in: the pool the
    #: graph, the lease and the staged records all speak through. Required
    #: whenever the lease port has one, and checked here at composition
    #: rather than discovered from a report that said ``clean`` about half
    #: a cycle. See :func:`~.commit.require_domain_for`.
    transaction: Any = None,
    terminal_sink: Any = None,
) -> None:
    self._runtime = runtime
    self._terminal_sink = terminal_sink
    self._cadence = Cadence(schedule or ConsolidationSchedule(), cycles, cursors)
    # Required, and not defaulted to the in-process table: a default that
    # silently coordinated one process would be the exact failure this port
    # exists to prevent, arrived at by omission.
    if leases is None:
        raise MemoryContractError(
            "a consolidation coordinator needs a lease port. Two cycles "
            "over one scope corrupt each other's counters and double its "
            "model spend, and the processes that would collide -- a "
            "scheduler and a manual backfill -- do not share memory. Pass "
            "``PostgresLeases(pool)``, or "
            "``InProcessLeases(single_process=True)`` if this deployment "
            "really is one process."
        )
    self._leases = leases
    self._domain = require_domain_for(leases, runtime, transaction)
    self._lease_ttl = lease_ttl_seconds
    self._tasks = OwnedCycles()
    #: Whether the turn waits for the cycle. A deployment that wants the
    #: answer held until its memories are consolidated sets this ``False``.
    self.background = background
    for cycle in cycles:
        if not runtime.serves(cycle):
            raise MemoryContractError(
                f"this coordinator would schedule the {cycle.value} cadence and "
                f"its runtime cannot serve it (registered: "
                f"{', '.join(runtime.registered) or 'nothing'}). Build the "
                "runtime from ``quick_phases`` so the roster is complete "
                "before anything is scheduled against it."
            )

leases property

leases: Any

The lease port. Published so a composition root can prepare it.

An adapter backed by a table has schema to create, and the app that opened the pool is the thing that knows when to do it.

registered property

registered: tuple[str, ...]

The phases a cycle from this coordinator will run, by name.

Published because "which phases does my nap run?" is a question an operator asks of the thing they configured, and answering it by reaching for the runtime inside would be the private access this capability exists to remove.

after_turn

after_turn(scope: MemoryScope) -> ConsolidationCycle | None

Record a successful turn and answer which cycle is now due.

Source code in src/symfonic/capabilities/memory/napping.py

def after_turn(self, scope: MemoryScope) -> ConsolidationCycle | None:
    """Record a successful turn and answer which cycle is now due."""
    return self._cadence.after_turn(scope)

cursor

cursor(scope: MemoryScope) -> ScheduleCursor

This scope's counter. A scope nobody has served yet has a fresh one.

Source code in src/symfonic/capabilities/memory/napping.py

def cursor(self, scope: MemoryScope) -> ScheduleCursor:
    """This scope's counter. A scope nobody has served yet has a fresh one."""
    return self._cadence.cursor(scope)

run async

run(scope: MemoryScope, cycle: ConsolidationCycle, *, run_id: str = '', root_run_id: str = '') -> ConsolidationState

Run one cycle over one scope, alone.

run_id/root_run_id are the turn this cycle is attributable to. A quick nap fires from a turn and phase 13 may spend a model call, so the cost belongs to that tenant and that root run rather than to whichever turn happened to be in flight when the task was scheduled.

Bound as the ambient run identity for the cycle, not merely passed on the phase context: a provider reads the ContextVar, and a phase calls the model through the provider rather than through anything that could be handed the ids. Bound explicitly rather than inherited from the task that created this one, so a cycle a scheduler ran outside any turn carries no identity instead of quietly borrowing whichever run happened to be ambient when the scheduler ticked.

Returns a state whose status is already_running when another holder has the scope. A None would have been the smaller change and the wrong one: the caller needs a record saying why nothing happened, and an absent one reads like a cycle that ran and found nothing.

Source code in src/symfonic/capabilities/memory/napping.py

async def run(
    self,
    scope: MemoryScope,
    cycle: ConsolidationCycle,
    *,
    run_id: str = "",
    root_run_id: str = "",
) -> ConsolidationState:
    """Run one cycle over one scope, alone.

    ``run_id``/``root_run_id`` are the turn this cycle is attributable to.
    A quick nap fires from a turn and phase 13 may spend a model call, so
    the cost belongs to that tenant and that root run rather than to
    whichever turn happened to be in flight when the task was scheduled.

    Bound as the ambient run identity for the cycle, not merely passed on
    the phase context: a provider reads the ContextVar, and a phase calls
    the model through the provider rather than through anything that could
    be handed the ids. Bound explicitly rather than inherited from the task
    that created this one, so a cycle a scheduler ran outside any turn
    carries no identity instead of quietly borrowing whichever run happened
    to be ambient when the scheduler ticked.

    Returns a state whose ``status`` is ``already_running`` when another
    holder has the scope. A ``None`` would have been the smaller change and
    the wrong one: the caller needs a record saying *why* nothing happened,
    and an absent one reads like a cycle that ran and found nothing.
    """
    owner = new_owner_token()
    lease = await self._leases.acquire(scope, owner=owner, ttl_seconds=self._lease_ttl)
    if lease is None:
        logger.info(
            "consolidation cycle=%s scope=%s skipped: already running",
            cycle.value,
            scope.path,
        )
        return already_running(scope, cycle)
    try:
        # Renewed while the cycle runs, so the TTL can stay short enough
        # for a crash to be noticed without a slow-but-living worker
        # losing its scope to its own model call.
        async with heartbeat(self._leases, lease):
            with attributed(run_id, root_run_id):
                return await self._runtime.run(
                    scope,
                    cycle,
                    run_id=run_id,
                    root_run_id=root_run_id,
                    # The fence. Checked before every durable write, so a
                    # cycle whose lease lapsed anyway stops rather than racing
                    # the worker that took the scope over.
                    still_authorised=lambda: self._leases.holds(lease),
                    # The commit-time check, which locks the lease row and
                    # keeps it until the transaction ends, so a rival
                    # acquisition queues behind this commit rather than
                    # landing between the answer and the batch it allowed.
                    commit_authority=lambda: self._leases.hold_for_update(lease),
                    domain=self._domain,
                )
    finally:
        # On success, on failure and on cancellation. Owner-checked inside
        # the port, so a slow worker whose lease was taken over does not
        # free it for whoever holds it now.
        await self._leases.release(lease)