Error Categorization & Logging for Grant Pipelines

This guide is part of the Data Ingestion & Grant Parsing Workflows reference. It defines the discrete control plane that intercepts, classifies, and routes pipeline anomalies — the subsystem every other ingestion stage hands its failures to.

The scope is strictly confined to error classification, structured audit logging, and deterministic fallback routing. This subsystem operates as a read-only enforcement layer: it validates structural integrity, enforces semantic boundaries, and guarantees immutable auditability, but it never mutates an upstream payload. It does not perform document extraction, field normalization, financial aggregation, or rule adjudication. Binary extraction belongs to PDF Grant Application Parsing; canonical field translation belongs to Field Mapping & Normalization; spreadsheet reconciliation belongs to Excel Budget Template Sync; concurrent dispatch belongs to Async Batch Processing Pipelines. This stage assumes a parsed dictionary or a raised exception on the way in and emits exactly two things on the way out: an append-only audit record and a routing directive.

Nonprofit operations teams, grant program managers, Python automation engineers, and compliance officers should treat this stage as a deterministic routing switch, not a corrective or transformational engine. No heuristic correction, auto-repair, or inline mutation is permitted at this boundary — repair, if any, happens in an isolated downstream consumer that reads from the quarantine queue.

Prerequisites #

This control plane targets a current, supported runtime and a small set of pinned, audit-stable dependencies. Version pinning is itself a compliance control: an unpinned validation library can silently change rejection behavior between deployments and break reproducibility of the audit trail.

• Python: 3.11 or newer (the examples use enum.StrEnum and tomllib-era typing semantics).
• Pinned packages (requirements.txt):
  • pydantic==2.7.1 — frozen schema validation at the subsystem boundary.
  • structlog==24.1.0 — JSON-only structured log rendering.
  • pytest==8.2.0 — verification harness.
  • hypothesis==6.100.1 — property-based payload fuzzing.
• Environment variables:
  • GRANT_ERROR_WAL_PATH — absolute path to the append-only write-ahead log (default /var/log/grant_pipelines/error_wal.jsonl); the directory must be writable by the pipeline service account and backed by retained storage.
  • GRANT_SCHEMA_VERSION — the active validation_rule_id (for example v2.1.0) bound to every emitted record.
  • GRANT_MAX_TRANSIENT_RETRIES — integer cap on transient retries before forced quarantine (default 3).
• Upstream stage dependencies: a parsed record dictionary from the ingestion stages above, plus the canonical schema module published by Field Mapping & Normalization. Transient-error retry policy is shared with Pipeline Fallback & Retry Logic; this stage classifies and defers to that policy rather than reimplementing backoff timing.

Error Taxonomy & Deterministic Routing Matrix #

All pipeline exceptions must conform to a strict, enumerated taxonomy. Categories are defined by failure origin, compliance impact, and required routing behavior. The taxonomy eliminates ambiguity and ensures consistent handling across distributed batch workers.

Routing decisions are deterministic and version-controlled. INFRASTRUCTURE_TRANSIENT errors trigger isolated retry queues with capped exponential backoff. STRUCTURAL_PARSE and SEMANTIC_VALIDATION failures immediately quarantine the record and emit a structured alert to the operations dashboard. COMPLIANCE_RULE violations halt downstream propagation and require explicit compliance officer sign-off before any further processing. Rate-limit exhaustion arriving from API Polling & Rate Limiting is the canonical INFRASTRUCTURE_TRANSIENT source and is the only category permitted to re-enter the pipeline automatically.

Core Implementation #

Implementation relies on canonical Python tooling to guarantee deterministic output and machine-readable auditability. The standard logging module is wrapped with structlog to enforce JSON-formatted log lines, and validation runs against a frozen pydantic schema snapshot. Deterministic record hashing (hashlib.sha256 over the canonicalized payload) ensures traceability across retries and prevents duplicate alerting. The logging pipeline operates independently of business logic, writing to a write-ahead log (WAL) before any fallback routing executes. This guarantees that even a catastrophic process failure preserves a complete, queryable audit trail for regulatory review.

