Building Graceful Degradation for Offline Fare Readers

The task this page solves is narrow and operational: keep an on-board or gate validator charging correct, auditable fares during the seconds and minutes when it cannot reach the central clearinghouse. When cellular backhaul fails, edge validators lose rule synchronization, or a vehicle enters an RF-dead tunnel, the reader has three choices — halt boarding, wave everyone through for free, or execute a deterministic offline protocol. For transit operations managers, revenue analysts, and the Python developers who ship validator firmware, only the third is defensible. This is the degraded-mode path of the broader Fallback Calculation Chains inside the Fare Rule Validation & Calculation Engines pipeline. The goal is never to replicate the full clearinghouse on a bus — it is to hold fare integrity through localized rule evaluation, bounded risk, and reconciliation that provably settles once the link returns.

Localized Validation Architecture

At the foundation of a resilient reader is a lightweight, state-aware validator that runs with no assumption of network availability. It caches fare matrices, zone boundaries, concession parameters, and product entitlements under strict cryptographic integrity — the same signed-manifest discipline described in AFC System Security Boundaries. When connectivity drops, the reader transitions from synchronous backend validation to an asynchronous, store-and-forward model that follows the offline caching approach in Fallback Routing Strategies. The edge Python runtime keeps a minimal dependency footprint: SQLite for persistent state, Pydantic for schema validation, and a deterministic evaluation loop that holds sub-150 ms tap-to-acknowledge latency.

The state diagram below captures the validator’s transition between online validation and offline store-and-forward, including the post-sync reconciliation step:

The Offline Fallback Chain

Once offline, transition logic prioritizes deterministic outcomes over real-time optimization. In production this is a layered evaluation pipeline, each tap descending through it until a tier resolves or the tap routes to the dead-letter queue:

1. Product cache validation — verify the tapped credential against a locally stored, cryptographically signed product registry.
2. Temporal and spatial resolution — apply cached zone boundaries and time-of-day multipliers.
3. Conservative defaulting — if routing complexity cannot be resolved offline, cap at a flat fare or the highest applicable tier for the tapped product.
4. Telemetry emission — record fallback depth, applied rule, and a confidence score for post-sync reconciliation.

The layered pipeline below shows how each tap descends through the offline chain, incrementing fallback depth until it resolves or routes to the dead-letter queue:

Conservative defaulting is where money is decided, so it must use exact arithmetic. Given a cached base fare $b$ (in whole cents), a cached zone or peak multiplier $m$, and a product cap $c_{\max}$, the offline fare is:

$$f_{\text{final}} = \min\bigl(\operatorname{round\_half\_up}(b \cdot m),\ c_{\max}\bigr)$$

The multiplier is applied with Decimal, never float, and rounded half-up to whole cents before the cap is enforced — a binary-float 1.25 would drift a rounded fare by a cent on high-volume routes and desynchronize reconciliation. Each layer is idempotent and explicitly versioned, and the whole chain is wrapped so that schema drift, storage exhaustion, or a failed integrity check routes to a local dead-letter queue instead of halting the validator.

Step-by-Step: A Hardened Offline Validator

The script below is a runnable offline validator with explicit type hints, Decimal fare math, structured audit trails, and deterministic fallback routing. It leans on the standard library plus Pydantic for schema enforcement. For deployment, consult the SQLite documentation for WAL-mode tuning on embedded storage.

import sqlite3
import logging
from decimal import Decimal, ROUND_HALF_UP
from datetime import datetime, timezone
from typing import Optional, Dict, Any, List, Tuple
from enum import Enum
from pydantic import BaseModel, Field, ValidationError

# Structured audit logging configuration
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    handlers=[logging.StreamHandler()],
)
AUDIT_LOGGER = logging.getLogger("transit.revenue_audit")

BASE_MULTIPLIER = Decimal("1.00")
PEAK_MULTIPLIER = Decimal("1.25")


class TapStatus(str, Enum):
    APPROVED = "APPROVED"
    FALLBACK_FLAT = "FALLBACK_FLAT"
    REJECTED = "REJECTED"
    DEAD_LETTER = "DEAD_LETTER"


class FareTap(BaseModel):
    card_id: str = Field(..., pattern=r"^[A-Z0-9]{12}$")
    tap_timestamp: datetime
    vehicle_id: str
    zone_id: Optional[str] = None
    route_id: Optional[str] = None


class ValidationResult(BaseModel):
    tap_id: str
    status: TapStatus
    fare_amount_cents: int
    fallback_depth: int = 0
    confidence_score: float = 1.0
    applied_rule: str
    processed_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


