Quarantining Bad OSM Features to a Dead-Letter Store Jump to heading#

from __future__ import annotations

import json
import logging
import sqlite3
import traceback
from collections.abc import Callable, Iterable
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger("osm.errors.quarantine")

SCHEMA = """
CREATE TABLE IF NOT EXISTS dead_letter (
    id            INTEGER PRIMARY KEY,
    osm_type      TEXT NOT NULL,
    osm_id        INTEGER NOT NULL,
    stage         TEXT NOT NULL,
    error_type    TEXT NOT NULL,
    error_message TEXT NOT NULL,
    traceback     TEXT,
    raw_input     TEXT NOT NULL,      -- sufficient to reproduce, alone
    run_id        TEXT NOT NULL,
    code_version  TEXT NOT NULL,
    quarantined_at TEXT NOT NULL,
    status        TEXT NOT NULL DEFAULT 'open',  -- open | resolved | permanent
    resolved_at   TEXT
);
CREATE INDEX IF NOT EXISTS idx_dl_open ON dead_letter (status, stage, error_type);
"""


@dataclass(frozen=True)
class Context:
    run_id: str
    code_version: str


class DeadLetterStore:
    def __init__(self, path: Path, context: Context) -> None:
        self.conn = sqlite3.connect(path)
        self.conn.executescript(SCHEMA)
        self.context = context

    def quarantine(self, osm_type: str, osm_id: int, stage: str,
                   raw_input: dict, error: BaseException) -> None:
        self.conn.execute("""
            INSERT INTO dead_letter (osm_type, osm_id, stage, error_type,
              error_message, traceback, raw_input, run_id, code_version,
              quarantined_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (osm_type, osm_id, stage, type(error).__name__, str(error),
              "".join(traceback.format_exception(error))[-4000:],
              json.dumps(raw_input, default=str),
              self.context.run_id, self.context.code_version,
              datetime.now(timezone.utc).isoformat(timespec="seconds")))
        self.conn.commit()

    def summary(self) -> list[tuple]:
        rows = self.conn.execute("""
            SELECT stage, error_type, count(*) AS n
            FROM dead_letter WHERE status = 'open'
            GROUP BY stage, error_type ORDER BY n DESC
        """).fetchall()
        for stage, error_type, n in rows:
            logger.info("%6d  %-20s %s", n, stage, error_type)
        return rows

    def replay(self, process: Callable[[dict], None],
               stage: str | None = None) -> tuple[int, int]:
        """Re-run quarantined records against the CURRENT code.

        The stored raw input is replayed, not a fresh extract: the point is to
        test the fix against exactly what broke, which a new extract may no
        longer contain.
        """
        query = "SELECT id, raw_input FROM dead_letter WHERE status = 'open'"
        params: list = []
        if stage:
            query += " AND stage = ?"
            params.append(stage)

        fixed = still_failing = 0
        for record_id, raw in self.conn.execute(query, params).fetchall():
            try:
                process(json.loads(raw))
            except Exception:
                still_failing += 1
                continue
            self.conn.execute(
                "UPDATE dead_letter SET status = 'resolved', resolved_at = ? "
                "WHERE id = ?",
                (datetime.now(timezone.utc).isoformat(timespec="seconds"),
                 record_id))
            fixed += 1
        self.conn.commit()
        logger.info("replay: %d resolved, %d still failing", fixed, still_failing)
        return fixed, still_failing

    def age_alert(self, days: int = 14) -> int:
        """An unexamined quarantine is a slow leak. Make it visible."""
        stale, = self.conn.execute("""
            SELECT count(*) FROM dead_letter WHERE status = 'open'
            AND quarantined_at < datetime('now', ?)
        """, (f"-{days} days",)).fetchone()
        if stale:
            logger.warning("%d record(s) have been quarantined for over %d days",
                           stale, days)
        return stale


def guarded(store: DeadLetterStore, stage: str, process: Callable[[dict], None]
            ) -> Callable[[Iterable[dict]], int]:
    """Wrap a stage so one bad feature never stops the run."""
    def run(features: Iterable[dict]) -> int:
        processed = 0
        for feature in features:
            try:
                process(feature)
                processed += 1
            except Exception as error:
                store.quarantine(feature.get("type", "?"),
                                 int(feature.get("id", 0)), stage,
                                 feature, error)
        return processed
    return run


if __name__ == "__main__":
    logger.info("store the raw input, replay after a fix, promote what passes")