Coordinating Concurrent Polls Against Delaware’s Division of Corporations #

This guide belongs to the Async Polling & Rate Limiting area of the Secretary of State Portal & API Ingestion framework, and it addresses a problem the per-process token bucket in that area cannot solve on its own: when a dozen worker processes on separate hosts all poll Delaware’s Division of Corporations at once, no single worker knows how many requests the fleet has in flight. A local limiter caps each worker; nothing caps their sum. What Delaware’s portal actually enforces is a ceiling on simultaneous connections from an origin, so the control that matters is a single global in-flight count shared across every worker — a distributed concurrency gate.

What This Page Covers #

This page builds one thing: a fleet-wide gate that guarantees at most N requests are outstanding against the Delaware portal at any instant, regardless of how many workers are running. It uses a Redis-backed asynchronous semaphore with lease expiry so that a worker crashing mid-request cannot permanently leak a permit. It is not about pacing the interval between one worker’s requests — that steady-state cadence is the token bucket’s job, and adaptive cadence against a drifting limit is covered separately in Tuning Poll Intervals for California BizFile Rate Limits. Concurrency and rate are orthogonal knobs: you can be slow and still exceed a connection cap if enough workers are slow simultaneously, which is exactly what happens on the first of March.

The March 1 Surge That Forces a Global Cap #

Delaware fixes a single, immovable date that concentrates load. Domestic corporations owe franchise tax and the annual report by March 1 under 8 Del. C. § 502, with a $200 late penalty plus 1.5% monthly interest accruing under § 510, and continued default leading toward loss of good standing and eventual administrative action. Because every Delaware corporation shares that deadline, an entity-management platform verifying filing status across its portfolio hits the Division of Corporations hardest in the final days of February — precisely when the portal is already saturated by the rest of the country doing the same. Scaling out the worker fleet to clear the backlog faster is the obvious move and the wrong one if it is uncoordinated: ten workers each politely capped at two concurrent requests still put twenty connections on a portal that tolerates eight, and the whole origin gets throttled or temporarily blocked. The gate exists so the fleet can scale horizontally for throughput while the number of simultaneous requests stays pinned to a value the portal will accept.

Prerequisites #

• Python 3.10+ for X | Y unions, asyncio, and async context managers.
• redis 5.x (redis.asyncio) as the shared coordination store; a single-node instance is sufficient, and the Lua script keeps the acquire atomic.
• httpx 0.27+ for the async HTTP client each worker uses to reach the portal.
• Standard library asyncio, contextlib, json, logging, time, and uuid.
• Every worker in the fleet must point at the same Redis key; the gate is only global if the counter is shared.
• Deadline metadata for the § 502 obligation from the State Filing Deadline Calendars layer, so surge windows can widen the acquire budget in advance.

Implementation: A Redis-Backed Concurrency Gate #

The gate is a counting semaphore whose state lives in one Redis sorted set. Each in-flight request is a member scored by its acquisition timestamp; a permit is granted only while the set holds fewer than limit members, and members older than the lease TTL are reaped on every attempt so a dead worker’s permit is reclaimed automatically. Making the reap-check-add sequence atomic is essential — two workers evaluating “is there room?” independently will both say yes and overshoot the cap — so the acquire runs as a single Lua script inside Redis. The topology below shows a fleet larger than the cap funnelling through one gate to the portal.

The module is one runnable file: GateConfig holds the cap and timings, DistributedConcurrencyGate wraps the atomic acquire and release behind an async with context manager, DelawarePollWorker performs a single guarded poll, and run_fleet starts more workers than permits to prove the cap holds. The Lua script is the load-bearing part — everything else is orchestration around it.

from __future__ import annotations

import asyncio
import contextlib
import json
import logging
import time
import uuid
from dataclasses import dataclass
from typing import AsyncIterator

import httpx
import redis.asyncio as redis

