Async Polling & Queue Management for DSR Pipelines

Within the broader Cross-System Data Discovery & Sync architecture, asynchronous polling is the control plane that keeps discovery inside its statutory clock. Data Subject Request (DSR) fulfillment operates under hard regulatory deadlines — the “without undue delay and in any event within one month” mandate of GDPR Article 12(3) and the 45-day baseline of CCPA §1798.130(a)(2) — yet the systems being queried are fragmented, rate-limited, and unpredictably slow. A synchronous fan-out that blocks a worker on the slowest vendor collapses under load: the pool starves, one throttled API stalls the whole request, and the SLA clock keeps ticking. This page addresses the specific gap between “discovery must query dozens of heterogeneous systems” and “discovery must finish within a legally bounded window” by decoupling ingestion from execution through bounded async polling, explicit priority routing, and deterministic failure categorization.

Discovery consumes attested, jurisdiction-scoped requests from the DSR Architecture & Intake Routing layer and dispatches one queued task per source system. Each task flows through payload validation, priority routing, bounded polling with backpressure, and explicit failure classification before its result is emitted to the audit trail:

Phase 1: Non-Blocking Ingestion & Polling Loops #

Polling loops must never block the main execution thread. Using asyncio with pooled connections lets a single worker run concurrent fetch cycles across relational databases, object stores, and SaaS endpoints, yielding during every network wait instead of holding a thread hostage to the slowest response. This is what makes it possible to respect each vendor’s rate ceiling independently while the overall request stays inside its GDPR Article 12(3) window. The official Python asyncio documentation describes the event-loop patterns that prevent thread starvation during high-throughput ingestion.

Exponential backoff is applied per task so that upstream throttling never triggers a retry storm. The connector coordinates with the typed pools defined in Database Connector Configuration so that concurrency limits and pool sizes are set from one place rather than drifting apart.

import asyncio

import aiohttp
from aiohttp import ClientResponseError
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=15),
    retry=retry_if_exception_type(ClientResponseError),
)
async def poll_endpoint(
    session: aiohttp.ClientSession,
    url: str,
    headers: dict[str, str],
    timeout: int = 15,
) -> dict:
    """Fetch one discovery endpoint without blocking the event loop.

    Retries only on ClientResponseError (transient upstream faults);
    permanent failures propagate to the caller for quarantine.
    """
    async with session.get(
        url, headers=headers, timeout=aiohttp.ClientTimeout(total=timeout)
    ) as resp:
        resp.raise_for_status()
        return await resp.json()

Phase 2: Priority Routing & Concurrency Control #

DSR pipelines handle heterogeneous workloads with sharply different regulatory urgency, so routing must be explicit rather than first-in-first-out. Access requests under GDPR Article 15 can tolerate the full one-month window, but erasure requests under GDPR Article 17 carry the added obligation to propagate the deletion to recipients “without undue delay” — meaning a deletion task should never wait behind a large access backlog. Priority queues let high-urgency deletion tasks preempt standard discovery so erasure mandates are satisfied within their timeframe.

Worker concurrency must be strictly bounded. An unbounded semaphore exhausts the connection pool, trips upstream rate limits, and produces the exact 429/503 storms that async polling exists to avoid. A per-run asyncio.Semaphore caps in-flight tasks; for distributed orchestration across many workers, Implementing Celery for async polling supplies named priority queues, rate-limiting middleware, and predictable worker scaling.

import asyncio
from collections.abc import Awaitable, Callable


async def run_bounded(
    tasks: list[Callable[[], Awaitable[dict]]],
    max_concurrency: int = 8,
) -> list[dict]:
    """Execute discovery tasks with a hard ceiling on in-flight requests."""
    semaphore = asyncio.Semaphore(max_concurrency)

    async def _guard(task: Callable[[], Awaitable[dict]]) -> dict:
        async with semaphore:
            return await task()

    return await asyncio.gather(*(_guard(t) for t in tasks))

Phase 3: Payload Validation & Schema Enforcement #

Queue consumers must validate every payload before any extraction logic runs. A malformed request that slips through can corrupt a downstream state machine, raise a false compliance alert, or cause silent data loss — all of which are reportable failures under an accountability regime. Strict type enforcement and field-presence checks act as an immutable gatekeeper at the ingestion boundary, mirroring the discovery-wide contract described in Schema Validation Rules. The Pydantic documentation covers the v2 validation idioms used below.

from pydantic import BaseModel, ConfigDict, EmailStr, ValidationError, field_validator
from typing import Literal


class DSRTaskPayload(BaseModel):
    """Validated unit of work handed to a queue consumer."""

    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    request_id: str
    subject_email: EmailStr
    action: Literal["access", "deletion", "rectification"]
    source_system: str
    priority: int = 1

    @field_validator("priority")
    @classmethod
    def clamp_priority(cls, v: int) -> int:
        """Clamp priority into the 1-10 band the router understands."""
        return max(1, min(10, v))


def validate_task(raw_data: dict) -> DSRTaskPayload:
    """Reject malformed payloads before they reach extraction logic."""
    try:
        return DSRTaskPayload(**raw_data)
    except ValidationError as exc:
        raise ValueError(f"Invalid DSR payload: {exc}") from exc

Setting extra="forbid" enforces data minimization at the boundary: any undeclared field is rejected rather than silently persisted, keeping the pipeline aligned with GDPR Article 5(1)©.

Phase 4: Retry Logic & Failure Categorization #

