Async Privilege Batching

Continuous RBAC drift detection has to enumerate the privilege state of every database in the estate, on a schedule, without disturbing the workloads running on those databases. The naive approach — a synchronous loop that opens a connection, sweeps the catalog, and moves to the next host — fails in a specific and expensive way at scale. The failure scenario this page prevents is scan-induced contention: a nightly compliance job fans out serially across a few hundred instances, each catalog sweep holds a connection and a shared lock on system relations longer than expected, connection pools saturate, application queries start timing out, and the on-call engineer kills the scan. Once the scan is unreliable, the compliance snapshot goes stale, drift accumulates silently, and the audit trail develops gaps exactly where auditors look first. Async privilege batching removes that contention by decoupling extraction from downstream processing: a bounded number of read-only catalog reads run concurrently, each producing a deterministic batch that lands in a staging buffer, so extraction latency never couples to how long parsing, diffing, or remediation take. This is the ingestion front-end for Cross-Environment Privilege Extraction & Parsing, which normalizes the disparate grant syntaxes those batches carry into one canonical, diff-ready state representation.

Figure — Async privilege batching. A bounded semaphore caps concurrency while the orchestrator dispatches read-only catalog chunks and aggregates deterministic batches into a staging buffer, decoupling extraction latency from downstream processing.

Prerequisites and Scope

This page assumes a Python 3.11+ runtime and the async database drivers appropriate to each target: asyncpg 0.29+ for PostgreSQL 12–16, and aiomysql 0.2+ (or asyncmy) for MySQL 8.0+, whose role catalog (mysql.role_edges) does not exist before 8.0. The extraction principal needs only read access to the relevant catalog relations — on PostgreSQL, membership in pg_read_all_stats or explicit SELECT on pg_catalog views is sufficient, and pg_read_all_data widens object-grant visibility; on MySQL, SELECT on mysql.role_edges and the information_schema privilege views. No account used for batching should hold DDL or GRANT privileges; extraction is a read-only role by contract, which keeps the safety argument below trivial to make to an auditor.

Two scope boundaries matter. First, batching owns acquisition only — turning live catalogs into raw, per-environment privilege rows tagged with a batch identity. The moment those rows are structurally validated they leave this stage for the schema validation pipelines that verify role hierarchies and object ownership, and the vendor-specific DDL they carry is translated to canonical form by the cross-DB parser adapters. Second, the shape of each catalog query is not this page’s concern — reducing per-chunk I/O and avoiding sequential scans is governed by System Catalog Query Optimization, and batching simply dispatches whatever tuned query that work produces.

Core Implementation Walkthrough

The engine is four moving parts: a chunk planner that partitions each environment’s catalog into non-overlapping work units, a bounded semaphore that caps concurrency below the pool ceiling, a per-chunk fetcher that runs the read-only query, and a deterministic identity function that stamps every batch so repeated runs are recognizable.

1. Partition each environment into non-overlapping chunks

A chunk is the smallest unit of work the semaphore schedules. Chunking by schema (or by role prefix, or privilege class) keeps each query bounded, lets a single failed chunk retry without re-reading the whole catalog, and makes batch identity stable. The planner reads only the list of schemas — a cheap catalog query — then emits one chunk descriptor per schema.

from __future__ import annotations
import dataclasses

@dataclasses.dataclass(frozen=True, slots=True)
class Chunk:
    environment: str          # logical env name, e.g. "prod-us-east"
    engine: str               # "postgresql" | "mysql"
    schema: str               # partition key
    window_start: str         # ISO-8601 timestamp bounding this run

async def plan_chunks(conn, environment: str, window_start: str) -> list[Chunk]:
    """Return one read-only work unit per non-system schema."""
    rows = await conn.fetch(
        """
        SELECT n.nspname AS schema
        FROM pg_catalog.pg_namespace n
        WHERE n.nspname NOT LIKE 'pg\\_%'
          AND n.nspname <> 'information_schema'
        ORDER BY n.nspname
        """
    )
    return [
        Chunk(environment=environment, engine="postgresql",
              schema=r["schema"], window_start=window_start)
        for r in rows
    ]

The NOT LIKE 'pg\_%' predicate (with the underscore escaped) drops the built-in schemas so the planner never emits chunks for catalogs that only contain system roles.

2. Cap concurrency with a bounded semaphore

The concurrency limit is the single most important safety control. It must sit strictly below the smallest connection-pool ceiling across the estate, because every in-flight chunk holds one connection. An asyncio.Semaphore sized to that budget guarantees the batcher can never open more catalog connections than the pool allows, no matter how many chunks the planner produced. The bounded-concurrency pattern here follows the primitives in the official Python asyncio documentation.

import asyncio
import asyncpg   # PostgreSQL async driver

