Building a Backup Error Severity Classifier

This page implements the decision core of the parent error categorization framework: a runnable Python classifier that takes the raw failure events emitted by a checksum validation pipeline, a page corruption scan, or any other validation stage, and sorts each one into a severity tier that maps to a single orchestrator action — a deterministic corruption becomes QUARANTINE, a transient storage error becomes RETRY with backoff, and an objective breach against a target set by your RTO/RPO mapping becomes a capacity signal rather than a page. The classifier encodes its rules as data so the taxonomy is auditable and changes without a code deploy, applies tolerance windows so a single flaky read never halts a drill, and suppresses duplicate alerts so that one corrupt volume does not generate a thousand pages; the tiered verdicts it emits are the events that a validation telemetry and metrics layer counts and trends over time.

Architecture and Execution Model

Figure. Each raw failure event is matched against ordered data rules, assigned a tier, run through a tolerance-and-dedupe window, and turned into a single orchestrator action and exit code.

The classifier is deliberately a pure function wrapped in a small stateful gate. The matching stage is stateless: given an event and an ordered rule list, the first matching rule assigns a tier, so the taxonomy lives entirely in data and can be reviewed as a table by an auditor who never reads Python. The suppression stage is where state lives, and it enforces two independent policies — a tolerance window that only escalates a transient signal once it recurs beyond a count within a time span, and a dedupe window that collapses repeat alerts for the same fingerprint so a storm of identical corruption events pages once, not continuously. Separating the two matters operationally: tolerance is about not overreacting to a single blip, while dedupe is about not drowning the on-call in a real but repetitive incident, and conflating them either hides a worsening trend or reintroduces the alert fatigue the framework exists to remove.

Prerequisites

• Python 3.9+ — the classifier uses dataclasses, enum, and collections.deque; everything is standard library with no runtime dependency.
• A rule table describing your taxonomy. It ships inline below as a Python list, but in production render it from a reviewed YAML or JSON file so changes are a config commit, not a deploy.
• A source of failure events: the structured records emitted by upstream validators (checksum mismatch, unreadable volume, restore-time breach). Each event needs a code, a source, and a detail string.
• An optional metrics sink: if you export to Prometheus, reserve counter names now (see the integration notes) so tiers and actions are trended from day one.

Production Implementation

The classifier encodes tiers, rules, and actions as data, matches each event to the first applicable rule, and then runs the verdict through a tolerance-and-dedupe gate before returning the orchestrator action and the process exit code. It runs as-is under python3 error_classifier.py (a self-test main is included) and exits 0 when no event escalated, 1 when at least one event demands a halt, and 2 on a malformed rule table.

#!/usr/bin/env python3
"""error_classifier.py — map backup-validation failures to severity tiers.

Rules are data, not code. Each event is matched to the first applicable rule,
assigned a severity tier, then run through tolerance and dedupe windows before
an orchestrator action and a process exit code are returned:
    0  nothing escalated (or only retries/signals) -> drill continues
    1  at least one QUARANTINE-tier event           -> halt and escalate
    2  malformed rule table                         -> abort
"""
import sys
import time
from collections import deque
from dataclasses import dataclass, field
from enum import IntEnum
from typing import Deque, Dict, List, Optional, Tuple

EXIT_OK = 0
EXIT_HALT = 1
EXIT_USAGE = 2


class Tier(IntEnum):
    """Ordered so a numerically higher tier is more severe."""
    INFO = 0        # objective signal, no action beyond a metric
    RETRY = 1       # transient; retry with backoff
    QUARANTINE = 2  # deterministic corruption; halt and isolate


ACTION_BY_TIER = {
    Tier.INFO: "signal_capacity",
    Tier.RETRY: "retry_with_backoff",
    Tier.QUARANTINE: "quarantine_and_halt",
}


@dataclass(frozen=True)
class Rule:
    """One taxonomy row. `match_codes` and `match_sources` are OR-ed within,
    AND-ed across; an empty set means 'any'."""
    name: str
    tier: Tier
    match_codes: frozenset = field(default_factory=frozenset)
    match_sources: frozenset = field(default_factory=frozenset)


@dataclass
class Event:
    code: str
    source: str
    detail: str = ""
    ts: float = field(default_factory=time.time)

    def fingerprint(self) -> str:
        return f"{self.source}:{self.code}"


