# Asyncio patterns for processing 100+ RFPs overnight

When a quarter-end funding cycle drops a hundred-plus solicitations across the National Institutes of Health (NIH), the National Science Foundation (NSF), and the Department of Defense (DoD) in a single day, a serial pipeline that fetches, parses, and validates one Request for Proposals (RFP) at a time cannot clear the queue before research administrators arrive the next morning — and every solicitation left unprocessed is a compliance matrix that a downstream validator never sees and a deadline nobody flags. This page is the concurrency-internals companion to async batch processing for large RFPs: where the parent workflow establishes the offload-to-a-process-pool architecture, this reference tunes the event loop itself — bounded concurrency, rate-limit back-pressure, structured cancellation, and audit-safe instrumentation — so an unattended overnight run of a hundred documents finishes clean, stays inside the host’s memory ceiling, and resolves every failure to a routed record rather than a crash. The four phases below move from shaping the run, through the core ingestion-and-validation loop, into the agency-specific edge cases, and end at the verification that clears the batch for handoff.

# Phase 1 — Shape the overnight run and bound its concurrency

The first mistake at scale is spawning one task per document. A hundred asyncio.create_task calls launched at once will blow past federal portal rate limits, exhaust the host’s file-descriptor ceiling, and hold a hundred response bodies resident simultaneously. The run must be shaped before a single request leaves the machine.

Implementation steps:

1. Build a work queue, not a task fan-out. Load the hundred-plus (rfp_id, url) pairs into a list or asyncio.Queue so the orchestrator controls admission rather than the OS scheduler.
2. Cap in-flight requests with a semaphore. An asyncio.Semaphore sized to the slowest agency’s rate limit is the single throttle that keeps the run inside every portal’s tolerance and the host’s socket budget.
3. Share one connection pool. A single aiohttp.ClientSession reuses TCP connections and its own connector limit across the whole batch; opening a session per request defeats keep-alive and multiplies the handshake cost a hundredfold.
4. Separate I/O concurrency from CPU parallelism. Network fetches belong on the event loop; the CPU-bound work — pdfplumber text extraction and regex-heavy form matching — is offloaded to a process pool via loop.run_in_executor() so a synchronous parser never freezes scheduling for the rest of the batch.

Encapsulating the session lifecycle and the concurrency cap in one async context manager makes the bound explicit and guarantees the pool is closed even if the run is cancelled mid-flight:

import asyncio
import aiohttp
from typing import Any


class RFPIngestionPipeline:
    """Bounded async client for an overnight RFP fetch run."""

    def __init__(self, max_concurrency: int = 15, timeout: float = 30.0) -> None:
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self.session: aiohttp.ClientSession | None = None

    async def __aenter__(self) -> "RFPIngestionPipeline":
        # One session, one connection pool, shared across every task.
        self.session = aiohttp.ClientSession(timeout=self.timeout)
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        if self.session:
            await self.session.close()

With the run shaped this way, max_concurrency becomes the one dial that trades throughput against politeness, and every request inherits the same timeout and pool without per-call bookkeeping.

# Phase 2 — Fetch, retry, and validate inside the bound

The core transformation is the per-document coroutine: acquire a semaphore slot, fetch under a retry-with-backoff loop that respects HTTP 429/503 responses, then route the payload through a deterministic compliance suite. Federal endpoints such as Grants.gov, NIH RePORTER, and DoD Broad Agency Announcement (BAA) portals return 429 (Too Many Requests) or 503 the moment request velocity exceeds their threshold, so a fetch that does not back off will trigger an automated IP block partway through the night.