async def fetch_chunk(chunk: Chunk, pool: asyncpg.Pool,
                      sem: asyncio.Semaphore) -> "Batch":
    async with sem:                      # never exceed the concurrency budget
        async with pool.acquire() as conn:
            async with conn.transaction(readonly=True):
                rows = await conn.fetch(
                    """
                    SELECT grantee, table_schema, table_name,
                           privilege_type, is_grantable
                    FROM information_schema.role_table_grants
                    WHERE table_schema = $1
                    ORDER BY grantee, table_name, privilege_type
                    """,
                    chunk.schema,
                )
    return build_batch(chunk, rows)

async def run_environment(env: str, dsn: str, window_start: str,
                          concurrency: int = 8) -> list["Batch"]:
    pool = await asyncpg.create_pool(dsn, min_size=1, max_size=concurrency)
    sem = asyncio.Semaphore(concurrency)
    try:
        async with pool.acquire() as c:
            chunks = await plan_chunks(c, env, window_start)
        return await asyncio.gather(
            *(fetch_chunk(ch, pool, sem) for ch in chunks)
        )
    finally:
        await pool.close()

The readonly=True transaction is not decoration — it makes the read-only contract enforceable at the database rather than trusted in code, so a GRANT/REVOKE accidentally reaching this path fails loudly instead of mutating state. The deterministic ORDER BY guarantees two runs over an unchanged catalog produce byte-identical row streams, which is what makes the batch identity in the next step stable.

3. Stamp each batch with a deterministic identity

Idempotency is enforced at acquisition. Every batch carries an identifier derived from the chunk descriptor plus a content hash of its sorted rows. Two runs over an unchanged schema yield the same batch_id and the same content_sha256; a run over a schema whose grants changed yields the same batch_id (same partition) but a different content hash. Downstream stages use that pair to deduplicate work — a batch whose content hash already exists in the staging buffer is skipped without a redundant parse or diff.

import dataclasses
import hashlib
import json

@dataclasses.dataclass(frozen=True, slots=True)
class Batch:
    batch_id: str          # stable per (environment, schema, window)
    content_sha256: str    # changes only when the grants change
    environment: str
    engine: str
    schema: str
    rows: tuple[dict, ...]

def build_batch(chunk: Chunk, rows) -> Batch:
    payload = [dict(r) for r in rows]
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    content_hash = hashlib.sha256(canonical.encode()).hexdigest()
    identity = f"{chunk.environment}:{chunk.schema}:{chunk.window_start}"
    batch_id = hashlib.sha256(identity.encode()).hexdigest()[:16]
    return Batch(
        batch_id=batch_id,
        content_sha256=content_hash,
        environment=chunk.environment,
        engine=chunk.engine,
        schema=chunk.schema,
        rows=tuple(payload),
    )

Because both hashes are pure functions of their inputs, a duplicate or replayed run produces an identical baseline state with no redundant network round-trips and no unintended mutation — the invariant that lets the scheduler run this stage as often as compliance demands.

4. Absorb transient failures without thundering herds

Catalog reads fail transiently — a failover, a brief pool exhaustion, a network blip. The retry policy must distinguish transient faults (retry with backoff) from structural ones (surface immediately). Exponential backoff with jitter prevents every chunk that failed during a failover from retrying in lockstep and re-saturating the recovering instance.

import asyncio
import random

TRANSIENT = (asyncpg.PostgresConnectionError, asyncio.TimeoutError,
             ConnectionResetError)

async def fetch_with_retry(chunk, pool, sem, *, max_attempts=5):
    for attempt in range(1, max_attempts + 1):
        try:
            return await fetch_chunk(chunk, pool, sem)
        except TRANSIENT as exc:
            if attempt == max_attempts:
                raise                       # to the dead-letter queue
            delay = min(2 ** attempt, 30) + random.uniform(0, 1.0)
            await asyncio.sleep(delay)      # backoff + jitter

Chunks that exhaust max_attempts, or that fail with a non-transient error such as a permission denial, are routed to a dead-letter queue for review rather than silently dropped — a missing environment in a compliance snapshot is itself a finding, not a gap to hide.

Figure — Chunk lifecycle. The happy path stamps a deterministic hash and buffers the batch; a transient fault re-queues through backoff without re-reading the whole catalog, while exhausted retries or a permission denial route to the dead-letter queue so a missing environment surfaces as a finding rather than a silent gap.

The MySQL Fetcher: Same Contract, Different Catalog

The four moving parts above are engine-agnostic by design — only the per-chunk fetcher changes between targets, because only the fetcher touches vendor catalogs. The PostgreSQL fetcher reads information_schema.role_table_grants inside an asyncpg read-only transaction; the MySQL equivalent reads information_schema.TABLE_PRIVILEGES (and, for role membership, mysql.role_edges) through aiomysql. The batch identity, chunk planner, semaphore, and retry wrapper are shared verbatim, which is what keeps a mixed PostgreSQL/MySQL estate on one code path. The dialect divergence in the grant rows those two fetchers emit is reconciled downstream by the cross-DB parser adapters, not here.

