Recovering batch and crawl jobs after a worker restart

Asynchronous scraping changes the failure model. POST /batch, /crawl, and /search return a job_id before the work finishes. If the process that submitted the request exits, the server-side job can keep running, but a client that discarded the identifier has no reliable way to collect the result.

Persist the job identifier before starting the polling loop. Treat the identifier as a durable handle, not as a temporary variable in a web request.

Store the enqueue record

import os
import time
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}


def enqueue_batch(urls: list[str], save_job) -> str:
    response = requests.post(
        f"{API}/batch",
        headers=HEADERS,
        json={"urls": urls, "formats": ["markdown"]},
        timeout=30,
    )
    response.raise_for_status()       # 202 Accepted
    job_id = response.json()["job_id"]

    # Save before returning control to the rest of the pipeline.
    save_job({
        "job_id": job_id,
        "kind": "batch",
        "state": "submitted",
        "submitted_urls": len(urls),
    })
    return job_id

The persisted record should include the operation type, the input reference, enqueue time, and a local processing state. Store the URLs in object storage or a database table when the list is large; do not make a 50-item request payload the only copy of your input.

Resume with GET /jobs/{job_id}

The same status endpoint serves jobs created by batch, crawl, and search. Polling is recommended every 2 seconds. The response contains a status and, when the job succeeds, the results. A restarted worker can read all records in submitted or polling state and resume without submitting duplicates.

TERMINAL = {"SUCCESS", "FAILURE", "REVOKED"}


def collect_job(job_id: str, timeout_s: int = 600) -> dict:
    deadline = time.monotonic() + timeout_s
    last_status = "UNKNOWN"

    while time.monotonic() < deadline:
        response = requests.get(
            f"{API}/jobs/{job_id}",
            headers=HEADERS,
            timeout=30,
        )
        response.raise_for_status()
        job = response.json()
        last_status = job["status"]

        if last_status in TERMINAL:
            return job
        time.sleep(2)

    raise TimeoutError(f"{job_id} remained {last_status} for {timeout_s}s")

A 404 from the status endpoint is not the same as PENDING. The API checks job ownership by tenant and returns 404 when the identifier does not exist or belongs to another account. Mark the local record as an operator-visible lookup failure and do not keep polling an identifier that the API cannot resolve.

Commit results idempotently

A worker can crash after receiving SUCCESS but before marking the local record complete. The next worker may see the same terminal job again. Write results with a uniqueness key such as (job_id, item_index) or (job_id, url), then mark the job complete in the same database transaction where possible.

def persist_results(job: dict, insert_if_missing, mark_complete) -> None:
    if job["status"] != "SUCCESS":
        raise RuntimeError(job.get("error") or job["status"])

    for item in job.get("results", []):
        insert_if_missing(
            job_id=job["job_id"],
            url=item["url"],
            scrape_status=item["scrape_status"],
            markdown=item.get("markdown"),
        )

    mark_complete(job["job_id"], credits_used=job.get("credits_used", 0))

insert_if_missing must be safe to call twice. Do not use the number of rows inserted as proof that the remote job ran only once. The remote job is identified by job_id; local delivery is a separate state machine.

Interpret partial success correctly

A batch or crawl can reach job-level SUCCESS while individual result entries carry blocked_antibot, timeout, or extraction_failed. Reconcile per-item status and the job's credits_used. Do not retry the entire job because one page failed; select only statuses your policy considers retriable, and submit a new job with a new local attempt record.

For search jobs, inspect stopped_reason as well. A token_cap result means the search stopped before processing every source, not that the status endpoint failed. Preserve that reason next to the results so downstream consumers know whether the response was complete.

Make restart recovery boring

The recovery loop is simple when three records are durable: the original input, the remote job_id, and the local delivery state. On startup, query unfinished records, poll each identifier with a bounded deadline, and commit terminal results idempotently. Never infer completion from a process exit, and never enqueue a replacement merely because the submitting process disappeared.

Asynchronous APIs are easier to operate when the client treats job submission and result delivery as two separate steps. Persisting the handle closes the gap between them and turns a worker restart into a normal retry of local observation, not a duplicate scrape.