Cron-Triggered Batch Runner with Backpressure

A cron entry that fires a caption QC batch every five minutes eventually meets a backlog it cannot clear in five minutes — the next fire overlaps the last, two runs process the same queue, memory balloons as both eagerly load thousands of assets, and the box is OOM-killed mid-scan. The runner below prevents all three failure modes at once: a per-run fcntl lock makes an overlapping cron fire exit immediately, an asyncio.Semaphore bounds concurrency so only N assets are ever resident, a producer/consumer asyncio.Queue with a fixed maxsize applies backpressure so the scan cannot outrun processing, and an idempotent skip means an asset already recorded in the scheduled QC telemetry is never re-processed. A SIGTERM handler drains in-flight work cleanly instead of leaving half-written reports.

The bounded, idempotent async runner

# batch_runner.py — cron-safe asyncio caption QC runner with backpressure + a run lock
from __future__ import annotations

import asyncio
import fcntl
import os
import signal
import sys
from collections.abc import AsyncIterator
from contextlib import contextmanager
from pathlib import Path

MAX_WORKERS = min(8, (os.cpu_count() or 2) * 2)   # bound resident assets; I/O-bound → 2x cores, capped
QUEUE_BOUND = MAX_WORKERS * 4                      # backpressure: scan blocks when the buffer fills
LOCK_PATH = Path("/var/run/caption_qc/batch.lock")


@contextmanager
def run_lock(path: Path):
    """Exclusive, non-blocking fcntl lock — a second cron fire fails fast instead of overlapping."""
    path.parent.mkdir(parents=True, exist_ok=True)
    fh = path.open("w")
    try:
        fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)   # LOCK_NB → raise, don't wait
    except BlockingIOError:
        fh.close()
        raise SystemExit("previous run still active — skipping this cron fire")
    try:
        fh.write(str(os.getpid()))
        fh.flush()
        yield
    finally:
        fcntl.flock(fh, fcntl.LOCK_UN)
        fh.close()


def already_processed(asset: Path, done_dir: Path) -> bool:
    """Idempotency: a marker file per asset makes a re-driven window a no-op for done work."""
    return (done_dir / f"{asset.stem}.done").exists()


def mark_processed(asset: Path, done_dir: Path) -> None:
    (done_dir / f"{asset.stem}.done").write_text("ok", encoding="utf-8")


async def scan_queue(root: Path, done_dir: Path) -> AsyncIterator[Path]:
    """Lazily yield unprocessed assets — never materialize the whole backlog in a list."""
    for asset in sorted(root.glob("*.scc")):
        if not already_processed(asset, done_dir):
            yield asset
        await asyncio.sleep(0)                    # cooperative: let consumers run during a long scan


async def process_asset(asset: Path, done_dir: Path) -> None:
    """Stand-in for the real QC pass (parse + validate + write telemetry)."""
    await asyncio.sleep(0.05)                     # placeholder for I/O-bound validation work
    mark_processed(asset, done_dir)               # mark only AFTER success → crash-safe idempotency


async def worker(name: str, queue: asyncio.Queue, sem: asyncio.Semaphore, done_dir: Path) -> None:
    while True:
        asset = await queue.get()
        if asset is None:                         # poison pill → this worker drains and exits
            queue.task_done()
            return
        try:
            async with sem:                       # cap concurrent in-flight assets
                await process_asset(asset, done_dir)
        except Exception as exc:                  # a poison message must not kill the worker
            print(f"[{name}] FAILED {asset.name}: {exc}", file=sys.stderr)
        finally:
            queue.task_done()


async def run(root: Path, done_dir: Path) -> None:
    done_dir.mkdir(parents=True, exist_ok=True)
    queue: asyncio.Queue = asyncio.Queue(maxsize=QUEUE_BOUND)   # bounded → producer backpressure
    sem = asyncio.Semaphore(MAX_WORKERS)
    workers = [asyncio.create_task(worker(f"w{i}", queue, sem, done_dir)) for i in range(MAX_WORKERS)]

    stopping = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGTERM, stopping.set)       # graceful drain, not a hard kill
    loop.add_signal_handler(signal.SIGINT, stopping.set)

    async for asset in scan_queue(root, done_dir):
        if stopping.is_set():
            break                                 # stop enqueueing new work; drain what's in flight
        await queue.put(asset)                    # blocks here when the queue is full = backpressure

    for _ in workers:
        await queue.put(None)                     # one poison pill per worker
    await queue.join()                            # wait until every queued asset is done
    await asyncio.gather(*workers)


if __name__ == "__main__":
    with run_lock(LOCK_PATH):                     # whole run guarded by the exclusive lock
        asyncio.run(run(Path("/ingest/queue"), Path("/ingest/.done")))

Code walkthrough

run_lock is the overrun guard, and it wraps the entire run. fcntl.flock with LOCK_EX | LOCK_NB takes an exclusive, non-blocking lock on a file the first process holds for its whole lifetime. When cron fires a second copy while the first is still clearing a backlog, the second flock raises BlockingIOError immediately, and the runner exits with a message instead of waiting or, worse, running a second scan over the same queue. The lock is an OS-level advisory lock tied to the open file descriptor, so it releases automatically if the process is killed — there is no stale lockfile to clean up manually.

