Secretary Of State Portal Api Ingestion

Bulk Good-Standing Certificate Polling and Status Diffing

This guide is part of the Good-Standing Certificate Automation area within the Secretary of State Portal & API Ingestion framework. It builds the concrete engine at the centre of that pipeline: the scheduled sweep that reads good-standing status for an entire portfolio in one pass and computes a precise, typed diff of each read against the last-known state, so a good → delinquent → not-in-good-standing transition surfaces as a first-class event rather than a value buried in a table.

Scope of This Page

This page covers two tightly coupled mechanics: the bulk poll — fanning a status read across every entity under a concurrency bound without overrunning any single registry — and the status diff — joining that fresh read against a persisted snapshot to produce a typed StatusChange set classified as regression, recovery, or lateral move. It deliberately excludes what happens to those changes next: routing, deduplication, and severity ranking belong to Alerting on Good-Standing Status Changes Across a Portfolio, and the per-registry pacing that keeps the sweep polite is owned by Async Polling & Rate Limiting. Here we assume a paced client and a canonical status enum already exist, and focus only on sweeping the portfolio and computing the delta correctly.

The Constraint That Forces a Persisted Snapshot

A diff needs two operands, and only one of them — today’s read — is available at run time. The other, the last-known status, must be persisted between runs, and its integrity is a compliance concern rather than a caching convenience. Good standing is a statutory status: a California entity that lapses under Cal. Corp. Code § 2205 loses corporate powers on a specific date, and a Delaware corporation that misses its March 1 obligation under 8 Del. C. § 502 ceases to be in good standing on a specific date. The whole point of diffing is to pin when that transition happened relative to your observation window, which is only possible if the prior state was durably recorded with its own timestamp. A snapshot that is lost, stale, or silently reset does not merely lose data — it manufactures a false transition on the next run, either fabricating a regression that was already known or masking one that is real. The engine therefore treats the snapshot store as an append-only status time series, never a mutable single row, so every observation is retained and the diff is always against a dated prior.

Prerequisites

  • Python 3.10+ — for match on the canonical status, X | Y unions, and asyncio.TaskGroup-style structured concurrency (or gather on 3.10).
  • httpx 0.27+ — async client with pooling, injected already-paced from the parent area so this engine never sets its own rate policy.
  • A persistence layer — any store that supports an append-only status series keyed by (entity_id, jurisdiction); the example uses a narrow protocol so SQLAlchemy, Postgres, or a document store all satisfy it.
  • A canonical GoodStandingStatus — the ordered enum defined by the parent area, with three real states plus an UNKNOWN fail-safe.
  • Standard library — asyncio, dataclasses, enum, json, logging, datetime for the diff, the change set, and structured logging.

Implementation: A Bulk Poller with Snapshot Diffing

The engine loads the last-known snapshot for the portfolio, fans a status poll across every entity under a semaphore so no registry is hammered, then joins the fresh reads against the snapshot to emit a typed StatusChange set. An UNKNOWN read never overwrites a known baseline, so a flaky portal produces a retry signal rather than a phantom regression; only entities whose status genuinely moved appear in the change set, and the fresh reads are persisted as the next run’s baseline. Comments mark the compliance-critical lines.

The diff pipeline: join a persisted snapshot against a fresh bulk poll and emit only genuine, typed transitions A persisted previous snapshot and a fresh async bulk poll of the same portfolio feed a diff engine that joins them per entity and emits a typed StatusChange set — regression, recovery, or a dropped lateral/unchanged result — while UNKNOWN reads preserve the baseline and fresh reads become the next run's snapshot. Two operands in, a typed change set out: only genuine transitions are emitted snapshot_prev persisted last-known status per entity append-only, dated fresh_poll async bulk sweep semaphore-bounded one read per entity Diff engine join on entity_id compare enum order UNKNOWN → hold baseline no phantom transitions REGRESSION current < previous good → delinquent → not-in-GS RECOVERY current > previous reinstated / cured UNCHANGED current == previous dropped, not emitted fresh reads persisted → next run's snapshot_prev 
from __future__ import annotations

import asyncio
import enum
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Awaitable, Callable, Protocol, Sequence

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("good_standing.bulk_diff")