# The taxonomy as data. Order matters: first match wins, so specific rules
# precede the catch-all. In production this list is rendered from reviewed YAML.
DEFAULT_RULES: List[Rule] = [
    Rule("checksum_divergence", Tier.QUARANTINE,
         frozenset({"DIGEST_MISMATCH", "PAGE_CRC_FAIL"})),
    Rule("unreadable_page", Tier.QUARANTINE,
         frozenset({"TORN_PAGE", "UNREADABLE_BLOCK"})),
    Rule("transient_storage", Tier.RETRY,
         frozenset({"IO_TIMEOUT", "CONN_RESET", "THROTTLED"})),
    Rule("rpo_breach", Tier.INFO,
         frozenset({"RPO_EXCEEDED", "WINDOW_SHORTFALL"})),
    Rule("catch_all", Tier.RETRY),  # unknown -> retry once, never silently pass
]


class SuppressionGate:
    """Tolerance windows (escalate only on recurrence) + dedupe (rate-limit)."""

    def __init__(self, tolerance: int = 3, window_s: float = 300.0,
                 dedupe_s: float = 60.0):
        self.tolerance = tolerance
        self.window_s = window_s
        self.dedupe_s = dedupe_s
        self._recent: Dict[str, Deque[float]] = {}
        self._last_alert: Dict[str, float] = {}

    def escalate_transient(self, fp: str, now: float) -> bool:
        """A RETRY-tier signal only escalates once it recurs past tolerance."""
        hits = self._recent.setdefault(fp, deque())
        hits.append(now)
        while hits and now - hits[0] > self.window_s:
            hits.popleft()
        return len(hits) >= self.tolerance

    def should_alert(self, fp: str, now: float) -> bool:
        """Rate-limit identical fingerprints so a storm pages once."""
        last = self._last_alert.get(fp)
        if last is not None and now - last < self.dedupe_s:
            return False
        self._last_alert[fp] = now
        return True


def match_rule(event: Event, rules: List[Rule]) -> Rule:
    for rule in rules:
        code_ok = not rule.match_codes or event.code in rule.match_codes
        src_ok = not rule.match_sources or event.source in rule.match_sources
        if code_ok and src_ok:
            return rule
    # Unreachable when a catch-all exists, but never fail open.
    return Rule("implicit_halt", Tier.QUARANTINE)


def classify(event: Event, rules: List[Rule],
             gate: SuppressionGate) -> Tuple[Tier, str, bool]:
    """Return (effective_tier, action, alert?) for one event."""
    rule = match_rule(event, rules)
    tier = rule.tier
    fp = event.fingerprint()

    if tier is Tier.RETRY:
        # A transient that keeps recurring is really a capacity/corruption
        # problem: promote it to QUARANTINE once it breaks tolerance.
        if gate.escalate_transient(fp, event.ts):
            tier = Tier.QUARANTINE

    alert = gate.should_alert(fp, event.ts) if tier >= Tier.RETRY else False
    return tier, ACTION_BY_TIER[tier], alert


def classify_batch(events: List[Event], rules: List[Rule],
                   gate: SuppressionGate) -> int:
    halt = False
    for ev in events:
        tier, action, alert = classify(ev, rules, gate)
        flag = "ALERT" if alert else "muted"
        print(f"{ev.fingerprint():28} | {tier.name:10} | {action:20} | {flag}")
        if tier is Tier.QUARANTINE:
            halt = True
    return EXIT_HALT if halt else EXIT_OK


def _validate_rules(rules: List[Rule]) -> bool:
    return bool(rules) and all(isinstance(r.tier, Tier) for r in rules)


def main() -> int:
    if not _validate_rules(DEFAULT_RULES):
        print("malformed rule table", file=sys.stderr)
        return EXIT_USAGE

    gate = SuppressionGate(tolerance=3, window_s=300.0, dedupe_s=60.0)
    now = time.time()
    sample = [
        Event("IO_TIMEOUT", "checksum_pipeline", "slow NFS read", now),
        Event("IO_TIMEOUT", "checksum_pipeline", "slow NFS read", now + 1),
        Event("IO_TIMEOUT", "checksum_pipeline", "slow NFS read", now + 2),
        Event("DIGEST_MISMATCH", "checksum_pipeline", "offset 2952790016", now + 3),
        Event("RPO_EXCEEDED", "restore_drill", "lag 14m > 5m target", now + 4),
    ]
    return classify_batch(sample, DEFAULT_RULES, gate)