# Structured JSON logging — every permit grant, denial, and release is an auditable event.
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("ingestion.delaware.gate")

# Atomic acquire: reap expired leases, then grant only if the fleet is below the cap.
# Running this server-side is what prevents two workers from both seeing room and overshooting.
_ACQUIRE_LUA = """
local key   = KEYS[1]
local now   = tonumber(ARGV[1])
local ttl   = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])
local token = ARGV[4]
redis.call('ZREMRANGEBYSCORE', key, 0, now - ttl)   -- reclaim permits leaked by dead workers
if redis.call('ZCARD', key) < limit then
    redis.call('ZADD', key, now, token)             -- take a permit, scored by acquisition time
    redis.call('PEXPIRE', key, ttl)                 -- backstop: whole set self-expires if idle
    return 1
end
return 0
"""


@dataclass(frozen=True)
class GateConfig:
    key: str = "sos:de:inflight"     # ONE key for the whole fleet — the cap is only global if shared
    limit: int = 8                   # max concurrent requests to Delaware across every worker
    lease_ms: int = 15_000           # a permit auto-expires if the holder crashes mid-request
    poll_backoff_s: float = 0.20     # cooperative wait between acquire attempts when saturated
    acquire_timeout_s: float = 30.0  # give up acquiring after this and surface upstream


class DistributedConcurrencyGate:
    """A Redis-backed counting semaphore enforcing one global in-flight cap for the fleet."""

    def __init__(self, client: redis.Redis, cfg: GateConfig | None = None) -> None:
        self._redis = client
        self._cfg = cfg or GateConfig()
        self._acquire = client.register_script(_ACQUIRE_LUA)

    async def _try_acquire(self, token: str) -> bool:
        now_ms = int(time.time() * 1000)
        granted = await self._acquire(
            keys=[self._cfg.key],
            args=[now_ms, self._cfg.lease_ms, self._cfg.limit, token],
        )
        return bool(granted)

    @contextlib.asynccontextmanager
    async def hold(self) -> AsyncIterator[str]:
        """Block until a global permit is free, then release it on exit even if the body raises."""
        token = uuid.uuid4().hex
        deadline = time.monotonic() + self._cfg.acquire_timeout_s
        attempts = 0
        while not await self._try_acquire(token):
            attempts += 1
            if time.monotonic() >= deadline:
                # The gate stayed full past our budget — do not force past the cap; surface it.
                logger.error(json.dumps({
                    "event": "gate_acquire_timeout", "key": self._cfg.key,
                    "limit": self._cfg.limit, "attempts": attempts,
                }))
                raise TimeoutError("global in-flight gate saturated")
            await asyncio.sleep(self._cfg.poll_backoff_s)
        logger.info(json.dumps({
            "event": "gate_acquired", "token": token[:8],
            "attempts": attempts, "limit": self._cfg.limit,
        }))
        try:
            yield token
        finally:
            # Release explicitly; the lease TTL is only a crash backstop, not the normal path.
            await self._redis.zrem(self._cfg.key, token)
            logger.info(json.dumps({"event": "gate_released", "token": token[:8]}))


class DelawarePollWorker:
    """One fleet worker; N of these share a single global cap through the gate."""

    def __init__(self, worker_id: str, gate: DistributedConcurrencyGate, http: httpx.AsyncClient) -> None:
        self._id = worker_id
        self._gate = gate
        self._http = http

    async def poll(self, entity_id: str) -> dict[str, object]:
        url = f"https://icis.corp.delaware.gov/api/entities/{entity_id}/franchise-status"
        async with self._gate.hold():
            # Only inside the gate does this request count against the global cap — the
            # difference between a clean March 1 sweep and a throttled origin (8 Del. C. s.502).
            resp = await self._http.get(url, timeout=httpx.Timeout(20.0))
        if resp.status_code == 429:
            logger.warning(json.dumps({
                "event": "de_throttled", "worker": self._id, "entity_id": entity_id,
            }))
            return {"entity_id": entity_id, "status": "rate_limited"}
        resp.raise_for_status()
        return {"entity_id": entity_id, "status": "ok", "body_len": len(resp.content)}