class GoodStandingStatus(enum.IntEnum):
    NOT_IN_GOOD_STANDING = 0   # ordered worst-to-best: a lower value is a worse status
    DELINQUENT = 1
    GOOD_STANDING = 2
    UNKNOWN = 3                 # read failed/ambiguous — special-cased, never a comparison


class ChangeKind(str, enum.Enum):
    REGRESSION = "regression"  # current worse than previous — the compliance signal
    RECOVERY = "recovery"      # current better than previous — cured / reinstated
    LATERAL = "lateral"        # same severity rank, different label (rare)


@dataclass(frozen=True)
class Observation:
    entity_id: str
    jurisdiction: str
    status: GoodStandingStatus
    observed_at: datetime


@dataclass(frozen=True)
class StatusChange:
    entity_id: str
    jurisdiction: str
    previous: GoodStandingStatus
    current: GoodStandingStatus
    kind: ChangeKind
    observed_at: datetime


class SnapshotStore(Protocol):
    """Append-only status series. load() returns the latest per entity; persist() appends."""
    def load(self, entity_ids: Sequence[str]) -> dict[str, GoodStandingStatus]: ...
    def persist(self, observations: Sequence[Observation]) -> None: ...


# A poll function is injected already-paced by Async Polling & Rate Limiting.
PollFn = Callable[[str, str], Awaitable[GoodStandingStatus]]


class BulkGoodStandingDiffer:
    """Sweep a portfolio's good-standing status and diff it against a persisted snapshot."""

    def __init__(self, store: SnapshotStore, poll: PollFn, max_concurrency: int = 8) -> None:
        self._store = store
        self._poll = poll
        self._sem = asyncio.Semaphore(max_concurrency)  # bound so no registry is flooded

    async def _poll_one(self, entity_id: str, jurisdiction: str) -> Observation:
        async with self._sem:  # concurrency gate keeps the sweep within pacing budgets
            try:
                status = await self._poll(entity_id, jurisdiction)
            except Exception as exc:  # a transport failure is UNKNOWN, not a status change
                logger.warning(json.dumps({
                    "event": "poll_failed", "entity_id": entity_id,
                    "jurisdiction": jurisdiction, "error": type(exc).__name__,
                }))
                status = GoodStandingStatus.UNKNOWN
        return Observation(entity_id, jurisdiction, status, datetime.now(timezone.utc))

    async def poll_all(self, entities: Sequence[tuple[str, str]]) -> list[Observation]:
        results = await asyncio.gather(
            *(self._poll_one(eid, juris) for eid, juris in entities)
        )
        logger.info(json.dumps({"event": "sweep_complete", "polled": len(results)}))
        return list(results)

    def diff(
        self, previous: dict[str, GoodStandingStatus], fresh: Sequence[Observation]
    ) -> list[StatusChange]:
        changes: list[StatusChange] = []
        for obs in fresh:
            # A failed read must never clobber a known baseline — hold and retry next run.
            if obs.status is GoodStandingStatus.UNKNOWN:
                continue
            prior = previous.get(obs.entity_id)
            if prior is None or prior is GoodStandingStatus.UNKNOWN:
                continue  # first real observation establishes a baseline, emits nothing
            if prior == obs.status:
                continue  # unchanged — dropped, the common case on a healthy portfolio
            kind = (ChangeKind.REGRESSION if obs.status < prior
                    else ChangeKind.RECOVERY)
            change = StatusChange(obs.entity_id, obs.jurisdiction, prior,
                                  obs.status, kind, obs.observed_at)
            changes.append(change)
            logger.info(json.dumps({
                "event": "status_change", "entity_id": obs.entity_id,
                "jurisdiction": obs.jurisdiction, "previous": prior.name,
                "current": obs.status.name, "kind": kind.value,
            }))
        return changes

    async def run(self, entities: Sequence[tuple[str, str]]) -> list[StatusChange]:
        """One scheduled sweep: load baseline, poll, diff, persist real reads, return changes."""
        previous = self._store.load([eid for eid, _ in entities])
        fresh = await self.poll_all(entities)
        changes = self.diff(previous, fresh)
        # Persist only successful reads so an UNKNOWN cannot overwrite a good baseline.
        real = [o for o in fresh if o.status is not GoodStandingStatus.UNKNOWN]
        self._store.persist(real)
        logger.info(json.dumps({
            "event": "run_complete", "entities": len(entities),
            "changes": len(changes),
            "regressions": sum(c.kind is ChangeKind.REGRESSION for c in changes),
        }))
        return changes

