Run an Action-Runner Loop

A subscription normally pushes: a write lands, WarmHub POSTs to your webhook, your server handles it. An action runner pulls instead. It holds an exclusive lease on the subscription, and while it does, matching writes wait for the runner rather than going to the webhook. The runner picks each one up, does the work, and marks it done. No inbound HTTP endpoint is involved.

The loop has four verbs, all on client.action:

acquire — acquire_lease: take the subscription.
claim — claim_delivery: take one waiting delivery.
complete — complete_delivery: mark it done.
release — release_lease: give the subscription back.

heartbeat_lease keeps the lease alive while the loop runs, and live_feed is how the runner sees what is waiting.

Before you start: a subscription, and a shape for your output

The runner leases an existing webhook subscription. A lease does not replace the webhook, it takes precedence over it: a delivery the runner does not claim in time, or that arrives while no lease is held, goes to the webhook as normal. That fallback POST carries the matched operations, job data included, so webhookUrl must be an endpoint you control — your own service, or a receiver that alerts you that the runner fell behind. Never leave a placeholder there. WarmHub validates the URL when you create the subscription; it must be a public HTTPS address.

from warmhub import Add, WarmHubClient, WarmHubError

client = WarmHubClient.from_env()
org, repo_name = "acme", "imaging"
jobs = client.repository(f"{org}/{repo_name}")

client.subscription.create(
    {
        "orgName": org,
        "repoName": repo_name,
        "name": "jobs",
        "kind": "webhook",
        "eventType": "commit",
        "webhookUrl": "https://jobs.acme.example/warmhub-fallback",  # an endpoint you own
        "filterJson": {"shape": "Job", "operation": "add"},
    }
)

Write the runner’s output under a different shape from the one the subscription matches. Here jobs arrive as Job and results are written as JobResult. If the runner wrote its results as Job, every result would queue another delivery for the same runner.

See Creating subscriptions for filters and webhook requirements.

Acquire the lease

HOLDER = "imaging-runner-1"
TTL_MS = 90_000

lease = client.action.acquire_lease(
    org, repo_name, "jobs", HOLDER, "sdk", grace_ms=60_000, ttl_ms=TTL_MS
)
if not lease.ok:
    raise SystemExit(f"cannot run: {lease.reason}")

holder_id is any string that identifies this runner process; every later call must pass the same one. holder_type is "sdk". Two durations matter:

ttl_ms — how long the lease lives without a heartbeat. The default is 30 seconds.
grace_ms — how long a new delivery waits for this runner to claim it before WarmHub sends it to the webhook instead. The default is 2 seconds. A runner that polls every few seconds needs more than that, or most deliveries will leave before it looks.

The result is a tagged union on ok. ok=True carries reused, which is True when this holder already held the lease. ok=False carries reason LEASE_ALREADY_HELD: another holder has it. A refused acquire is a return value, not an exception.

Claim, work, complete

The runner finds work in live_feed_iter: one row per delivery for the subscription, newest first. Rows waiting for this runner have status lease_wait. Each row carries the operations that matched the filter in matched_operations.

A delivery is per commit, not per operation. Three jobs written in one commit arrive as one delivery with three matched operations.

Only additions create work. Give every task a fresh name; do not revise, retract, rename or reuse job names. Result names are derived from those immutable job identities, so a partial multi-operation retry can reuse already-written results.

import threading
import time


def process(item, check):
    """Each uniquely named job is immutable; a retry can reuse its result."""
    for match in item.matched_operations:
        job = match.operation["name"]
        task = match.operation["data"]["task"]
        check()
        result = jobs.apply(
            f"result for {job}",
            [Add(name=f"result-{job}", shapes=["JobResult"],
                 data={"job": job, "output": f"done: {task}"}, skip_existing=True)],
        )
        if any(op.status == "error" for op in result.operations):
            raise RuntimeError(f"result write failed for {job}")