class OfflineValidator:
    """Deterministic offline fare validator with explicit fallback routing and audit trails."""

    def __init__(self, db_path: str = "offline_fare_state.db") -> None:
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.conn.execute("PRAGMA journal_mode=WAL;")
        self.conn.execute("PRAGMA synchronous=NORMAL;")
        self._init_schema()

    def _init_schema(self) -> None:
        self.conn.executescript("""
            CREATE TABLE IF NOT EXISTS fare_cache (
                product_id TEXT PRIMARY KEY,
                base_fare_cents INTEGER NOT NULL,
                max_cap_cents INTEGER NOT NULL,
                valid_from TEXT,
                valid_to TEXT
            );
            CREATE TABLE IF NOT EXISTS dead_letter_queue (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                payload TEXT NOT NULL,
                error_trace TEXT NOT NULL,
                created_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS audit_trail (
                tap_id TEXT PRIMARY KEY,
                status TEXT NOT NULL,
                fare_amount_cents INTEGER NOT NULL,
                fallback_depth INTEGER NOT NULL,
                rule_applied TEXT NOT NULL,
                processed_at TEXT NOT NULL
            );
        """)
        self.conn.commit()

    def _evaluate_fallback_chain(self, tap: FareTap) -> ValidationResult:
        depth = 0
        try:
            # Layer 1: product cache validation
            cursor = self.conn.execute(
                "SELECT base_fare_cents, max_cap_cents FROM fare_cache WHERE product_id = ?",
                (tap.card_id,),
            )
            row: Optional[Tuple[int, int]] = cursor.fetchone()
            if not row:
                raise ValueError("Product not in local cache")
            base_fare, max_cap = row
            depth += 1

            # Layer 2: temporal & spatial resolution
            multiplier = BASE_MULTIPLIER
            if tap.zone_id and tap.zone_id.startswith("Z_"):
                multiplier = PEAK_MULTIPLIER  # cached peak / zone uplift
            depth += 1

            # Layer 3: conservative defaulting — exact Decimal math, then cap
            calculated = int(
                (Decimal(base_fare) * multiplier).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
            )
            final_fare = min(calculated, max_cap)

            return ValidationResult(
                tap_id=tap.card_id,
                status=TapStatus.FALLBACK_FLAT,
                fare_amount_cents=final_fare,
                fallback_depth=depth,
                confidence_score=0.85 if multiplier != BASE_MULTIPLIER else 0.98,
                applied_rule="offline_zone_default",
            )
        except Exception as exc:  # deliberate: any failure routes to DLQ, never halts the reader
            depth += 1
            AUDIT_LOGGER.error("Fallback chain failed at depth %d: %s", depth, exc)
            self._push_to_dlq(tap.model_dump_json(), str(exc))
            return ValidationResult(
                tap_id=tap.card_id,
                status=TapStatus.DEAD_LETTER,
                fare_amount_cents=0,
                fallback_depth=depth,
                confidence_score=0.0,
                applied_rule="dlq_bypass",
            )

    def _push_to_dlq(self, payload: str, error_trace: str) -> None:
        self.conn.execute(
            "INSERT INTO dead_letter_queue (payload, error_trace, created_at) VALUES (?, ?, ?)",
            (payload, error_trace, datetime.now(timezone.utc).isoformat()),
        )
        self.conn.commit()

    def process_tap(self, tap: FareTap) -> ValidationResult:
        try:
            result = self._evaluate_fallback_chain(tap)
            self._persist_audit(result)
            return result
        except ValidationError as ve:
            AUDIT_LOGGER.critical("Schema drift detected during tap processing: %s", ve)
            raise
        except sqlite3.Error as sqle:
            AUDIT_LOGGER.critical("Storage exhaustion or corruption: %s", sqle)
            raise RuntimeError("Local state corrupted. Halting validator.") from sqle

    def _persist_audit(self, result: ValidationResult) -> None:
        self.conn.execute(
            "INSERT OR REPLACE INTO audit_trail VALUES (?, ?, ?, ?, ?, ?)",
            (
                result.tap_id,
                result.status.value,
                result.fare_amount_cents,
                result.fallback_depth,
                result.applied_rule,
                result.processed_at.isoformat(),
            ),
        )
        self.conn.commit()
        AUDIT_LOGGER.info(
            "AUDIT | %s | %s | %dc | depth:%d",
            result.tap_id, result.status.value, result.fare_amount_cents, result.fallback_depth,
        )

    def flush_reconciliation_queue(self) -> List[Dict[str, Any]]:
        """Extract DLQ payloads for post-sync clearinghouse reconciliation."""
        cursor = self.conn.execute(
            "SELECT id, payload, error_trace FROM dead_letter_queue ORDER BY id ASC"
        )
        rows = cursor.fetchall()
        if not rows:
            return []
        self.conn.execute("DELETE FROM dead_letter_queue")
        self.conn.commit()
        return [{"id": r[0], "payload": r[1], "error": r[2]} for r in rows]

Layer 1 fails closed: an unknown or unsigned product raises before any fare is computed, so an attacker cannot force a cheap flat fare by presenting a credential the reader has never cached. Layers 2 and 3 always produce a number, and every path — success or dead-letter — writes exactly one audit row, which is what makes the offline ledger reconcilable later.

