Fallback Routing Logic

Modern agricultural automation runs across connectivity that is anything but reliable: cellular dead zones behind a tree line, RF interference from irrigation pivots, and multi-second satellite latency during a storm all routinely stall the primary telemetry link. When an upstream path degrades, routing logic must transition deterministically to a secondary path — and then to durable local storage — without losing a single spray event, misordering a sequence, or breaking an audit trail. This subsystem sits inside the Agricultural Automation System Architecture & Compliance framework, which mandates state-preserving failover that keeps field-level operational context intact through network loss and replays it cleanly on reconnect.

Problem Framing: Why Silent Path Loss Is Expensive

The sub-problem this subsystem owns is narrow but unforgiving: given a telemetry payload and a set of egress paths of varying quality, choose a path deterministically, prove delivery or persist locally, and record exactly what happened — every time, under partial failure. The failure mode is not a crash. A dropped connection that silently discards a payload produces no error the operator sees in the cab; the record simply never arrives, and the gap surfaces months later during an inspection.

The cost is asymmetric. A lost restricted-use pesticide application record is not a missing data point — it is a compliance gap that can invalidate a certification or a crop-insurance claim, because the operator can no longer prove the application stayed inside its label rate and re-entry interval. Worse, naive retry logic that fires a payload at two paths concurrently can produce a duplicate application record, which reads as an over-application during audit. Deterministic fallback routing exists to make both classes of error structurally impossible: exactly-once semantics on the happy path, guaranteed local durability when every network path is down, and an immutable per-payload audit log that reconstructs the full routing history of any event.

Fallback routing failover pipeline and audit log A horizontal chain of gates. A TelemetryPayload enters a schema and clock-skew gate; invalid or clock-skewed frames drop down to a QUARANTINED dead-letter. Valid payloads attempt the primary egress under a 3-second timeout; on timeout or connection failure they fall over to the secondary egress under a 5-second timeout; on further failure they persist to a fsync-durable local buffer. Buffered payloads are replayed upstream in timestamp order on reconnect. Dashed lines run from the payload, the quarantine, the primary, secondary and buffer stages down into a single append-only per-payload audit log. TelemetryPayload validated · typed Schema + skew gate reject → quarantine Primary egress 3 s timeout Secondary egress 5 s timeout Local buffer fsync-durable QUARANTINED dead-letter Append-only per-payload audit log immutable · travels with the payload · replayed on reconnect in timestamp order valid on fail on fail deferred sync on reconnect · timestamp order invalid

Prerequisites and Dependencies

The router is a coordination layer. It assumes payloads are already typed and validated, and that concrete transport clients and a local store are injected into it; it does not parse raw hardware frames or open sockets itself.

  • Normalized, validated telemetry. Payloads must already conform to the identifiers and physical-range constraints defined by the Field Schema Design specification — a stable device_id, a canonical field-zone reference, and typed metrics. Ingestion and parsing of raw ISOBUS or MQTT frames is owned upstream by the Equipment Telemetry Parsing service in the Farm Data Ingestion & Field Boundary Synchronization framework.
  • A compliance outcome to route. High-priority payloads carry the compliance state produced by the EPA/USDA Rule Mapping engine. A NON_COMPLIANT or degraded outcome changes routing priority — those events must never be dropped in favor of routine telemetry.
  • Transport clients and a durable local store. Two async transport clients (primary and secondary — e.g. cellular MQTT and an Iridium short-burst gateway) plus a local persistence layer with fsync-backed durability (SQLite in WAL mode or an on-disk append log). The store must survive a hard power cycle of the field gateway.
  • Python runtime and libraries. The engine targets Python 3.10+ so it can use match/case for outcome routing. It relies only on the standard library: asyncio for non-blocking timeout enforcement (see the official asyncio task reference), plus logging, dataclasses, enum, and datetime.
Dependency Minimum version Role in the subsystem
python 3.10 match/case directive routing, structural typing
asyncio (stdlib) — Non-blocking wait_for timeout enforcement per path
logging (stdlib) — Structured, one-object-per-line audit emission
Local store (SQLite WAL / append log) — fsync-durable buffering that survives power loss

Architecture of This Subsystem

The router is a chain-of-responsibility evaluated inside a single async task. Each link is a gate: a payload either exits successfully at that gate or falls through to the next, and every transition writes an audit entry onto the payload itself before control moves on. Because the audit log travels with the payload, a record that is eventually buffered locally still carries the full history of the paths it tried and why each failed.

Four gates compose the chain:

  1. Schema and skew gate — rejects structurally invalid payloads or ones with clock skew beyond tolerance, quarantining them so a bad frame never consumes scarce satellite bandwidth.
  2. Primary egress — the low-latency path, with a strict short timeout. Most traffic exits here.
  3. Secondary egress — the resilient path, with a relaxed timeout, tried only after the primary fails.
  4. Local buffer — durable persistence for deferred sync, the terminal guarantee that no payload is ever lost when every network path is down.
