Tracking Batch Job Progress

A pipeline processing 50,000 records might run for hours. During that time, someone — an engineer debugging a slowdown, an operator deciding whether to wait or intervene, a dashboard showing stakeholders that things are moving — needs a reliable answer to "how far along is this, and is it healthy?" without reading raw logs or guessing. Progress tracking is the part of a pipeline responsible for answering that question at any moment, and it needs to be designed deliberately rather than bolted on as an afterthought with a single print(f"{i}/{total}") statement.

What Progress Tracking Actually Needs to Report

A useful progress report for a bulk AI processing job needs more than a raw completion count. At minimum, it should track:

• Total items in the job.
• Completed items, broken down by outcome (succeeded vs. failed), not lumped into one "done" number — a job that is 90% "done" but where a third of that is failures is in a very different state than one that's 90% succeeded.
• In-progress items currently being processed.
• Pending items not yet started.
• Elapsed time and, from it, an estimated time remaining.

from dataclasses import dataclass, field
import time


@dataclass
class ProgressTracker:
    total: int
    succeeded: int = 0
    failed: int = 0
    in_progress: int = 0
    started_at: float = field(default_factory=time.monotonic)

    @property
    def completed(self) -> int:
        return self.succeeded + self.failed

    @property
    def pending(self) -> int:
        return self.total - self.completed - self.in_progress

    @property
    def elapsed_seconds(self) -> float:
        return time.monotonic() - self.started_at

    @property
    def estimated_remaining_seconds(self) -> float:
        if self.completed == 0:
            return float("inf")
        rate = self.completed / self.elapsed_seconds
        return self.pending / rate if rate > 0 else float("inf")

    def report_started(self, count: int = 1) -> None:
        self.in_progress += count

    def report_succeeded(self, count: int = 1) -> None:
        self.in_progress -= count
        self.succeeded += count

    def report_failed(self, count: int = 1) -> None:
        self.in_progress -= count
        self.failed += count

    def summary(self) -> str:
        eta = self.estimated_remaining_seconds
        eta_str = "unknown" if eta == float("inf") else f"{eta:.0f}s"
        return (
            f"{self.completed}/{self.total} done "
            f"({self.succeeded} ok, {self.failed} failed), "
            f"{self.pending} pending, ETA {eta_str}"
        )

A few design choices here are deliberate and worth explaining. time.monotonic() is used instead of time.time() because it is guaranteed never to go backward (it isn't tied to the system's wall clock, which can be adjusted by NTP or a manual clock change) — for measuring elapsed durations, monotonic time is always the correct choice. estimated_remaining_seconds divides pending items by the observed completion rate so far rather than assuming a fixed per-item time, which makes the ETA self-correct as the job's actual throughput becomes clearer — an ETA computed from only the first few completed items will be noisy, but it stabilizes as completed grows. Separating succeeded and failed inside completed means a caller can immediately see whether the job is on track or quietly accumulating failures, which feeds directly into the failure-handling decisions in Lesson 7.

Wiring Progress Tracking into the Worker Pool

The tracker needs to be updated from inside the concurrent worker logic introduced in Lesson 4, at the points where an item starts and finishes:

import asyncio


async def tracked_worker(name, queue, client, results, tracker: ProgressTracker):
    while True:
        record = await queue.get()
        if record is None:
            queue.task_done()
            break

        tracker.report_started()
        try:
            response = await client.responses.create(
                model="gpt-5.6-terra",
                input=record.prompt,
            )
            record.model_response = response.output_text
            record.status = RecordStatus.SUCCEEDED
            tracker.report_succeeded()
        except Exception as exc:
            record.error = str(exc)
            record.status = RecordStatus.FAILED
            tracker.report_failed()

        results.append(record)
        queue.task_done()

Because report_started, report_succeeded, and report_failed are called from within concurrently running coroutines, and Python's asyncio guarantees that only one coroutine actually executes at any given instant (cooperative multitasking, not true parallel threads), simple attribute increments like these are safe without an explicit lock — there is no await point inside the increment itself where another coroutine could interleave and corrupt the count. This would not be true if the same ProgressTracker were shared across multiple OS threads or processes, which would require a real lock or a thread-safe counter instead.

Periodic Reporting

A tracker that's only ever inspected when something goes wrong isn't very useful. Pair it with a background coroutine that prints or logs a summary on a fixed interval for the life of the job:

async def report_progress_periodically(tracker: ProgressTracker, interval: float = 5.0):
    while tracker.pending > 0 or tracker.in_progress > 0:
        print(tracker.summary())
        await asyncio.sleep(interval)
    print(tracker.summary())  # final report