import hashlib
import json
import logging
from enum import Enum
from pathlib import Path
from typing import Any

import structlog
from pydantic import ValidationError

logger = logging.getLogger("grant.error_control_plane")


# ---------------------------------------------------------------------------
# Canonical Taxonomy & Routing
# ---------------------------------------------------------------------------
class ErrorCategory(str, Enum):
    STRUCTURAL_PARSE = "STRUCTURAL_PARSE"
    SEMANTIC_VALIDATION = "SEMANTIC_VALIDATION"
    COMPLIANCE_RULE = "COMPLIANCE_RULE"
    INFRASTRUCTURE_TRANSIENT = "INFRASTRUCTURE_TRANSIENT"


class RoutingAction(str, Enum):
    QUARANTINE = "QUARANTINE"
    RETRY_BACKOFF = "RETRY_BACKOFF"
    COMPLIANCE_HOLD = "COMPLIANCE_HOLD"


_CATEGORY_TO_ACTION: dict[ErrorCategory, RoutingAction] = {
    ErrorCategory.INFRASTRUCTURE_TRANSIENT: RoutingAction.RETRY_BACKOFF,
    ErrorCategory.COMPLIANCE_RULE: RoutingAction.COMPLIANCE_HOLD,
    ErrorCategory.STRUCTURAL_PARSE: RoutingAction.QUARANTINE,
    ErrorCategory.SEMANTIC_VALIDATION: RoutingAction.QUARANTINE,
}