RouteStatus transitions and actuator directives A vertical guard chain of decision nodes on the left maps to terminal outcomes on the right. route_payload writes ROUTE_INIT and enters the schema and clock-skew gate; an invalid or skewed payload branches right to QUARANTINED, which maps to route_to_dead_letter. A valid payload falls through to the primary attempt (asyncio.wait_for, 3 seconds); a confirmed send returns PRIMARY, mapping to mark_delivered. On timeout or failure it falls through to the secondary attempt (relaxed, 5 seconds); a confirmed send returns SECONDARY, also mark_delivered. On further failure it falls through to a durable local buffer persist; success returns LOCAL_BUFFER, mapping to mark_offline_sync_pending. A hard persistence failure raises RoutingError, a fail-loud terminal that alerts on-call. route_payload() ROUTE_INIT audited Schema + skew gate structure · clock guard Primary attempt asyncio.wait_for · 3 s Secondary attempt relaxed · 5 s Local buffer persist fsync-durable valid timeout / fail timeout / fail invalid / skew send confirmed send confirmed persisted local persist fails QUARANTINED route_to_dead_letter PRIMARY mark_delivered SECONDARY mark_delivered LOCAL_BUFFER mark_offline_sync_pending RoutingError fail-loud · alert on-call

Integration points with sibling subsystems:

  • Inputs: validated TelemetryPayload objects with an optional compliance flag set, plus injected primary/secondary transport clients and a local store.
  • Outputs: a single RouteStatus per payload, an immutable per-payload audit log, and — when buffered — a deferred-sync obligation replayed in chronological order on reconnect.
  • Cross-references: the agronomic sibling Fallback Application Chains applies the same conservative-default philosophy to decision fallback, where this subsystem handles transport fallback; polling cadence and reconnect detection are governed by the Async Polling Strategies service.

Step-by-Step Implementation

The dispatch runs in four deterministic stages. Each stage either terminates the chain with a RouteStatus or records a failover and advances.

  1. Validate structure and clock skew. Reject empty or malformed payloads and any timestamp drifting more than the skew tolerance into the future. A failure here quarantines the payload — it is never retried against the network.
  2. Attempt the primary path under a strict timeout. Wrap the send in asyncio.wait_for so a stalled link cannot pin the task. On success, return immediately; on timeout or connection error, audit and fall through.
  3. Attempt the secondary path under a relaxed timeout. The resilient path tolerates higher latency. On success, return; on failure, audit and fall through.
  4. Persist to the local buffer. Durably store the payload for deferred sync. Only a hard storage failure escapes as a RoutingError — a fail-loud terminal condition, because at that point data really is at risk.

The parameters that tune this behavior:

Name Type Default Purpose
primary_timeout_s float 3.0 Latency ceiling for the primary path before failover
secondary_timeout_s float 5.0 Relaxed ceiling for the resilient secondary path
clock_skew_tolerance_s int 300 Max future drift (seconds) tolerated before quarantine
primary_client transport injected Low-latency egress client (e.g. cellular MQTT)
secondary_client transport injected Resilient egress client (e.g. satellite short-burst)
local_store store injected fsync-durable buffer for deferred sync

The following module is production-shaped: strict validation, per-path timeouts, an immutable per-payload audit log, match/case directive routing, and a fail-loud terminal condition.

python 
import asyncio
import json
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[logging.StreamHandler()],
)
logger = logging.getLogger("farm_telemetry_router")


class RouteStatus(str, Enum):
    PRIMARY = "primary"
    SECONDARY = "secondary"
    LOCAL_BUFFER = "local_buffer"
    QUARANTINED = "quarantined"


@dataclass
class RouterConfig:
    primary_timeout_s: float = 3.0
    secondary_timeout_s: float = 5.0
    clock_skew_tolerance_s: int = 300


@dataclass
class TelemetryPayload:
    device_id: str
    timestamp: float
    metrics: dict[str, Any]
    compliance_flags: list[str] = field(default_factory=list)
    routing_attempts: int = 0
    audit_log: list[dict[str, Any]] = field(default_factory=list)


class RoutingError(Exception):
    """Raised when the fallback chain is exhausted and local persistence fails."""