if __name__ == "__main__":
    sys.exit(main())

The escalation rule in classify is the classifier’s most important behavior: a transient IO_TIMEOUT is retried on its first two occurrences, but the third within the five-minute window trips the tolerance threshold and is promoted to QUARANTINE, because a storage error that will not clear is no longer transient — it is a failing volume that must halt the drill. The dedupe gate is orthogonal: even after promotion, identical fingerprints only page once per dedupe_s, so the on-call sees one actionable alert instead of a flood. Because QUARANTINE is the fail-closed default of match_rule, an event that matches no rule and somehow slips past the catch-all still halts rather than being silently promoted.

Step-by-Step Execution Walkthrough

1. Run the self-test. python3 error_classifier.py classifies the built-in sample; watch the third IO_TIMEOUT promote from RETRY to QUARANTINE and the run exit 1.
2. Externalize the rules. Move DEFAULT_RULES into a reviewed YAML file and load it at startup so taxonomy changes are a config commit an auditor can approve.
3. Feed real events. Replace the sample list with the structured records your validators emit; keep code, source, and ts populated so matching and windowing work.
4. Tune the windows to your noise floor. Widen dedupe_s for a chatty source, and raise tolerance where a couple of transient reads per drill are normal and expected.
5. Branch on the exit code. The orchestrator reads $?: 0 lets the drill continue, 1 halts and escalates on any quarantine-tier verdict, 2 aborts on a malformed rule table.

Verification and Expected Output

The self-test run demonstrates promotion and dedupe, then exits 1 because a deterministic mismatch reached the quarantine tier:

checksum_pipeline:IO_TIMEOUT  | RETRY      | retry_with_backoff   | ALERT
checksum_pipeline:IO_TIMEOUT  | RETRY      | retry_with_backoff   | muted
checksum_pipeline:IO_TIMEOUT  | QUARANTINE | quarantine_and_halt  | muted
checksum_pipeline:DIGEST_MISMATCH | QUARANTINE | quarantine_and_halt | ALERT
restore_drill:RPO_EXCEEDED    | INFO       | signal_capacity      | muted
$ echo $?
1

Success means three observable facts: the first IO_TIMEOUT alerts while its immediate repeats are muted by dedupe, the third IO_TIMEOUT is promoted to QUARANTINE by the tolerance window, and the RPO_EXCEEDED event is an INFO capacity signal that never pages. An all-transient batch that never breaks tolerance exits 0.

Failure Modes and Troubleshooting

Integration Notes

The classifier is the routing layer between raw validator output and orchestration, so wire it where those signals are produced. Upstream, it consumes the exit codes and divergence records from the checksum validation pipeline and page corruption scanning; downstream, its QUARANTINE verdict halts an Airflow DAG or raises a Celery Retry only for genuinely transient tiers. Export the tier and action as labelled counters — dr_validation_events_total{tier="quarantine",action="quarantine_and_halt"} and a dr_validation_alerts_suppressed_total for the dedupe rate — so the suppression policy itself is observable and you can prove alert volume dropped without hiding incidents. Trend those counters in the validation telemetry and metrics layer, and confirm every RPO_EXCEEDED capacity signal reconciles against the objectives in your RTO/RPO mapping. For the standard-library primitives the gate relies on, see the Python collections documentation; for the counter and label semantics the metrics export assumes, see the Prometheus documentation.

Frequently Asked Questions

Related

• Error Categorization Frameworks — the parent framework this classifier implements as its decision core.
• Checksum Validation Pipelines — a primary source of the failure events this classifier routes.
• Page Corruption Scanning Techniques — emits the storage-engine corruption codes that map to the quarantine tier.
• Validation Telemetry and Metrics — counts and trends the tiered verdicts and suppression rate this classifier produces.

This page is one component of the broader Error Categorization Frameworks topic; from there you can move up to the Automated Backup Integrity Check Implementation overview.