MAX_WORKERS and QUEUE_BOUND are the two numbers that keep memory flat. QC work is I/O-bound — parse a file, run checks, write a row — so the worker count is 2 x cores capped at 8; past that, more coroutines contend for the same disk without finishing faster. The asyncio.Semaphore(MAX_WORKERS) is a hard ceiling on assets in flight: even though there are MAX_WORKERS worker coroutines, the semaphore guarantees no more than MAX_WORKERS assets are simultaneously loaded and validated, so peak resident memory is a function of the ceiling, not the backlog size. This is the same bounded-concurrency discipline documented in async batch caption processing, applied to a scheduled trigger.

The asyncio.Queue(maxsize=QUEUE_BOUND) is where backpressure lives. The producer — scan_queue, driving the async for loop in run — calls await queue.put(asset), and when the queue is full that await blocks until a worker frees a slot. The scan literally cannot run ahead of processing, so a million-asset backlog never becomes a million-element in-memory list. Contrast a naive asyncio.gather(*[process(a) for a in all_assets]): that schedules every asset at once and defeats the semaphore’s memory bound because every coroutine’s stack and buffers are allocated up front. Making scan_queue an async generator that yields lazily is what keeps the backlog on disk instead of in RAM.

Idempotency is enforced by already_processed / mark_processed, and the ordering matters: mark_processed runs only after process_asset succeeds. If the box is killed mid-asset, that asset has no .done marker and is re-processed on the next fire — at-least-once with a crash-safe marker, never a silent skip of unfinished work. scan_queue checks the marker as it yields, so a re-driven window is a no-op for everything already finished, exactly the property the parent scheduler needs when it re-runs a date range against the Parquet telemetry schema.

Shutdown is graceful. The SIGTERM and SIGINT handlers set an asyncio.Event rather than tearing the loop down; the producer sees stopping.is_set() and stops enqueueing, then the runner sends one None poison pill per worker so each drains its current asset and exits cleanly. await queue.join() blocks until every enqueued asset has called task_done, guaranteeing no half-written report when the orchestrator (Kubernetes, systemd) sends its termination signal. A per-worker try/except around process_asset means a single poison message — a corrupt file that raises — is logged and skipped, never allowed to kill the worker and stall the drain.

Threshold reference table

Parameter	Value / formula	Rationale
`MAX_WORKERS`	`min(8, cores x 2)`	I/O-bound ceiling on in-flight assets
Queue bound (`maxsize`)	`MAX_WORKERS x 4`	backpressure buffer before the scan blocks
Lock type	`flock` `LOCK_EX \| LOCK_NB`	fail-fast on overlapping cron fire
Lock timeout	none (non-blocking)	second fire exits, never queues
Run cadence	fire interval > p95 run time	headroom so runs rarely overlap
Marker write	after success only	crash-safe at-least-once idempotency
Drain signal	`SIGTERM` → drain, no hard kill	no half-written telemetry
Poison-message policy	log + skip, keep worker alive	one bad asset never stalls the batch

The QC checks each worker runs — reading-rate, drift, syntax — carry their own regulatory thresholds, enumerated in the scheduled QC reporting reference and enforcing character-rate limits in QC; this runner is the concurrency shell around them, not the validator itself. Retention and audit obligations for the telemetry it drives trace to 47 CFR § 79.1.

Edge cases & known gotchas

Cron overrun. If a run routinely takes longer than the cron interval, the lock silently drops every other fire and the backlog only grows. Alert on repeated “previous run still active” exits — that message is the signal to raise MAX_WORKERS, shard the queue, or lengthen the interval, not to remove the lock.
Partial failures. A worker that raises on one asset must not abort the batch. The try/except logs and continues, and because mark_processed never ran, the failed asset is retried next fire — but a persistently failing asset will retry forever unless you add a failure counter and route it to a dead-letter directory.
Poison messages. A truncated or malformed file can raise on parse; caught and skipped here, it retries indefinitely. Cap retries by writing a .attempts sidecar and quarantining an asset after N failures so one bad file cannot occupy a worker slot every run.
SIGTERM drain window. Kubernetes sends SIGTERM then SIGKILL after terminationGracePeriodSeconds (default 30 s). If a single asset’s QC pass can exceed that, either raise the grace period or checkpoint mid-asset, or the drain is cut short and that asset is re-processed.
Lock file on a network mount. flock semantics are unreliable over NFS; keep LOCK_PATH on a local filesystem (/var/run, /run) so the exclusive lock is honored. A lock on shared storage can be granted twice and let two runs overlap anyway.

Integration hook

This runner is the execution shell that drives the scheduled QC reporting job on a fixed cadence: cron fires it, the lock guarantees one run, and each worker performs the parse-validate-record pass whose output lands in the Parquet telemetry schema for the warehouse and is projected into the compliance dashboard JSON for the front end. Because processing is idempotent per asset and the writer is idempotent per window, re-driving any backlog is safe — the runner and the report share the same at-least-once, marker-guarded contract, and its concurrency model extends the pool discipline of async batch caption processing to a scheduled trigger.

Scheduled QC report generation — parent reference: the idempotent report job this runner invokes on a cadence.
Async batch caption processing — the bounded worker-pool and queue model this scheduled runner builds on.
Parquet telemetry schema design — the idempotent-per-window writer that consumes each worker’s output.
Compliance dashboard JSON output format — the versioned contract the runner’s telemetry feeds.

Part of: Automated QC Validation & Reporting — the deterministic caption QC and reporting reference.