class Runner:
    def __init__(self, client, renewal_client, org, repo_name, subscription, holder, ttl_ms, process):
        self.client, self.renewal_client = client, renewal_client
        self.org, self.repo_name = org, repo_name
        self.subscription, self.holder = subscription, holder
        self.ttl_ms, self.process = ttl_ms, process
        self.pending = {}
        self.feed = None
        self.unowned = set()
        self.stop = threading.Event()
        self.failed = threading.Event()
        self.failure = None
        self.last_renewed = 0.0
        self.thread = threading.Thread(target=self.renew_loop)

    def renew(self):
        started = time.monotonic()
        try:
            result = self.renewal_client.action.heartbeat_lease(
                self.org, self.repo_name, self.subscription, self.holder, self.ttl_ms
            )
            if not result.ok:
                raise RuntimeError(f"heartbeat refused: {result.reason}")
            # Count from dispatch, conservatively allowing for response latency.
            self.last_renewed = started
        except Exception as exc:
            self.failure = exc
            self.failed.set()

    def renew_loop(self):
        while not self.stop.wait(self.ttl_ms / 3000):
            self.renew()
            if self.failed.is_set():
                return

    def check(self):
        if self.failed.is_set() or time.monotonic() - self.last_renewed >= self.ttl_ms / 1000:
            raise RuntimeError("lease ownership uncertain; stop work and arrange recovery") from self.failure

    def __enter__(self):
        self.renew()
        self.check()
        self.thread.start()
        return self

    def __exit__(self, *exc):
        self.stop.set()
        self.thread.join()
        # Never release here: failed work may still be claimed.

    def finish_pending(self):
        for delivery_id, item in list(self.pending.items()):
            self.check()
            try:
                self.process(item, self.check)
            except Exception:
                self.check()
                print(f"delivery {delivery_id} failed; retrying next pass")
                continue
            self.check()
            result = self.client.action.complete_delivery(
                self.org, self.repo_name, self.holder, delivery_id=delivery_id
            )
            if not result.ok:
                self.failed.set()
                raise RuntimeError(f"completion refused: {result.reason}; recovery required")
            del self.pending[delivery_id]

    def run_pending(self):
        self.check()
        if self.pending:
            self.finish_pending()
            return
        if self.feed is None:
            self.feed = self.client.action.live_feed_iter(self.org, self.repo_name, self.subscription)
        for item in self.feed:
            self.check()  # Also checks after a slow page fetch.
            if item.status == "leased":
                if item.delivery_id not in self.unowned:
                    self.unowned.add(item.delivery_id)
                    print(f"unowned delivery {item.delivery_id}: recovery required")
                continue
            if item.status != "lease_wait":
                continue
            claim = self.client.action.claim_delivery(
                self.org, self.repo_name, self.holder, delivery_id=item.delivery_id
            )
            if claim.ok:
                self.pending[item.delivery_id] = item
                break
        else:
            self.feed = None
        self.check()
        self.finish_pending()

    def stop_cleanly(self, timeout_s=60.0):
        deadline = time.monotonic() + timeout_s
        while self.pending:
            self.check()
            if time.monotonic() >= deadline:
                raise TimeoutError("owned claims remain; continue draining with heartbeats running")
            self.finish_pending()  # Drain owned claims without feed scans.
            if self.pending:
                time.sleep(0.5)
        self.check()
        released = self.client.action.release_lease(
            self.org, self.repo_name, self.subscription, self.holder
        )
        if not released.ok:
            raise RuntimeError(f"release refused: {released.reason}")


# A separate SDK client lets renewal proceed during blocking work and pagination.
with WarmHubClient.from_env() as renewal_client:
    with Runner(client, renewal_client, org, repo_name, "jobs", HOLDER, TTL_MS, process) as runner:
        while not stopping():
            runner.run_pending()
            time.sleep(0.5)
        while True:
            try:
                runner.stop_cleanly()
                break
            except TimeoutError:
                print("drain timed out: alert the operator; continuing with renewal active")