Blind retries violate SLA boundaries, exhaust API quotas, and obscure root-cause analysis. A retry policy must classify failures explicitly rather than retrying everything the same number of times. Transient faults — 5xx responses, DNS timeouts, connection resets — warrant exponential backoff with jitter so recovering vendors are not hammered in lockstep. Permanent faults — 401 unauthorized, 422 schema mismatch, revoked OAuth tokens — must be quarantined immediately, because retrying them only burns the clock. Aligning SaaS API Sync Strategies with this classification reduces false-positive SLA breaches and keeps the audit trail clean.

from aiohttp import ClientResponseError

TRANSIENT_STATUS = {429, 500, 502, 503, 504}


def is_transient(exc: Exception) -> bool:
    """Classify a fault as retryable (transient) or terminal (permanent)."""
    if isinstance(exc, ClientResponseError):
        return exc.status in TRANSIENT_STATUS
    return isinstance(exc, (asyncio.TimeoutError, ConnectionError))

Phase 5: Dead-Letter Handling & Compliance Auditing #

Tasks that exhaust their retry budget must be routed to a dead-letter queue (DLQ). A DLQ is not merely a technical fallback — under an accountability regime it is a regulated compliance artifact. Failed payloads require secure, immutable storage for regulatory auditing and manual triage, so DLQ storage must carry retention policies, encryption-at-rest keyed per NIST SP 800-57 guidance, and role-based access controls. Crucially, a DLQ entry preserves the proof of process a supervisory authority expects: which system failed, the terminal status code, and evidence that the failure was escalated rather than dropped.

Operational metrics derived from these queues — retry rates, DLQ volume, per-vendor latency — must be aggregated over hashed identifiers before reporting to leadership, so internal analytics never expose subject identifiers or request patterns. When a soft internal deadline is breached, the pipeline emits an escalation event that gives compliance a documented basis to invoke the GDPR Article 12(3) two-month extension or the CCPA §1798.130 90-day extension while the successful systems stage as a partial manifest.

Edge Cases & Conflict Resolution #

Real traffic breaks clean phase boundaries, and the queue layer must resolve the ambiguities rather than defer them:

• Conflicting priority signals. A request tagged access by the subject but flagged deletion by an upstream policy engine must resolve to the higher-urgency action. The router treats erasure as dominant, since delaying a GDPR Article 17 obligation carries greater regulatory exposure than expediting an access read.
• Duplicate task delivery. At-least-once queue semantics mean the same request_id + source_system pair can arrive twice. Consumers must be idempotent, deduplicating on that composite key so a redelivered task never double-writes to the audit trail.
• Poison messages. A payload that repeatedly crashes the consumer (not merely fails the upstream call) must be routed straight to the DLQ after a bounded crash count, so one malformed message cannot wedge an entire consumer group.
• UNKNOWN source system. If a task names a system not present in the signed connector registry, it fails closed to quarantine rather than polling an unvetted endpoint — over-reaching to an unregistered system would breach the read-only, scoped-discovery contract.

Performance & Scale Considerations #

Throughput is bounded by the strictest upstream rate limit, not by worker count, so scaling means partitioning cleanly rather than adding unbounded concurrency. Partition the work queue by source_system so each vendor’s tasks land on a dedicated consumer group and one throttled API cannot backlog others. A Redis-backed rate limiter shared across workers enforces a global token budget per vendor, preventing the aggregate request rate from exceeding a ceiling that any single worker cannot see on its own.

Cache short-lived discovery cursors and OAuth tokens in Redis with a TTL below their expiry so token-refresh traffic does not itself become a rate-limit source. Monitor consumer lag per partition: rising lag on the high-priority deletion queue is the earliest signal that erasure SLAs are at risk, and it should trigger autoscaling of that queue’s consumer group before any soft deadline is breached.

Testing & Compliance Verification #

Verification for a queue layer means proving both the happy path and the failure taxonomy, then asserting the compliance invariants hold under load:

• Payload matrix. Test valid access/deletion/rectification payloads, plus rejected cases: missing subject_email, out-of-band priority, undeclared extra fields, and an unknown source_system.
• Failure classification. Assert that each of 429/500/503 is retried and that 401/422 goes straight to the DLQ without consuming the retry budget.
• Idempotency regression. Redeliver an identical task and assert exactly one audit-trail record results.
• Deadline escalation. Simulate a persistently failing connector and assert an escalation event fires before the statutory limit rather than after.

import pytest


def test_permanent_failure_skips_retry() -> None:
    """A 422 must be classified terminal and never retried."""
    exc = ClientResponseError(request_info=None, history=(), status=422)
    assert is_transient(exc) is False


def test_rejects_undeclared_field() -> None:
    """extra='forbid' must reject over-collected fields (GDPR Art. 5(1)(c))."""
    with pytest.raises(ValueError):
        validate_task(
            {
                "request_id": "r-1",
                "subject_email": "a@example.com",
                "action": "access",
                "source_system": "crm",
                "unexpected": "over-collected",
            }
        )

Frequently Asked Questions #

Related #

• Implementing Celery for async polling — named priority queues, rate-limiting middleware, and worker scaling for distributed discovery.
• Database Connector Configuration — typed pools and concurrency limits that async polling coordinates with.
• SaaS API Sync Strategies — pagination, token refresh, and rate-limit handling for cloud connectors.
• Schema Validation Rules — the Pydantic gates that reject malformed payloads before fulfillment.
• Cross-System Data Discovery & Sync — the discovery stage this queue layer belongs to.