async def fetch_solicitation(
    self, rfp_id: str, url: str
) -> dict[str, Any]:
    """Fetch one solicitation with bounded retries and exponential backoff."""
    async with self.semaphore:  # never exceed the in-flight cap
        for attempt in range(3):
            try:
                async with self.session.get(url) as resp:
                    resp.raise_for_status()
                    payload = await resp.text(encoding="utf-8")
                    return {"id": rfp_id, "status": "success", "payload": payload}
            except aiohttp.ClientResponseError as exc:
                # Only 429/503 are worth retrying; back off, then try again.
                if exc.status in (429, 503):
                    await asyncio.sleep(2 ** attempt)
                    continue
                return {"id": rfp_id, "status": "failed", "error": str(exc)}
            except asyncio.TimeoutError:
                return {"id": rfp_id, "status": "timeout", "error": "request timed out"}
    return {"id": rfp_id, "status": "exhausted", "error": "max retries exceeded"}

The retry loop is deliberately narrow: it retries only the rate-limit and transient-unavailable statuses and never a 4xx client error, because retrying a malformed request or an authentication failure just wastes the night. The branching that governs each attempt is shown below.

Once a payload is in hand, validation must run deterministically. Federal solicitations carry non-negotiable structural requirements — mandatory forms such as the SF-424, specific page and font floors, and submission deadlines tied to an agency time zone — and asyncio.TaskGroup (Python 3.11+) runs the independent validators as structured concurrency, propagating the first failure and cancelling its siblings together.

import re
from dataclasses import dataclass, field
from datetime import datetime, timezone


@dataclass(slots=True)
class ComplianceResult:
    rfp_id: str
    is_compliant: bool
    violations: list[str] = field(default_factory=list)
    validated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


class ComplianceViolationError(Exception):
    """Raised when a solicitation fails a mandatory compliance check."""


async def validate_deadline(payload: str, rfp_id: str) -> None:
    match = re.search(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}", payload)
    if not match:
        raise ComplianceViolationError(f"missing submission deadline in {rfp_id}")
    deadline = datetime.fromisoformat(match.group(0)).replace(tzinfo=timezone.utc)
    if deadline <= datetime.now(timezone.utc):
        raise ComplianceViolationError(f"deadline already expired for {rfp_id}")


async def validate_mandatory_forms(payload: str, rfp_id: str) -> None:
    # SF-424 is required for most federal grants; add only the forms your
    # agency mix actually mandates (PHS-398 for NIH, DD-1494 for some DoD).
    required = ["SF-424"]
    missing = [name for name in required if name not in payload.upper()]
    if missing:
        raise ComplianceViolationError(f"missing forms in {rfp_id}: {', '.join(missing)}")


async def run_validation_suite(rfp_id: str, payload: str) -> ComplianceResult:
    violations: list[str] = []
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(validate_deadline(payload, rfp_id))
            tg.create_task(validate_mandatory_forms(payload, rfp_id))
    except* ComplianceViolationError as eg:
        violations = [str(e) for e in eg.exceptions]
    return ComplianceResult(rfp_id, not violations, violations)

The except* (exception-group) syntax is what makes TaskGroup usable for compliance: rather than letting the first violation cancel the batch, it collects every validator’s complaint into one ComplianceResult so a solicitation that is both past deadline and missing a form is flagged once with both reasons. The section-detection heuristics that identify which bytes hold the deadline and form references belong to NLP section-boundary detection, and the structural coercion of the result belongs to the schema validation with Pydantic layer.

# Phase 3 — Agency overrides, back-pressure, and structured cancellation

Three classes of edge case separate a demo from a pipeline that survives an unattended night across NIH, NSF, and DoD.

Per-agency rate limits and document sizes. A single global max_concurrency is wrong because the agencies do not behave alike: DoD BAA appendices run to hundreds of scanned pages and demand a lower concurrency with a longer timeout, while NSF solicitations are small enough to poll more aggressively. Drive the cap from a per-agency profile rather than a constant, cross-referencing the volume limits captured during DoD BAA requirement extraction.

Time-zone-correct deadlines. A deadline compared in the wrong zone silently passes a solicitation that is actually expired. Convert every extracted deadline to Coordinated Universal Time (UTC) before comparison, and persist the original offset in the compliance record so an auditor can reconstruct the local cutoff months later. This is the same authority-of-record discipline the NIH FOA schema mapping process applies to page and font ceilings.