async def run_fleet(entity_ids: list[str], worker_count: int = 12) -> None:
    pool = redis.from_url("redis://localhost:6379/0")
    gate = DistributedConcurrencyGate(pool, GateConfig(limit=8))  # 12 workers, only 8 permits
    queue: asyncio.Queue[str] = asyncio.Queue()
    for eid in entity_ids:
        queue.put_nowait(eid)

    async with httpx.AsyncClient() as http:
        async def worker(worker_id: str) -> None:
            w = DelawarePollWorker(worker_id, gate, http)
            while True:
                try:
                    entity_id = queue.get_nowait()
                except asyncio.QueueEmpty:
                    return
                try:
                    logger.info(json.dumps({"event": "de_result", **await w.poll(entity_id)}))
                except TimeoutError:
                    # Gate saturated past the acquire budget: defer this entity and surrender this
                    # worker slot rather than force the fleet over the cap during the surge.
                    await queue.put(entity_id)
                    return

        await asyncio.gather(*(worker(f"w{i}") for i in range(worker_count)))

    await pool.aclose()

Two properties make this safe under contention. Reaping expired leases inside the same atomic call as the count-and-add means a worker that dies holding a permit cannot deadlock the fleet — its permit ages out after lease_ms and the next acquire reclaims it. And scoring each permit by acquisition time gives the sorted set a natural basis for a fair, roughly FIFO grant order if you later extend the script to enforce it. A leaky-bucket variant — draining permits at a fixed rate instead of on explicit release — is a small change to the same script when the constraint is throughput-shaped rather than connection-shaped, but for a portal that caps simultaneous connections the release-on-completion semaphore models the real limit directly.

Configuration Reference #

The gate’s parameters are portal-connection facts and fleet-topology decisions, so each is externalised with a justification rather than hard-coded.

Failure Modes and Fallback Routing #

Each fault maps onto the parent area’s Error Categorization & Retry Logic taxonomy — transient, statutory, data-validation, and system — and the fleet responds to each without ever exceeding the cap.

1. A worker crashes while holding a permit (system). Without lease expiry the permit is leaked forever and the fleet’s effective cap shrinks with every crash until it deadlocks. The ZREMRANGEBYSCORE reap on each acquire reclaims any permit older than lease_ms, so a dead holder self-heals within one lease window and the gate recovers its full capacity automatically.
2. Redis itself becomes unreachable (system, coordination loss). The gate is now blind and must fail closed, not open: a worker that cannot confirm a free permit must not send the request, because assuming a permit under partition is exactly how the fleet overshoots and gets the origin blocked. Treat a Redis error as an acquire failure, back the worker off, and let the deadline layer decide whether to escalate the affected entities.
3. The gate stays saturated through the whole surge (statutory, deadline-threatening). During the final hours before March 1 the fleet may never drain the queue at limit = 8. Workers hitting acquire_timeout_s defer their entity and surrender their slot rather than forcing the cap; the deferred, deadline-critical obligations are handed to a low-concurrency priority channel so the § 502 clock is respected instead of trading a statutory miss for raw throughput.
4. Delaware returns a 429 despite the cap being honoured (transient). The portal’s real ceiling is below your configured limit, or another origin shares your egress. Lower limit, confirm egress is partitioned per jurisdiction so no other sweep contends for the same connection budget, and let the exponential backoff engine pace the individual retry — concurrency control and per-request backoff are complementary, not substitutes.

Frequently Asked Questions #

Related #

• Async Polling & Rate Limiting
• Implementing Exponential Backoff for Secretary of State APIs
• Error Categorization & Retry Logic
• State Filing Deadline Calendars
• Secretary of State Portal & API Ingestion