class FallbackRouter:
    def __init__(self, primary_client, secondary_client, local_store,
                 config: RouterConfig | None = None):
        self.primary = primary_client
        self.secondary = secondary_client
        self.local_store = local_store
        self.config = config or RouterConfig()

    async def _audit(self, payload: TelemetryPayload, event: str, details: str) -> None:
        """Append an immutable audit entry to the payload and emit a structured log."""
        entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "device_id": payload.device_id,
            "event": event,
            "details": details,
            "routing_attempts": payload.routing_attempts,
            "compliance_flags": payload.compliance_flags,
        }
        payload.audit_log.append(entry)
        logger.info(json.dumps(entry, default=str, sort_keys=True))

    def _validate(self, payload: TelemetryPayload) -> bool:
        """Strict structural and clock-skew validation before any network attempt."""
        if not payload.device_id or not isinstance(payload.metrics, dict) or not payload.metrics:
            return False
        # Reject payloads whose timestamp drifts too far into the future (skew guard)
        if payload.timestamp > time.time() + self.config.clock_skew_tolerance_s:
            return False
        return True

    async def _try_path(self, client, payload: TelemetryPayload,
                        timeout: float, label: str) -> bool:
        """Attempt one egress path under a strict timeout. Returns True on confirmed send."""
        try:
            await asyncio.wait_for(client.send(payload), timeout=timeout)
            await self._audit(payload, "ROUTE_SUCCESS", f"{label} egress confirmed")
            return True
        except asyncio.TimeoutError:
            await self._audit(payload, f"{label}_TIMEOUT", f"Latency exceeded ({timeout}s)")
        except ConnectionError as exc:
            await self._audit(payload, f"{label}_CONN_FAIL", str(exc))
        except Exception as exc:  # never let an unexpected transport bug drop the payload
            await self._audit(payload, f"{label}_UNEXPECTED",
                              f"{type(exc).__name__}: {exc}")
        return False

    async def route_payload(self, payload: TelemetryPayload) -> RouteStatus:
        await self._audit(payload, "ROUTE_INIT", "Dispatch initiated")

        # Gate 1: schema + clock-skew validation
        if not self._validate(payload):
            await self._audit(payload, "SCHEMA_FAIL",
                              "Quarantined: invalid structure or clock skew")
            return RouteStatus.QUARANTINED

        # Gate 2: primary path, strict timeout
        if await self._try_path(self.primary, payload,
                                 self.config.primary_timeout_s, "PRIMARY"):
            return RouteStatus.PRIMARY

        payload.routing_attempts += 1
        await self._audit(payload, "FAILOVER_TRIGGER", "Transitioning to secondary path")

        # Gate 3: secondary path, relaxed timeout
        if await self._try_path(self.secondary, payload,
                                 self.config.secondary_timeout_s, "SECONDARY"):
            return RouteStatus.SECONDARY

        payload.routing_attempts += 1
        await self._audit(payload, "FAILOVER_TRIGGER", "All upstream failed; persisting locally")

        # Gate 4: durable local buffer (terminal guarantee)
        try:
            self.local_store.persist(payload)
            await self._audit(payload, "LOCAL_PERSIST", "Payload buffered for deferred sync")
            return RouteStatus.LOCAL_BUFFER
        except Exception as exc:
            await self._audit(payload, "LOCAL_FAIL", f"Critical storage failure: {exc}")
            raise RoutingError("Fallback chain exhausted; data at risk") from exc


def actuator_directive(status: RouteStatus) -> str:
    """Map a routing outcome to a downstream tracking directive using match/case."""
    match status:
        case RouteStatus.PRIMARY | RouteStatus.SECONDARY:
            return "mark_delivered"
        case RouteStatus.LOCAL_BUFFER:
            return "mark_offline_sync_pending"
        case RouteStatus.QUARANTINED:
            return "route_to_dead_letter"

On the happy path the router emits one self-describing JSON audit line per transition — an init, then a success:

json 
{"compliance_flags": [], "details": "Dispatch initiated", "device_id": "pivot-07", "event": "ROUTE_INIT", "routing_attempts": 0}
{"compliance_flags": [], "details": "PRIMARY egress confirmed", "device_id": "pivot-07", "event": "ROUTE_SUCCESS", "routing_attempts": 0}

A degraded field gateway that fails the primary path but recovers on the secondary produces a timeout, a failover, then a success:

json 
{"details": "Latency exceeded (3.0s)", "device_id": "pivot-07", "event": "PRIMARY_TIMEOUT", "routing_attempts": 0}
{"details": "Transitioning to secondary path", "device_id": "pivot-07", "event": "FAILOVER_TRIGGER", "routing_attempts": 1}
{"details": "SECONDARY egress confirmed", "device_id": "pivot-07", "event": "ROUTE_SUCCESS", "routing_attempts": 1}

Edge Cases and Known Failure Modes

Most production incidents in this subsystem trace back to a handful of recurring conditions. Each has a deterministic remediation.