Back-pressure and structured cancellation. The semaphore already bounds how many response bodies are resident, but two failure modes remain. First, CPU-bound extraction must be offloaded — a synchronous pdfplumber call awaited on the loop starves every other coroutine — so route it through a ProcessPoolExecutor with loop.run_in_executor(), sized to min(cpu_count, memory_budget // peak_doc_mb) so a 400-page BAA cannot exhaust the heap. Second, an overnight service will eventually catch a SIGTERM from a deploy or a watchdog, and a run that dies mid-flight must not lose the documents it already validated. Wrapping the batch in a top-level TaskGroup and catching asyncio.CancelledError lets the run persist partial state before it exits.

async def run_overnight(pipeline: "RFPIngestionPipeline", queue: list[dict]) -> None:
    """Drain the queue, persisting partial results on cancellation."""
    done: list[ComplianceResult] = []
    try:
        async with asyncio.TaskGroup() as tg:
            for rfp in queue:
                tg.create_task(_ingest_one(pipeline, rfp, done))
    except* asyncio.CancelledError:
        # SIGTERM or watchdog fired: flush what completed before shutdown.
        await persist_partial(done)
        raise
    await persist_final(done)

# Phase 4 — Instrument, verify, and clear the batch for handoff

An overnight run has no operator watching it, so its trustworthiness rests entirely on what it records. print statements cannot survive an audit; correlate every request across coroutine boundaries with a contextvars.ContextVar and emit structured log records that a downstream compliance validation rule engine can reconcile against the payloads it later enforces.

import asyncio
import contextvars
import logging
from typing import AsyncIterator

request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id")
logger = logging.getLogger("rfp_pipeline")


async def process_batch(
    pipeline: "RFPIngestionPipeline", rfp_queue: list[dict[str, str]]
) -> AsyncIterator[ComplianceResult]:
    """Stream compliance results as each solicitation completes."""
    fetches = {
        asyncio.ensure_future(pipeline.fetch_solicitation(r["id"], r["url"])): r["id"]
        for r in rfp_queue
    }
    for future in asyncio.as_completed(fetches):
        payload = await future
        token = request_id.set(payload["id"])
        try:
            if payload["status"] == "success":
                result = await run_validation_suite(payload["id"], payload["payload"])
                logger.info("validated", extra={"rfp_id": payload["id"],
                                                 "compliant": result.is_compliant})
                yield result
            else:
                logger.warning("ingest failed", extra={"rfp_id": payload["id"],
                                                        "error": payload.get("error")})
        finally:
            request_id.reset(token)  # reset per result, never leak the context

Setting and resetting the ContextVar inside the per-result loop — rather than up front for the whole batch — is what keeps request_id correct under as_completed, where results arrive in completion order, not submission order. Every log line then carries the id of the solicitation that actually produced it.

Before the batch is released to the persistence layer and the compliance engines, walk a short acceptance checklist that a happy-path run will pass right over:

• Every queued rfp_id appears exactly once in the streamed results — no document silently dropped.
• The concurrency cap held: peak in-flight requests never exceeded max_concurrency under the full hundred-document load.
• Retries fired only on 429/503, and no run exhausted its retry budget against a healthy portal (a sign the cap is too high).
• Every deadline was compared in UTC, with the original offset preserved in the record.
• A simulated SIGTERM mid-run persisted partial results and exited cleanly, losing nothing already validated.
• Wall-clock time scaled sub-linearly with document count against a serial baseline.

Only when those hold is the overnight batch safe to hand to the assembly and submission stages that depend on it.

# Frequently asked questions

# Related

• Async batch processing for large RFPs — the parent workflow whose process-pool architecture this page tunes at the event-loop level.
• PDF text extraction with pdfplumber — the synchronous, CPU-bound extractor offloaded to the process pool.
• Schema validation with Pydantic — the strict layer that coerces each validated solicitation into a compliance matrix.
• NLP section-boundary detection — locates the deadline, form, and eligibility sections the validators read.
• Compliance validation rule engines — the downstream consumer that enforces agency pass/fail rules against the structured output.

Up one level: Async batch processing for large RFPs