The differ is deliberately a pure transformation over its two operands: given a baseline and a set of observations, its output is fully determined, which makes it replayable during an audit and trivially unit-testable. The only stateful actor is the snapshot store, and the engine touches it exactly twice per run — a load at the start and a persist of successful reads at the end — so the window in which the baseline can drift is bounded to a single sweep.

Configuration Reference

These parameters govern the sweep and diff behaviour and are configuration data, not branches in the loop, because they are dictated by portfolio size and registry tolerance rather than by code.

Parameter Suggested value Justification
max_concurrency 8 Bounds simultaneous registry hits; higher risks a 429 from a single portal, lower slows the sweep. Tune per portal, not globally.
Snapshot model append-only series A mutable single row loses transition history and cannot date a regression; the diff needs a dated prior.
UNKNOWN persistence never persisted A failed read persisted as baseline would mask the real prior status and fabricate a transition on the next successful read.
First-observation policy emit nothing With no dated prior there is no transition to report; the first real read only establishes the baseline.
Poll injection paced client from parent area The differ must not set its own rate policy; pacing belongs to Async Polling & Rate Limiting so limits stay per-registry.
Diff purity no I/O in diff() Keeps the delta replayable from a persisted baseline and observation set for audit reproduction.

Failure Modes and Fallback Routing

Each fault maps onto the four-tier scheme in the parent framework’s error categorization & retry logic — transient, statutory, data-validation, and system — and the engine responds to each differently.

  1. A registry times out mid-sweep (transient). _poll_one catches the exception, records the read as UNKNOWN, and the differ skips it, so the entity’s known baseline is preserved rather than dropped. The failure is logged for the retry layer to re-poll on the next window; it never enters the change set as a false regression.
  2. The snapshot store returns an empty or stale baseline (system). With no prior, diff emits nothing for that entity — correct, because there is no transition to report — but a portfolio-wide empty baseline is itself the alarm: if previous is empty for entities you know were observed, the store lost data and the sweep must halt rather than re-baseline silently on possibly-degraded reads.
  3. A registry changes its vocabulary so reads normalize to UNKNOWN en masse (data-validation). Because UNKNOWN is never persisted and never diffed, a vocabulary break produces a spike of held reads and poll-failure logs, not a wave of phantom regressions. The spike is the signal to fix the upstream label map before the baseline goes stale.
  4. Concurrency set too high trips a single portal’s limiter (transient/statutory-adjacent). A flooded registry starts returning 429; the paced client should absorb this, but if max_concurrency outpaces the per-registry budget the sweep degrades to UNKNOWN reads. Lower concurrency for that jurisdiction rather than globally, and route the throttled calls back through the parent area’s async polling & rate limiting budget.

Frequently Asked Questions

Why persist a full status series instead of just overwriting the current status per entity?

Because the diff has to date the transition, and a single mutable row cannot. Good standing is lost on a statutory date — a California suspension under Corp. Code § 2205, a Delaware default under § 502 — and the value of the diff is pinning that change to an observation window. An append-only series retains every dated prior, so a regression can be reported with a defensible “last seen good on” timestamp; overwriting throws that away and leaves you unable to prove when the entity lapsed.

What stops an intermittent portal failure from firing a false regression?

A failed or ambiguous read normalizes to UNKNOWN, and the differ both skips UNKNOWN when comparing and refuses to persist it as a baseline. So a portal that flaps produces a poll-failure log and a held baseline, never a StatusChange. Only a successful read that genuinely differs from the last successful read becomes a regression or recovery, which keeps transient transport noise out of the change set entirely.

How is the sweep kept from overrunning a single state's rate limit?

Two mechanisms. The max_concurrency semaphore bounds how many reads are in flight at once, and the poll function itself is injected already-paced by Async Polling & Rate Limiting, which applies a per-registry token budget. The differ never sets its own rate policy; it only bounds fan-out. If one portal is more fragile, lower concurrency for that jurisdiction rather than throttling the whole sweep.

Why does the first observation of an entity emit no change?

Because a diff needs two operands and the first read supplies only one. With no dated prior in the snapshot there is no transition to report, so the first successful read simply establishes the baseline and emits nothing. This is deliberate: treating an initial observation as a change would flood the change set every time a new entity is onboarded, drowning the real regressions the engine exists to surface.