stopping() stands for your shutdown signal. The renewal thread runs during fetching, work and draining, using its own client. This example uses a 90-second TTL to leave room for the SDK’s bounded request timeouts; it renews every 30 seconds. A refused or failed heartbeat stops further work at the next check. A synchronous job already running cannot be cancelled by the thread: check lease health before each side effect, and use idempotency for effects already in flight.

The runner holds at most one in-flight delivery: it retries that delivery before claiming another, and stops fetching as soon as a claim succeeds. One delivery can still contain multiple operations. The iterator is retained between passes until exhaustion, so completing a delivery does not restart pagination over its completed prefix. The pending dictionary contains only deliveries this process successfully claimed. Feed rows marked leased do not expose ownership, so pre-existing rows are reported and skipped before processing; they are not proven orphans.

claim_delivery and complete_delivery take exactly one of delivery_id= or run_id=. Use delivery_id= for a lease_wait delivery, which has no run yet; run_id= addresses a delivery that already has a managed run. Passing both or neither raises TypeError before any request is sent.

Heartbeat at least once per ttl_ms. A lease that expires is gone: the next holder to call acquire_lease gets it.

The public feed has only a page limit and an offset cursor; it has no waiting-status or since filter. After exhausting a traversal, this example starts another to find new or late-visible deliveries. Idle polling therefore still scans retained history, and arrivals during a traversal may wait until the next one. The claim grace deadline continues running during that wait, so a large backlog can send unclaimed jobs to the webhook fallback. The offset cursor is not a stable checkpoint for safely skipping previous rows. This is a small-volume example. Production throughput needs bounded discovery of waiting deliveries; increasing the lease TTL does not solve that API limitation. list_runs cannot substitute: waiting deliveries have no managed run yet.

When work or shutdown fails

A failed job stays in pending and is retried by this process under the same live lease. Results use skip_existing=True, and per-operation write failures prevent completion. Draining retries only these owned claims; it stops claiming new work and does not repeatedly scan completed history.

A drain timeout is an alert, not successful shutdown. The outer loop catches it inside the heartbeat context and continues draining. An operator must resolve permanent failures or arrange recovery before stopping the process. The timeout bounds retries between jobs; it cannot interrupt a blocking job.

Release the lease without outstanding owned claims

Release only after pending is empty. Leaving the context stops renewal; it does not release the lease. On fatal renewal failure or process termination, outstanding claims require operator recovery. Do not describe an exiting process as keeping a lease alive.

Claiming moves a delivery from lease_wait to leased. Webhook fallback applies only to lease_wait. Expiry or release unlinks claimed deliveries from the lease; a later lease cannot complete them (DELIVERY_LEASE_MISMATCH).

This example keeps ownership in memory and does not recover automatically after a crash or restart, even with the same holder ID. The feed does not identify a claim’s lease, so a restarted process cannot safely infer ownership of leased rows. Use an explicitly designed durable ownership and recovery contract before using this pattern for jobs that must survive process failure. The checks reduce ordinary expiry risk; they cannot make a network lease and a job side effect atomic.

Handling refusals

The verbs refuse in two ways, and a runner has to handle both:

Situation	What you get
Another holder has the lease when you acquire	`ok=False`, `reason="LEASE_ALREADY_HELD"`
Acquire again with the same holder	`ok=True`, `reused=True`
Complete a delivery that is already complete	`ok=False`, `reason="DELIVERY_LEASE_MISMATCH"`
Claim with a `holder_id` that does not hold the lease	`WarmHubError`, code `VALIDATION_ERROR`
Release a lease that is no longer held	`WarmHubError`, code `NOT_FOUND`
Complete a delivery claimed under a lease that has since ended	`ok=False`, `reason="DELIVERY_LEASE_MISMATCH"`

An ok=False result means the request was well formed but the delivery or lease is not in a state this holder can act on. A WarmHubError means the request named a lease this holder does not have. Branch on ok and on err.code, never on message text.

Delivery lifecycle lists every delivery status, including lease_wait, leased, and completed.
client.action reference
Keep a local cache with change drains is the polling alternative when you need state rather than events.