import aiomysql   # MySQL async driver, 0.2+

async def fetch_chunk_mysql(chunk: Chunk, pool: aiomysql.Pool,
                            sem: asyncio.Semaphore) -> "Batch":
    async with sem:                              # same concurrency budget
        async with pool.acquire() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                await cur.execute("SET SESSION TRANSACTION READ ONLY")
                await cur.execute(
                    """
                    SELECT GRANTEE, TABLE_SCHEMA, TABLE_NAME,
                           PRIVILEGE_TYPE, IS_GRANTABLE
                    FROM information_schema.TABLE_PRIVILEGES
                    WHERE TABLE_SCHEMA = %s
                    ORDER BY GRANTEE, TABLE_NAME, PRIVILEGE_TYPE
                    """,
                    (chunk.schema,),
                )
                rows = await cur.fetchall()
    return build_batch(chunk, rows)              # identical identity function

Three MySQL-specific facts change what this fetcher can see. SET SESSION TRANSACTION READ ONLY is the enforceable read-only contract on MySQL, standing in for asyncpg’s readonly=True — issue it before the read, and any stray GRANT on the connection errors instead of mutating. Role membership does not live in information_schema at all; it is in mysql.role_edges, which exists only on MySQL 8.0+, so a plan_chunks variant for MySQL must gate role-edge extraction on the server version and fall back to information_schema.SCHEMA_PRIVILEGES on 5.7. And MySQL’s information_schema privilege views reflect only granted roles, not the roles activated in the current session via SET ROLE; the extraction principal must therefore hold the visibility grants directly rather than through an unactivated role, or whole swaths of the estate’s grants become invisible to the scan. Because build_batch is called unchanged and aiomysql.DictCursor already yields plain dicts, a MySQL batch is byte-for-byte comparable to a PostgreSQL batch once the parser adapters have normalized the column names.

Estate-Level Fan-Out and the Two-Level Concurrency Budget

run_environment scans one instance. A real estate is hundreds of them, on mixed engines, and the scheduler must fan out across all of them on a schedule without letting the aggregate concurrency melt a shared network path, a bastion, or a central credentials broker. That calls for a two-level budget: a per-environment semaphore (already inside run_environment) that protects each instance’s own pool, and an outer, estate-wide semaphore that caps how many environments scan at once. The inner limit protects the databases; the outer limit protects everything between the scanner and them.

# One entry point per engine; each builds the right pool and fetcher but
# reuses the shared planner, semaphore, identity, and retry wrapper.
RUNNERS = {
    "postgresql": run_environment,          # asyncpg + fetch_chunk
    "mysql": run_environment_mysql,         # aiomysql + fetch_chunk_mysql
}

async def schedule_estate(environments: list[dict], window_start: str,
                          *, estate_limit: int = 16) -> dict[str, object]:
    """Fan out per-environment scans across the estate under a global cap."""
    estate_sem = asyncio.Semaphore(estate_limit)

    async def bounded_env(env: dict):
        async with estate_sem:                   # outer, estate-wide budget
            run = RUNNERS[env["engine"]]
            return await run(env["name"], env["dsn"], window_start,
                             concurrency=env["concurrency"])

    results = await asyncio.gather(
        *(bounded_env(e) for e in environments),
        return_exceptions=True,                   # one dead env ≠ dead snapshot
    )
    return {e["name"]: r for e, r in zip(environments, results)}

The return_exceptions=True is deliberate and load-bearing: a single unreachable instance must not abort the estate-wide snapshot. Each environment’s result is either its list of batches or the exception that stopped it, so the scheduler can persist every batch that succeeded and route the failures to the dead-letter queue — a partial snapshot with a named, explained gap is a compliance finding, whereas a snapshot that aborted on the first failure is just missing. Sizing the two limits is an operational decision, not a constant: set concurrency per environment strictly below that instance’s pool ceiling, and set estate_limit from the narrowest shared resource on the path — often the bastion’s connection cap or the credentials broker’s request budget, not the databases at all.

Figure — Two-level concurrency budget. The inner per-environment semaphore protects each database’s own pool; the outer estate-wide semaphore protects the shared path — bastion, network, credentials broker — between the scanner and every target. gather(return_exceptions=True) keeps one unreachable instance from aborting the estate snapshot: its batches persist while the failure lands in the dead-letter queue as a named gap.

Idempotency and Safety Contract

