Fallback Routing Logic for Municipal Waste Operations
Deterministic degradation state machines that preserve service continuity under solver failure.
When the primary optimization engine times out, diverges, or loses the telemetry it depends on, something still has to tell a driver which bin to empty next. Fallback routing logic is that something: a bounded, deterministic path that produces a legally dispatchable route from precomputed data when the live solver cannot. Without it, a single solver stall during morning dispatch cascades into missed stops, and missed stops in municipal waste collection are not a degraded objective value — they are contractual penalties, residential complaints, and, when a truck is sent past a driver’s hours ceiling or into a closed transfer station to “catch up,” a compliance violation that a public-records request surfaces years later.
This page is the implementation reference for that safety net. It sits under the Core Architecture & Compliance Mapping framework and inherits its central rule: given the same inputs, the fallback must produce byte-identical output and byte-identical audit hashes every time. Heuristic improvisation is the enemy here. The failure that triggered the fallback is exactly the moment you cannot afford a route that is impossible to replay or defend. Everything below is built so that a degraded dispatch decision can be re-hashed and justified to an auditor.
Prerequisites
The fallback tier deliberately depends on fewer moving parts than the primary engine, but the parts it does use must be pinned. It runs on Python 3.10+ (the code uses structural match and X | Y unions), the same OR-Tools routing core documented in the OR-Tools Implementation reference, and Pydantic v2 for the contingency schema.
python -m pip install \
"ortools==9.11.4210" \
"pydantic==2.9.2"Before the fallback can run it needs a frozen snapshot of the world it will route over: a precomputed distance matrix, cached service windows, and the active route assignments captured the instant degradation was detected. That snapshot is the data contract between the failing primary and the fallback, and it is validated with the same strict-schema posture as the Route Schema Design specification. Coerce nothing silently; reject anything malformed.
from __future__ import annotations
from pydantic import BaseModel, Field, ValidationError
class FallbackStop(BaseModel):
"""A single serviceable stop, stripped of live-optimization fields."""
stop_id: str
node_index: int = Field(ge=0) # row/col in the cached distance matrix
lat: float = Field(ge=-90, le=90)
lon: float = Field(ge=-180, le=180)
demand_kg: int = Field(ge=0)
service_seconds: int = Field(ge=0)
window_open_s: int = Field(ge=0) # seconds from shift start
window_close_s: int = Field(ge=0)
compliance_code: str = Field(pattern=r"^[A-Z]{3}-\d{4}$")
class FallbackSnapshot(BaseModel):
"""Frozen world-state handed from the failing primary to the fallback tier."""
route_id: str
vehicle_id: str
vehicle_capacity_kg: int = Field(gt=0)
depot_node: int = Field(ge=0)
shift_seconds: int = Field(gt=0) # HOS-bounded route-duration ceiling
matrix_seconds: list[list[int]] # symmetric travel-time matrix
stops: list[FallbackStop]
fallback_reason: str
def load_snapshot(raw: dict) -> FallbackSnapshot:
"""Parse-or-reject. A malformed snapshot must never reach the solver."""
return FallbackSnapshot.model_validate(raw)If load_snapshot raises ValidationError, the correct behaviour is to halt this route and escalate to manual review — never to guess a missing coordinate. A fallback that repairs its own inputs is no longer deterministic.
Core Implementation
The fallback path has three stages: decide whether to activate (trigger evaluation with hysteresis), take control transactionally (state handoff), and produce a route from static data under a hard time limit (bounded solve). Each stage is a separate, testable unit.
1. Trigger evaluation with a hysteresis gate
Threshold evaluation must be stateless per call and idempotent, or transient network flares will make the controller oscillate between primary and fallback. A hysteresis gate solves this: it only flips to the degraded state after the trigger condition holds for N consecutive evaluation cycles, and only flips back after the same number of healthy cycles.
from dataclasses import dataclass, field
@dataclass
class HysteresisGate:
"""Debounces a boolean health signal to prevent fallback oscillation."""
activate_after: int = 3 # consecutive breaches before FALLBACK_ACTIVE
recover_after: int = 5 # consecutive healthy cycles before PRIMARY_HEALTHY
_breaches: int = field(default=0)
_recoveries: int = field(default=0)
active: bool = field(default=False)
def observe(self, breached: bool) -> bool:
"""Feed one evaluation cycle; returns the debounced fallback state."""
if breached:
self._breaches += 1
self._recoveries = 0
if self._breaches >= self.activate_after:
self.active = True
else:
self._recoveries += 1
self._breaches = 0
if self._recoveries >= self.recover_after:
self.active = False
return self.active
def is_primary_degraded(metrics: dict[str, float]) -> bool:
"""Deterministic trigger matrix over rolling telemetry aggregates."""
return (
metrics.get("solver_p95_latency_s", 0.0) > 2.5
or metrics.get("gps_fix_loss_intervals", 0) >= 3
or metrics.get("solver_infeasible_rate", 0.0) > 0.05
)The rolling metrics themselves come from the Telematics Sensor Data Ingestion pipeline, which reconciles sensor gaps before they ever reach this gate; gps_fix_loss_intervals is one of the signals it emits. Feeding the gate raw, unreconciled telemetry would let a single dropped GPS burst trip a fleet-wide fallback.
2. Transactional state handoff
Taking control is the dangerous moment. If the secondary solver fails halfway, you must not leave orphaned dispatch records pointing at a half-built route. Wrap the handoff in a context manager that captures pre-transition metrics, commits atomically on success, and reverts to the last known stable state on any exception. Structured logging follows the site-wide JSONFormatter pattern — a custom class, because the standard logging module ships no JSON formatter of its own.
import json
import logging
from contextlib import contextmanager
from typing import Iterator
logger = logging.getLogger("fallback.controller")
class JSONFormatter(logging.Formatter):
"""Append-only, machine-parseable audit lines for compliance review."""
def format(self, record: logging.LogRecord) -> str:
payload: dict[str, object] = {
"ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
for key in ("route_id", "fallback_reason", "compliance_hash", "trigger"):
if hasattr(record, key):
payload[key] = getattr(record, key)
return json.dumps(payload, separators=(",", ":"), sort_keys=True)
class FallbackTransitionError(Exception):
"""Raised when a rollback executes during a fallback handoff."""
def capture_solver_metrics() -> dict[str, float]:
"""Return current primary-solver health metrics (stub: wire to your metrics store)."""
return {}
def commit_fallback_routes(snapshot: "FallbackSnapshot", route: list[int]) -> None:
"""Persist the fallback route to the dispatch queue (stub)."""
def revert_to_last_known_state(snapshot: "FallbackSnapshot") -> None:
"""Restore the pre-transition stable dispatch state (stub)."""
@contextmanager
def fallback_state_handoff(snapshot: "FallbackSnapshot") -> Iterator[list[int]]:
"""Atomic control transfer: commit the produced route or roll everything back."""
pre = capture_solver_metrics()
produced: list[int] = []
logger.info(
"Initiating fallback handoff",
extra={"route_id": snapshot.route_id, "fallback_reason": snapshot.fallback_reason},
)
try:
yield produced # caller appends the solved node order
commit_fallback_routes(snapshot, produced)
except Exception as exc:
logger.error("Fallback failed; reverting to last known state",
extra={"route_id": snapshot.route_id}, exc_info=True)
revert_to_last_known_state(snapshot)
raise FallbackTransitionError("state rollback executed") from exc
finally:
logger.info("Fallback handoff audit",
extra={"route_id": snapshot.route_id,
"fallback_reason": snapshot.fallback_reason})
_ = (pre, capture_solver_metrics()) # both snapshots persisted by the audit sink3. Bounded static solve
The fallback solver is intentionally boring: a single-vehicle traversal over the cached matrix with the two hard dimensions that can never be relaxed — cumulative capacity and the HOS-bounded route duration. Time windows are registered as a Time dimension. The critical difference from the primary engine is the search parameters: a strict wall-clock limit and a deterministic first-solution strategy, so the fallback always returns something dispatchable inside a predictable envelope rather than chasing an optimum it may never reach.
from ortools.constraint_solver import pywrapcp, routing_enums_pb2
def solve_fallback(snapshot: "FallbackSnapshot", time_limit_s: int = 2) -> list[int]:
"""Produce a deterministic, capacity- and HOS-feasible node order from cached data."""
stops = snapshot.stops
n = len(snapshot.matrix_seconds)
manager = pywrapcp.RoutingIndexManager(n, 1, snapshot.depot_node)
routing = pywrapcp.RoutingModel(manager)
# --- travel-time callback (also drives the Time dimension) ---
def travel_cb(from_index: int, to_index: int) -> int:
i = manager.IndexToNode(from_index)
j = manager.IndexToNode(to_index)
service = stops[i].service_seconds if i < len(stops) else 0
return snapshot.matrix_seconds[i][j] + service
transit_idx = routing.RegisterTransitCallback(travel_cb)
routing.SetArcCostEvaluatorOfAllVehicles(transit_idx)
# --- HARD dimension: route duration bounded by the driver's HOS ceiling ---
routing.AddDimension(
transit_idx, 0, snapshot.shift_seconds, True, "Time"
)
time_dim = routing.GetDimensionOrDie("Time")
for node, stop in enumerate(stops):
idx = manager.NodeToIndex(node)
time_dim.CumulVar(idx).SetRange(stop.window_open_s, stop.window_close_s)
# --- HARD dimension: cumulative load bounded by vehicle capacity ---
def demand_cb(from_index: int) -> int:
i = manager.IndexToNode(from_index)
return stops[i].demand_kg if i < len(stops) else 0
demand_idx = routing.RegisterUnaryTransitCallback(demand_cb)
routing.AddDimensionWithVehicleCapacity(
demand_idx, 0, [snapshot.vehicle_capacity_kg], True, "Load"
)
# --- bounded, deterministic search: return fast, never chase an optimum ---
params = pywrapcp.DefaultRoutingSearchParameters()
params.first_solution_strategy = (
routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
)
params.time_limit.FromSeconds(time_limit_s)
solution = routing.SolveWithParameters(params)
if solution is None:
raise FallbackTransitionError("fallback solver returned no feasible route")
order: list[int] = []
index = routing.Start(0)
while not routing.IsEnd(index):
order.append(manager.IndexToNode(index))
index = solution.Value(routing.NextVar(index))
order.append(manager.IndexToNode(index))
return orderWiring the three stages together, a degraded route flows through the gate, the handoff, and the solve as a single deterministic unit:
def run_fallback_cycle(snapshot: "FallbackSnapshot",
gate: "HysteresisGate",
metrics: dict[str, float]) -> list[int] | None:
if not gate.observe(is_primary_degraded(metrics)):
return None # primary still trusted; do nothing
with fallback_state_handoff(snapshot) as produced:
produced[:] = solve_fallback(snapshot)
return producedRegulatory Mapping
Every constraint the fallback keeps hard traces back to a specific rule, and those citations belong in code comments and audit records, not just prose. Degraded mode is precisely when regulators expect the boundaries to hold.
- Route-duration ceiling (
shift_seconds) → FMCSA Hours of Service, 49 CFR Part 395. TheTimedimension’s global span is bounded by the driver’s remaining drive time, including the 11-hour driving limit and 30-minute break rule. The fallback cannot extend a shift to “recover” missed stops; the full state-machine translation lives in How to map DOT hours-of-service to waste routes, and the parameter-by-parameter mapping in the DOT/FMCSA Rule Mapping reference. - Cumulative load (
Loaddimension) → GVWR / axle limits, 49 CFR 393 & state bridge postings. Capacity accumulates monotonically along the path exactly as it does in the primary Capacity & Weight Limits model, so the fallback can never emit a route that overloads the truck even under cost pressure. - Service windows (
window_open_s/window_close_s) → municipal noise and quiet-hour ordinances. These are the same hard Time Window Constraints the primary enforces; a fallback that arrives before a 07:00 residential window is dispatching an illegal collection, not an early one. - Waste-stream destination immutability → EPA e-Manifest, 40 CFR Part 262 (Uniform Hazardous Waste Manifest, EPA Form 8700-22). The
compliance_codeon each stop is frozen at snapshot time; the fallback may reorder stops but may never reroute a regulated waste stream to a different or non-permitted transfer facility.
Access to the snapshot and the dispatch queue it writes to is itself scoped by the Security & Access Boundaries model, so a fallback event cannot become a path around tenant isolation or credential checks.
Validation & Verification
You verify a fallback the same way you verify a fuse: by forcing the fault and confirming the safe behaviour, not by hoping. Three checks matter.
First, assert that the produced route honours both hard dimensions. This is a pytest that constructs a deliberately tight snapshot and confirms the solver never violates capacity or the shift ceiling.
def test_fallback_respects_hard_dimensions() -> None:
matrix = [
[0, 600, 900, 600],
[600, 0, 600, 900],
[900, 600, 0, 600],
[600, 900, 600, 0],
]
stops = [
FallbackStop(stop_id="depot", node_index=0, lat=40.0, lon=-75.0,
demand_kg=0, service_seconds=0,
window_open_s=0, window_close_s=28800, compliance_code="MSW-0001"),
FallbackStop(stop_id="s1", node_index=1, lat=40.1, lon=-75.1,
demand_kg=4000, service_seconds=300,
window_open_s=0, window_close_s=28800, compliance_code="MSW-0001"),
FallbackStop(stop_id="s2", node_index=2, lat=40.2, lon=-75.0,
demand_kg=4000, service_seconds=300,
window_open_s=0, window_close_s=28800, compliance_code="MSW-0001"),
FallbackStop(stop_id="s3", node_index=3, lat=40.1, lon=-74.9,
demand_kg=3000, service_seconds=300,
window_open_s=0, window_close_s=28800, compliance_code="MSW-0001"),
]
snap = FallbackSnapshot(
route_id="R-42", vehicle_id="V-7", vehicle_capacity_kg=12000,
depot_node=0, shift_seconds=28800, matrix_seconds=matrix,
stops=stops, fallback_reason="solver_timeout",
)
order = solve_fallback(snap, time_limit_s=1)
assert order[0] == snap.depot_node and order[-1] == snap.depot_node
assert sorted(order[1:-1]) == [1, 2, 3] # every stop served once
assert sum(s.demand_kg for s in stops) <= snap.vehicle_capacity_kgSecond, assert that the hysteresis gate does not oscillate. Feed it an alternating signal and confirm it stays put.
def test_gate_ignores_transient_flare() -> None:
gate = HysteresisGate(activate_after=3, recover_after=5)
# a single breach surrounded by healthy cycles must NOT activate
assert gate.observe(True) is False
assert gate.observe(False) is False
assert gate.observe(False) is FalseThird, confirm the audit trail is actually emitted. In production, attach the JSONFormatter to the handler and assert the log line for a handoff carries route_id and fallback_reason — these are the fields a compliance query filters on. A fallback that runs but leaves no structured record is, for audit purposes, a fallback that never happened.
Failure Modes & Edge Cases
The whole point of this subsystem is to handle failure, so its own failure modes deserve explicit, coded responses rather than a bare exception.
The fallback solve is itself infeasible. Even the static solver can fail — a truck already near capacity when the primary died, or a window that closed during the outage. Do not widen a hard boundary to force a solution. Instead, relax only the soft structure: drop the least-critical stops onto a follow-up route while every remaining stop stays fully compliant. OR-Tools expresses this as an optional-node penalty (a disjunction), so unreachable stops are deferred rather than the capacity or HOS ceiling being breached.
def solve_with_deferrals(snapshot: "FallbackSnapshot", drop_penalty: int = 10_000_000):
"""Allow stops to be deferred, but never relax a hard compliance dimension."""
# ... identical setup to solve_fallback (matrix, Time, Load dimensions) ...
# see solve_fallback above for the full model construction
manager = pywrapcp.RoutingIndexManager(len(snapshot.matrix_seconds), 1, snapshot.depot_node)
routing = pywrapcp.RoutingModel(manager)
for node in range(1, len(snapshot.stops)):
# high penalty => the solver drops a stop only when no compliant route includes it
routing.AddDisjunction([manager.NodeToIndex(node)], drop_penalty)
return routing # then build dimensions and solve exactly as aboveDeferred stops are logged with their stop_id and rolled into the next dispatch cycle, so service continuity degrades gracefully instead of a route failing wholesale.
Time windows widened by an operator, not the solver. When an outage genuinely spans a window, widening is a dispatcher decision with an audit note, never an automatic solver relaxation. The fallback surfaces the conflicting stop; a human with authority under the operations policy signs off, and that sign-off is hashed into the trail.
Secondary depot assignment. If the assigned depot’s tipping facility is closed or capped mid-outage, the fallback reassigns the route’s terminal node to a pre-approved alternate facility — but only one whose permit accepts the same compliance_code. Rerouting a hazmat stream to a non-permitted station to save miles is the exact failure this whole subsystem exists to prevent.
Rollback on partial commit. If commit_fallback_routes throws after some records are written, fallback_state_handoff invokes revert_to_last_known_state and re-raises as FallbackTransitionError. The route is quarantined for manual review rather than left in a half-dispatched limbo.
Integration Checklist
Complete every item before enabling fallback dispatch in production:
Frequently Asked Questions
Why not just let the primary solver retry instead of building a separate fallback tier?
A retry re-runs the same memory-intensive, telemetry-dependent code path that just failed, often for the same reason (a solver timeout retries into another timeout; a GPS dropout is still dropped). The fallback tier is deliberately decoupled: it uses cached matrices and static windows, so it fails independently of whatever broke the primary. Decoupling is what prevents a cascading failure during peak dispatch.
Can the fallback relax a hard constraint if there is genuinely no feasible route?
No — that inverts the design. Hard constraints (GVWR, HOS, permitted-facility routing) are never relaxed by the solver. When no compliant route exists, the fallback defers the least-critical stops via a disjunction penalty and escalates the conflict to a dispatcher. A human may widen a soft window with an audited sign-off; the software never crosses a regulatory boundary on its own.
How do I stop the controller from flapping between primary and fallback?
Use the HysteresisGate with asymmetric thresholds: require several consecutive breaches before activating and more consecutive healthy cycles before recovering. Combined with rolling telemetry aggregates from the ingestion pipeline (never raw single readings), this debounces transient flares so a one-second latency spike cannot trip a fleet-wide fallback.
What makes the fallback route auditable if it was generated under failure conditions?
Determinism plus append-only logging. The static solve uses a fixed first-solution strategy and a bounded time limit, so the same snapshot always yields the same route. The handoff emits structured JSONFormatter records carrying route_id, fallback_reason, and a compliance_hash, written to append-only storage. An auditor can replay the snapshot, re-derive the route, and re-hash it to confirm nothing was altered after the fact.
Where does the fallback get its distance matrix if the live routing service is down?
From a cache refreshed on a schedule independent of the live service — typically once per shift, versioned with a hash. The fallback never calls a live routing API during an outage; that dependency is exactly what may have failed. The trade-off is staleness (the matrix ignores real-time traffic), which is acceptable because the fallback optimizes for a legally dispatchable route, not a minimal-mileage one.
Related
- DOT/FMCSA Rule Mapping — the HOS and axle rules the fallback keeps hard under degradation.
- Route Schema Design — the canonical contract the fallback snapshot must validate against.
- Security & Access Boundaries — tenant isolation and signed chains that a fallback event must not bypass.
- Capacity & Weight Limits — the cumulative-load dimension mirrored in the static solve.
- Time Window Constraints — the service-window boundaries the fallback enforces identically to the primary.