# ---------------------------------------------------------------------------
# Structlog Configuration (JSON-only, ISO timestamps, strict schema)
# ---------------------------------------------------------------------------
structlog.configure(
    processors=[
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(sort_keys=True),
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    logger_factory=structlog.stdlib.LoggerFactory(),
)


# ---------------------------------------------------------------------------
# WAL Writer (decoupled from routing logic)
# ---------------------------------------------------------------------------
class AuditWAL:
    """Append-only write-ahead log. Persists before any routing occurs."""

    def __init__(self, wal_path: str) -> None:
        self.path = Path(wal_path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._log = structlog.get_logger("audit_wal")

    def write(self, entry: dict[str, Any]) -> None:
        with open(self.path, "a", encoding="utf-8") as fh:
            fh.write(json.dumps(entry, sort_keys=True) + "\n")
            fh.flush()
        self._log.info("wal_entry_persisted", record_hash=entry["record_hash"])


# ---------------------------------------------------------------------------
# Deterministic Classification + Routing — NO MUTATION OF THE PAYLOAD
# ---------------------------------------------------------------------------
def classify_and_route(
    payload: dict[str, Any],
    schema_version: str,
    wal: AuditWAL,
) -> RoutingAction:
    """Validate a payload against the frozen schema, log deterministically,
    and return a routing action. The input payload is never modified."""
    record_hash = hashlib.sha256(
        json.dumps(payload, sort_keys=True).encode("utf-8")
    ).hexdigest()

    bound = structlog.get_logger().bind(
        pipeline_stage="error_categorization",
        record_hash=record_hash,
        validation_rule_id=schema_version,
    )

    try:
        # GrantRecord is the versioned contract owned by the normalization stage.
        from validation_schemas import GrantRecord

        GrantRecord.model_validate(payload)
    except ValidationError as exc:
        category = _map_validation_error_to_category(exc)
    except (TimeoutError, ConnectionError) as exc:
        category = ErrorCategory.INFRASTRUCTURE_TRANSIENT
        logger.warning("transient_failure", exc_info=exc)
    else:
        # Validation passed: nothing to route here. Returning None would be a
        # contract violation, so callers must only invoke this on a known failure.
        bound.info("validation_passed_no_route")
        return RoutingAction.QUARANTINE

    action = _CATEGORY_TO_ACTION[category]
    audit_entry: dict[str, Any] = {
        "pipeline_stage": "error_categorization",
        "error_category": category.value,
        "record_hash": record_hash,
        "validation_rule_id": schema_version,
        "fallback_action": action.value,
        "compliance_flag": category is ErrorCategory.COMPLIANCE_RULE,
    }

    wal.write(audit_entry)  # WAL write strictly precedes routing.
    bound.error("validation_boundary_breach", **audit_entry)
    return action


def _map_validation_error_to_category(exc: ValidationError) -> ErrorCategory:
    """Deterministic mapping from Pydantic v2 error types to the taxonomy."""
    for err in exc.errors():
        err_type = err.get("type", "")
        if err_type == "missing":
            return ErrorCategory.STRUCTURAL_PARSE
        if err_type.startswith("compliance_"):  # custom validator namespace
            return ErrorCategory.COMPLIANCE_RULE
    return ErrorCategory.SEMANTIC_VALIDATION

Two contracts make this implementation auditable. First, the record_hash is computed once over the canonical (sorted-key) serialization and threads through every log line and queue message, so a single grant record is correlatable across an arbitrary number of retries. Second, the WAL write happens before classify_and_route returns, so the routing directive can never exist without a matching durable audit entry. Errors are returned as explicit RoutingAction values; exceptions are mapped, logged with exc_info, and re-expressed as taxonomy categories — never swallowed silently.

Field Mapping & Schema Contract #

Every audit record conforms to a fixed, version-controlled schema. Downstream observability tools, compliance queries, and the detailed walkthrough in Implementing automated error logging for grant pipelines all depend on these exact field names. Producers occasionally emit legacy aliases; the table below is the canonical resolution and coercion contract enforced at WAL ingress.

Alias resolution is intentionally one-directional: a producer may send payload_sha256, but the WAL only ever persists record_hash. This prevents the audit trail from fragmenting into synonymous keys that an examiner would have to reconcile by hand. Records that arrive with a value outside the enumerated taxonomy or routing set are themselves classified STRUCTURAL_PARSE and quarantined, because a malformed error record is a chain-of-custody defect in its own right.

Validation & Testing #

Because this stage is the last line of defense before financial reconciliation, its behavior is pinned with pytest and fuzzed with hypothesis. Tests assert two invariants: the correct taxonomy classification for known-bad payloads, and the presence of a durable WAL entry whose record_hash matches the routed record.

import json

import pytest
from hypothesis import given, strategies as st

from error_control_plane import (
    AuditWAL,
    ErrorCategory,
    RoutingAction,
    classify_and_route,
)

SCHEMA_VERSION = "v2.1.0"


@pytest.fixture()
def wal(tmp_path) -> AuditWAL:
    return AuditWAL(str(tmp_path / "error_wal.jsonl"))


def _read_wal(wal: AuditWAL) -> list[dict]:
    return [json.loads(line) for line in wal.path.read_text().splitlines()]


def test_missing_required_field_is_structural(wal: AuditWAL) -> None:
    # 'requested_amount' omitted -> Pydantic 'missing' -> STRUCTURAL_PARSE.
    payload = {"grant_id": "G-2026-0001"}
    action = classify_and_route(payload, SCHEMA_VERSION, wal)

    assert action is RoutingAction.QUARANTINE
    entries = _read_wal(wal)
    assert len(entries) == 1
    assert entries[0]["error_category"] == ErrorCategory.STRUCTURAL_PARSE.value
    assert entries[0]["validation_rule_id"] == SCHEMA_VERSION


def test_wal_write_precedes_routing(wal: AuditWAL) -> None:
    payload = {"grant_id": "G-2026-0002", "requested_amount": "not-a-number"}
    action = classify_and_route(payload, SCHEMA_VERSION, wal)

    entry = _read_wal(wal)[0]
    # The persisted hash must match the routed record, proving the audit
    # entry exists for every routing decision.
    assert entry["fallback_action"] == action.value
    assert len(entry["record_hash"]) == 64


@given(payload=st.dictionaries(st.text(min_size=1), st.text()))
def test_every_failure_produces_exactly_one_wal_entry(
    tmp_path_factory, payload: dict
) -> None:
    wal = AuditWAL(str(tmp_path_factory.mktemp("wal") / "wal.jsonl"))
    classify_and_route(payload, SCHEMA_VERSION, wal)
    # No malformed payload may route without leaving precisely one audit record.
    assert len(_read_wal(wal)) == 1

Expected results: the first two tests pass deterministically, and the property-based test confirms that no arbitrary dictionary can pass through classify_and_route without emitting exactly one audit line. Run the suite with pytest -q; a green run is the precondition for promoting any change to this boundary, since a missing WAL entry is a reportable internal-control failure rather than a mere bug.

Performance & Scale Considerations #

Nonprofit-scale ingestion is bursty rather than high-throughput: a deadline week may push tens of thousands of records through in a few hours, then return to near-idle. The control plane is tuned for that profile.

• Batch sizing: classify in batches of 500–1,000 records. The dominant cost is json.dumps(..., sort_keys=True) for hashing, not validation, so batching amortizes Python interpreter overhead without inflating peak memory.
• WAL durability vs. throughput: the example calls flush() per record for a strict ordering guarantee. Under sustained load, group-commit (flush once per batch with os.fsync on the directory at batch close) raises throughput several-fold while preserving the write-ahead invariant at batch granularity — an acceptable trade for most grant workloads.
• Concurrency limits: keep WAL writers single-writer per shard. Multiple processes appending to one JSONL file risk interleaved partial lines; instead shard by validation_rule_id or date and merge at query time.
• Memory ceilings: never hold the full input corpus in memory. Stream records from the upstream queue, hash, route, and discard. Peak memory should stay flat regardless of corpus size — target well under 256 MB for the worker, which fits comfortably on the small instances typical of nonprofit infrastructure.
• Backpressure: when the quarantine or compliance-hold queue depth exceeds its threshold, slow ingestion rather than dropping records. A dropped error is an audit gap; a delayed one is not.

Failure Modes & Troubleshooting #

Every row above is itself observable: a sustained shift in category volume is the primary operational signal. A spike in COMPLIANCE_RULE requires immediate workflow suspension and manual triage; an INFRASTRUCTURE_TRANSIENT spike should trigger infrastructure scaling or rate-limit renegotiation rather than wider retry windows.

Compliance Alignment #

This control plane satisfies specific record-keeping and internal-control obligations rather than generic “compliance,” and every log entry is auditable evidence for internal controls and external examinations.

Compliance officers can query the WAL with standard JSON-path filters to reconstruct the exact state of any grant record at the moment of failure, and the compliance_flag boolean enables rapid filtering for regulatory submissions. Under the 2 CFR §200.334 retention requirement, the append-only WAL at GRANT_ERROR_WAL_PATH is the retention artifact and must be preserved for the federal three-year minimum (longer where a funder or state statute extends it). Audit artifacts emitted here also conform to the field conventions defined in Compliance Metadata Standards, and access to the quarantine and compliance-hold queues is governed by Data Security & Access Boundaries. The subsystem guarantees that no malformed or non-compliant record silently propagates to financial reconciliation or funder reporting stages.

Related #

• Parent: Data Ingestion & Grant Parsing Workflows
• Implementing automated error logging for grant pipelines — tactical deployment, alert routing, and log aggregation
• Field Mapping & Normalization — owns the canonical schema this stage validates against
• Async Batch Processing Pipelines — the dispatch layer whose rejections flow into this sink
• API Polling & Rate Limiting — canonical source of INFRASTRUCTURE_TRANSIENT events
• Pipeline Fallback & Retry Logic — cross-domain terminal-failure and backoff policy