Batching is safe to run continuously because it satisfies three explicit properties. Read-only by construction: every catalog read runs inside a readonly=True transaction under a principal with no write grants, so the stage physically cannot mutate the databases it inspects — the dry-run and the “real” run are the same operation, and there is no separate destructive mode to guard. Convergent identity: the (batch_id, content_sha256) pair means repeated runs over an unchanged catalog are recognized as duplicates and short-circuited; the staging buffer converges to one row per partition regardless of how many times the scan fires. Deterministic ordering: the ORDER BY on every extraction query removes result-set ordering as a source of spurious hash differences, so a changed content hash always means a genuinely changed grant, never a reshuffled row.

To exercise the pipeline without touching the real staging buffer, point the buffer sink at an in-memory or scratch collection and inspect the emitted Batch objects — because acquisition is already read-only, this “dry run” reads production catalogs safely and its only difference from a live run is where the batches land. That property is what lets teams validate a new environment’s connectivity and permissions during a change window with zero blast radius.

Compliance Alignment

Continuous, non-disruptive privilege extraction is the evidence-generation step several controls depend on. It directly supports SOC 2 CC6.1 and CC6.3 (logical access is inventoried and reviewed) by producing a timestamped, per-environment record of who holds which grants; PCI-DSS Requirement 7 (restrict access by business need-to-know) by making the live grant set continuously observable rather than sampled at audit time; and HIPAA §164.312(a)(1) (access control) by evidencing that the privilege state of systems handling regulated data is inventoried on a schedule.

The artifact this stage produces for auditors is a per-batch JSON envelope: the deterministic batch_id, the content_sha256 proving the payload has not been altered since capture, the environment and engine, the extraction window_start, and the raw grant rows. Persisted append-only, that stream is a cryptographically verifiable inventory — an auditor can recompute any batch’s hash from its rows to confirm the record was not edited after capture. How much any deviation found in that inventory weighs toward an overall compliance score is decided downstream by Rule-Based Drift Scoring inside Drift Detection Engines & Diff Logic; batching’s contribution is the trustworthy, tamper-evident baseline everything else is measured against.

Troubleshooting Matrix

Operational Questions on Batch Scheduling

Should batching read from the primary or a read replica? Prefer a replica whenever the estate has one. Catalog reads are read-only, so replica lag never affects correctness of this stage — a grant that has not yet replicated simply appears in the next window’s batch, and the deterministic content hash makes that eventual-consistency delay visible rather than silent. Reading from replicas also removes catalog-scan load from the instance serving application writes, which is the whole point of decoupling extraction. The one exception is grants issued and revoked within a single replication interval; if second-level precision on ephemeral grants matters, read the primary for those specific environments and accept the extra load.

Why hash the sorted rows instead of trusting a catalog version or timestamp? Neither PostgreSQL nor MySQL exposes a reliable, monotonic “privileges last changed” marker at schema granularity. pg_catalog has no per-ACL modification timestamp, and MySQL’s grant tables carry none either. A content hash of the deterministically ordered rows is the only source of truth that is both cheap to compute and guaranteed to change exactly when a grant changes — which is why the content_sha256 in each batch, not any server-side clock, is what downstream deduplication keys on.

How does the concurrency budget interact with a transaction pooler like PgBouncer? In transaction pooling mode a client connection maps to a server connection only for the life of a transaction, so the semaphore should be sized to the pooler’s default_pool_size, not to the database’s max_connections. Because each fetcher holds its connection for exactly one read-only transaction, transaction pooling is a natural fit — the connection is released the instant the read commits, so a small pool absorbs far more chunks than a session-pooled setup would.

How often should the scan run? As often as compliance requires, because the idempotency contract makes over-scanning free of side effects: repeated runs over an unchanged catalog converge to one buffered batch per partition and cost only the read-only round-trips. Teams typically run a full estate sweep nightly for the audit baseline and a tighter loop over sensitive environments hourly, keying alerting off the batches whose content hash changed since the last window.

Implementing the Scraper End to End

The reference implementation that wires connection pooling, per-schema chunking, deterministic batch identity, and audit logging into one runnable program lives in Python scripts for async batch privilege scraping. It packages the four moving parts above into production-ready templates that abstract the PostgreSQL-vs-MySQL catalog differences while holding the read-only, least-privilege execution contract this page specifies.

Related

• Python scripts for async batch privilege scraping — the runnable end-to-end scraper implementing the chunking and batching contract described here.
• System Catalog Query Optimization — tuning the per-chunk extraction queries this engine dispatches so they avoid sequential scans and lock escalation.
• Schema Validation Pipelines — the structural integrity checks each batch flows into after acquisition.
• Cross-DB Parser Adapters — translating the vendor-specific grant syntax in each batch into a canonical, diff-ready format.
• Rule-Based Drift Scoring — weighting the deviations later found in the baseline this stage produces.

Up: Cross-Environment Privilege Extraction & Parsing