A publisher on one replica

Some work must run on exactly one replica at a time: a publisher that numbers what it sends, a snapshot builder, a coordinator. Run it on every replica and the output duplicates; run it on one fixed replica and it stops when that replica does. This recipe puts a publisher behind a leadership lease and fences its writes, so a leader whose lease has passed on cannot overwrite its successor. The model — terms, deadlines, why fencing — is in Run one replica's worth of work.

The runnable version lives at examples/recipes/leadership/ and runs on the in-memory mock, two replicas in one process.

The lease¶

The lock spec requires fencing tokens and a counter that can be moved up after a loss (requires_fence_advance), so a lock backend without them fails at startup instead of leaving the writes unfenced:

PUBLISHER = LeadershipSpec(
    lock=DistributedLockSpec(
        name="digest-publisher",
        ttl=timedelta(seconds=6),
        # Fencing tokens, and a counter that can be moved up after a loss; a backend
        # without both fails at startup.
        requires_fence_advance=True,
    ),
    key="digest-publisher",
    renew_every=timedelta(seconds=2),
    campaign_every=timedelta(milliseconds=500),
    margin=timedelta(milliseconds=500),
    yield_grace=timedelta(seconds=1),
    release_timeout=timedelta(milliseconds=500),
)
PUBLISH_EVERY = 0.5

renew_every + yield_grace + release_timeout + margin must be shorter than the TTL, and the spec refuses anything else when it is built: the body's grace, and its unwind if it is cancelled, have to end before the lease could lapse.

The fenced store¶

The store keeps the highest token that wrote. A term's first write claims its token — the stored fence must be strictly lower — and its later writes accept an equal one. The stand-in below is what the two SQL statements in its docstring do:

@attrs.define(slots=True)
class FencedDigests:
    """Stands in for the table the leader writes; each method is one conditional write.

    claim: UPDATE digests SET fence = :token WHERE name = :name AND fence < :token
    write: UPDATE digests SET body = :body, fence = :token WHERE name = :name AND fence <= :token
    """

    fence: int = 0
    published: list[tuple[int, str]] = attrs.field(factory=list)

    def claim(self, token: int) -> bool:
        if self.fence >= token:
            return False  # another holder of this token, or a newer one, got here first

        self.fence = token
        return True

    def write(self, token: int, body: str) -> bool:
        if self.fence > token:
            return False  # a newer holder has written: this one's lease has passed on

        self.fence = token
        self.published.append((token, body))
        return True

The claim is what settles a repeated token. After the lock backend loses its data, its counter can issue a token a stale holder already drew; both try the claim, the store serializes them, and exactly one finds the fence lower.

The step¶

The claim runs as the claim hook, before the body, and fence_floor reads the store's fence for the step to advance the lock's counter past while the lease is free. The body publishes one unit at a time, checks term.live() between units, and stops at the first write the fence refuses:

def publisher_step(store: FencedDigests, *, owner: str) -> LifecycleStep:
    async def claim(_ctx: ExecutionContext, term: Term) -> bool:
        # Runs before the body, so a stale holder of this token is stopped by the fence
        # before this term writes anything.
        return term.token is not None and store.claim(term.token)

    async def floor(_ctx: ExecutionContext) -> int:
        # Read while the lease is free: the counter is moved past it before acquiring.
        return store.fence

    async def publish(_ctx: ExecutionContext, term: Term) -> None:
        sequence = 0

        while term.live():  # one unit at a time; stop at a unit boundary
            sequence += 1

            if term.token is None or not store.write(term.token, f"{owner} #{sequence}"):
                return

            with suppress(TimeoutError):
                await asyncio.wait_for(term.ended.wait(), timeout=PUBLISH_EVERY)

    return leadership_lifecycle_step(
        PUBLISHER,
        owner=owner,
        lead=publish,
        claim=claim,
        fence_floor=floor,
    )

Register the step on every replica, with each replica's own owner:

from forze.application.execution import LifecyclePlan

lifecycle = LifecyclePlan.from_steps(publisher_step(store, owner=instance_id))

Run it¶

uv run python -m examples.recipes.leadership.app

One replica publishes with token 1. The script stops it the way a deploy does: its term ends as stopped, the body finishes its unit, the lease is released, and the other replica takes over within half a second with token 2.

After the lock backend loses its data¶

The script then starts a replica over an empty lock backend and a store that has already seen token 50 — what a Redis restart without persistence leaves behind:

async def recover_after_loss(
    *,
    store: FencedDigests | None = None,
    run_for: float = 1.0,
) -> FencedDigests:
    """A lock backend that lost its data, under a store that wrote up to token 50."""

    store = store if store is not None else FencedDigests(fence=50)
    ctx = ExecutionContext(
        deps=DepsRegistry.from_modules(MockDepsModule(state=MockState())).freeze().resolve()
    )
    await publisher_step(store, owner="pod-a").startup(ctx)
    await asyncio.sleep(run_for)  # the first term draws 51 and claims it at once
    await ctx.drainables.stop_all(grace=5)

    return store

The lease is free, so the step reads the floor, moves the counter to 50 and acquires: the first term draws 51 and its claim goes through at once. Without fence_floor the counter would start again at 1, every claim up to 50 would be refused, and the unit would go unled for fifty campaign intervals.

Notes¶

Keep the body's units short. It is told its term is ending yield_grace before the deadline and cancelled after that, and must then finish within release_timeout. A unit that cannot finish in the grace is cut off mid-way.
Fence every write the body makes as leader, not only the first. A leader paused past its lease stops at its next term.live() check, but finishes the unit it was paused in, and only the store can refuse that unit's writes.
Wire the store's check yourself. The framework issues the token; refusing a lower one is the store's job, and without it the lease is best-effort exclusion.