async def run_job_with_progress(records, concurrency=10):
    tracker = ProgressTracker(total=len(records))
    client = AsyncOpenAI()
    queue: asyncio.Queue = asyncio.Queue()
    results: list = []
    for r in records:
        queue.put_nowait(r)

    workers = [
        asyncio.create_task(tracked_worker(f"w{i}", queue, client, results, tracker))
        for i in range(concurrency)
    ]
    reporter = asyncio.create_task(report_progress_periodically(tracker))

    await queue.join()
    for _ in workers:
        queue.put_nowait(None)
    await asyncio.gather(*workers)
    await reporter

    return results

Running report_progress_periodically as its own task alongside the workers, rather than calling it inline somewhere in the worker loop, keeps the reporting cadence independent of how fast or slow individual items happen to complete — you get a report every five seconds regardless of whether that interval contained one completion or fifty.

Using tqdm for Interactive Progress Bars

When a pipeline runs interactively (in a terminal, during development or a manually triggered run), a visual progress bar communicates status far more immediately than periodic text lines. The tqdm library integrates cleanly with this pattern:

from tqdm import tqdm


async def run_job_with_progress_bar(records, concurrency=10):
    client = AsyncOpenAI()
    queue: asyncio.Queue = asyncio.Queue()
    results: list = []
    for r in records:
        queue.put_nowait(r)

    pbar = tqdm(total=len(records), desc="Processing records")

    async def worker_with_bar(queue, client, results):
        while True:
            record = await queue.get()
            if record is None:
                queue.task_done()
                break
            try:
                response = await client.responses.create(
                    model="gpt-5.6-terra", input=record.prompt
                )
                record.model_response = response.output_text
                record.status = RecordStatus.SUCCEEDED
            except Exception as exc:
                record.error = str(exc)
                record.status = RecordStatus.FAILED
            results.append(record)
            pbar.update(1)
            queue.task_done()

    workers = [
        asyncio.create_task(worker_with_bar(queue, client, results))
        for _ in range(concurrency)
    ]
    await queue.join()
    for _ in workers:
        queue.put_nowait(None)
    await asyncio.gather(*workers)
    pbar.close()
    return results

tqdm handles rendering an updating bar with a completion percentage and rate estimate in the terminal; pbar.update(1) should be called exactly once per completed item, which is why it sits at the same point in the loop where the record's final status is already decided. For an unattended production job (running on a server with no terminal to watch), the periodic-summary approach with ProgressTracker is more appropriate than a terminal progress bar, since a progress bar has no meaning in a log file — but nothing prevents using both in different contexts from the same underlying worker structure.

Persisting Progress for External Visibility

For a long-running unattended job, writing the current progress summary to a file or a database row that an external dashboard or health check can read is what actually makes the job's status "trackable" beyond whoever happens to be watching its logs at that moment:

import json


def write_progress_snapshot(tracker: ProgressTracker, path: str) -> None:
    snapshot = {
        "total": tracker.total,
        "succeeded": tracker.succeeded,
        "failed": tracker.failed,
        "pending": tracker.pending,
        "elapsed_seconds": round(tracker.elapsed_seconds, 1),
    }
    with open(path, "w") as f:
        json.dump(snapshot, f)

Calling this alongside the periodic reporter (writing to the same file every few seconds) turns the tracker into something a separate monitoring process, or a simple curl against a status endpoint backed by this file, can observe without needing access to the pipeline process itself.

Testing Progress Tracking

Because ProgressTracker has no I/O or async dependency, it's straightforwardly testable with plain synchronous assertions:

def test_progress_tracker_counts_correctly():
    tracker = ProgressTracker(total=10)
    tracker.report_started(3)
    tracker.report_succeeded(2)
    tracker.report_failed(1)

    assert tracker.completed == 3
    assert tracker.in_progress == 0
    assert tracker.pending == 7
    print("PASS: progress tracker counts started/succeeded/failed correctly")


test_progress_tracker_counts_correctly()

Common Mistakes

• Reporting only a single "done" count that merges successes and failures. This hides a job that's technically finishing but failing on a large fraction of items, which should be treated as a health signal, not a footnote.
• Computing ETA from a fixed assumed per-item duration instead of the observed rate. Actual per-item latency varies with prompt length, model load, and retries; an ETA that doesn't adapt to observed throughput becomes visibly wrong within minutes.
• Only checking progress by reading logs after the fact. Without a persisted, queryable snapshot, checking on a running job means grepping through scrolling log output, which doesn't scale past a handful of concurrent jobs or beyond the person who started it.

Best Practices

• Track succeeded, failed, in-progress, and pending as separate numbers, not a single aggregate, so that a degraded job is visible immediately rather than discovered only when it finishes with an unexpectedly high failure count.
• Persist a progress snapshot on a regular interval for any unattended job, so status can be checked externally without access to the running process or its console output.
• Use time.monotonic() for any elapsed-time or rate calculation, never wall-clock time, to avoid corrupted duration measurements from system clock adjustments.