Validation & Test Cases

Exercise the validator against a seeded cache. The normal case is a cached product on a peak zone; the edge case is an uncached credential that must land in the dead-letter queue without raising:

validator = OfflineValidator(":memory:")
validator.conn.execute(
    "INSERT OR REPLACE INTO fare_cache VALUES (?, ?, ?, ?, ?)",
    ("CARD001ABC", 250, 500, "2024-01-01", "2027-12-31"),
)
validator.conn.commit()

# Normal case: cached product, peak zone -> 250c * 1.25 = 313c (half-up), under the 500c cap
approved = validator.process_tap(
    FareTap(card_id="CARD001ABC", tap_timestamp=datetime.now(timezone.utc),
            vehicle_id="BUS-402", zone_id="Z_PEAK")
)
assert approved.status is TapStatus.FALLBACK_FLAT
assert approved.fare_amount_cents == 313
assert approved.fallback_depth == 2

# Edge case: uncached credential -> dead-letter, zero fare, no exception escapes
missing = validator.process_tap(
    FareTap(card_id="UNKNOWN00000", tap_timestamp=datetime.now(timezone.utc),
            vehicle_id="BUS-402")
)
assert missing.status is TapStatus.DEAD_LETTER
assert missing.fare_amount_cents == 0
assert len(validator.flush_reconciliation_queue()) == 1

The peak fare is 313, not 312: Decimal("250") * Decimal("1.25") is exactly 312.50, which rounds half-up to 313. A float path (int(250 * 1.25) = 312) truncates and would leak a cent per peak tap — invisible per rider, material across a fleet at settlement. The dead-letter assertion proves the guarantee that matters operationally: a credential the reader has never seen never blocks boarding and never charges a guessed amount; it is captured for the clearinghouse to price after sync.

Handling Transfer and Concession Dependencies

Offline readers struggle most with temporal dependencies such as transfer windows, which normally need cross-vehicle or cross-operator state. Locally, the validator keeps a rolling hash table of recent taps keyed by anonymized card identifier; a second tap inside the cached window applies a zero-fare transfer rule consistent with the operator’s Transfer Window Logic. If the window expires or the table exceeds its memory bound, the reader falls back to a conservative base fare and flags the event for backend reconciliation rather than guessing a discount it cannot verify offline.

Every offline decision is timestamped with a UTC monotonic clock and signed with a device-specific HMAC, so on reconnect the clearinghouse can verify the offline ledger’s integrity against the central tariff schedule. Structured logging through Python’s logging module keeps every fallback depth and confidence score traceable for revenue assurance; see the logging documentation for rotating file handlers on embedded Linux validators.

Post-Sync Reconciliation & Debugging

When deploying offline readers, work this diagnostic order to isolate degradation bottlenecks:

1. Verify WAL integrity — run PRAGMA integrity_check; after unexpected power cycles. Corruption in audit_trail points to improper COMMIT sequencing during fallback execution.
2. Monitor DLQ backlog — query dead_letter_queue size hourly. Sustained growth above ~5% of taps signals expired cache certificates or schema drift between edge and central models; replaying that queue is the same discipline covered in Schema Validation Pipelines.
3. Simulate RF dead zones — use tc qdisc to inject 100% packet loss for 300-second intervals and confirm tap-to-acknowledge stays under 150 ms while fallback_depth increments deterministically.
4. Check reconciliation drift — after backhaul restoration, compare SUM(fare_amount_cents) from the offline audit_trail against the clearinghouse’s expected yield. A discrepancy above 0.5% warrants a tariff override review and cache invalidation.
5. Validate the cache cryptographically — populate fare_cache only from signed manifests, and reject unsigned or expired tariff payloads before entering fallback mode to close the offline fare-evasion path.

Integration Note

This task is one branch of the parent Fallback Calculation Chains component: the offline reader is the most degraded tier of the same chain that, online, cascades through prioritized rule subsets and tolerance thresholds. Once the link returns, the flush_reconciliation_queue output feeds the online engine, where any dead-lettered tap is re-priced with full rule context — including peak thresholds tuned by the sibling task Dynamic Peak Pricing Threshold Adjustment Scripts, so an estimate captured offline settles against the same multiplier the central engine would have applied live.

FAQ

Related

• Fallback Calculation Chains — the parent component this offline path is the degraded tier of
• Fallback Routing Strategies — how taps are cached and routed locally when validators lose the link
• AFC System Security Boundaries — the signed-manifest and HMAC discipline the offline cache depends on
• Calculating Cross-Operator Transfer Windows with Python — the transfer rule readers approximate offline
• Dynamic Peak Pricing Threshold Adjustment Scripts — the peak multipliers the clearinghouse re-prices against after sync

Part of Fallback Calculation Chains, within Fare Rule Validation & Calculation Engines.