Condition Symptom Fix
Both paths slow but not dead Every payload waits the full 3s + 5s before buffering, backing up the queue Tune primary_timeout_s down to the p99 of the healthy link; do not raise it to “help” a bad link
Clock skew on an edge RTC module Bursts of SCHEMA_FAIL quarantine near midnight or after a cold boot Sync gateway time via NTP on boot; keep clock_skew_tolerance_s at 300, never widen it to mask drift
Transport client raises a non-ConnectionError Payload silently lost if the exception is uncaught The broad except Exception in _try_path audits and falls through — never narrow it to only ConnectionError
Local store fails (full disk / read-only mount) RoutingError raised, dispatch task stops Fail loud by design; alert on-call, free disk, and let the caller re-enqueue — do not swallow this error
Duplicate delivery after a false timeout Primary actually delivered but timed out; secondary re-sends → duplicate record Make client.send idempotent using the payload’s device_id + timestamp as a delivery key at the sink
Buffered payloads replayed out of order Deferred sync reorders events, corrupting the sequence downstream Replay the local buffer strictly by timestamp ascending on reconnect (see Compliance section)

Compliance and Audit Integration

Every routing decision is recoverable because the audit log lives on the payload and is append-only in spirit — the router only ever appends entries, it never rewrites or removes them. When a payload finally reaches a sink (immediately, or after deferred sync), its complete routing history travels with it and is written into the immutable ledger owned by the compliance layer. An inspector can therefore reconstruct not just that a spray event was recorded, but the exact path it took, how many attempts it cost, and whether it spent time buffered offline.

The buffered path is where compliance integrity is most at risk, so two guarantees are non-negotiable. First, ordering: deferred payloads are replayed strictly in ascending timestamp order on reconnect, so the downstream tracking layer reconstructs the true field sequence rather than the order the network happened to recover. Second, state honesty: when the router buffers, the actuator_directive maps the outcome to mark_offline_sync_pending rather than mark_delivered, so the tracking layer flags a device as offline-sync-pending, not offline-failed — a distinction that matters for EPA 40 CFR recordkeeping and USDA NRCS reporting cycles, where an apparent data gap can trigger penalties. Downstream, the Security & Access Boundaries subsystem governs who may read those audit trails and replay buffered events, and Python’s logging documentation covers the structured-handler setup that keeps every audit line machine-ingestible.

Verification

Confirm correct operation in staging with a reproducible injected-fault scenario before any routing change reaches production. Stub the primary client to always time out and the secondary to succeed, then assert the exact expected status and audit shape.

python 
class _StubTimeout:
    async def send(self, payload):
        await asyncio.sleep(10)  # forces the wait_for timeout


class _StubOk:
    async def send(self, payload):
        return None


class _StubStore:
    def __init__(self):
        self.buffered = []

    def persist(self, payload):
        self.buffered.append(payload)


def test_primary_timeout_falls_over_to_secondary():
    router = FallbackRouter(_StubTimeout(), _StubOk(), _StubStore(),
                            RouterConfig(primary_timeout_s=0.1, secondary_timeout_s=0.1))
    payload = TelemetryPayload(device_id="pivot-07", timestamp=time.time(),
                               metrics={"soil_moisture_pct": 22.4})
    status = asyncio.run(router.route_payload(payload))

    assert status is RouteStatus.SECONDARY
    assert actuator_directive(status) == "mark_delivered"
    events = [e["event"] for e in payload.audit_log]
    assert "PRIMARY_TIMEOUT" in events
    assert events[-1] == "ROUTE_SUCCESS"

The run must emit a PRIMARY_TIMEOUT, a FAILOVER_TRIGGER, and a final ROUTE_SUCCESS — and no LOCAL_PERSIST. Then swap the secondary for a second _StubTimeout and confirm the status becomes LOCAL_BUFFER, the directive becomes mark_offline_sync_pending, and exactly one payload lands in the stub store. A staging run that quarantines a valid payload, drops a payload without buffering it, or reorders the buffered replay is a release blocker.

Frequently Asked Questions

Why not send to the primary and secondary paths concurrently for speed?

Concurrent fan-out risks duplicate delivery: if both paths succeed, the sink records the application twice, which reads as an over-application during an audit. The chain is deliberately sequential so each payload has exactly one confirmed egress. If latency is the concern, tune primary_timeout_s down to the healthy link’s p99 rather than parallelizing.

What happens to a payload when every network path is down?

It is persisted to the fsync-durable local buffer and the router returns LOCAL_BUFFER, never a silent drop. The payload carries its full audit history into storage and is replayed in strict timestamp order on reconnect. Only a hard local-storage failure escalates — as a fail-loud RoutingError — so on-call can intervene before data is genuinely at risk.

How does the router avoid corrupting the event sequence during deferred sync?

Buffered payloads are replayed in ascending timestamp order, not in the order the network recovered, and each carries an idempotency key (device_id + timestamp) so a re-send after a false timeout collapses to a single record at the sink. The downstream tracking layer therefore reconstructs the true field sequence regardless of how connectivity came back.

Up: Agricultural Automation System Architecture & Compliance

Related