Implementing Priority Queues for Catastrophic Claims

When a hurricane or a regional freeze fires the CATASTROPHIC tier, a standard first-in-first-out queue buries life-safety claims behind thousands of low-severity submissions that arrived seconds earlier — this guide builds a heap-backed priority queue that surfaces the most urgent loss first, deterministically and without duplicate dispatch.

This page is a deep dive under Dynamic Threshold Tuning, the stage that emits the routing tier this queue must honor the instant a catastrophe override fires or a compressed high_severity_min boundary widens the senior pathway.

Catastrophic weather events, infrastructure failures, and regional emergencies produce instantaneous claim surges that overwhelm linear processing pipelines. The failure mode is specific: under FIFO ordering, a structural-collapse claim with trapped occupants waits behind every cosmetic-damage report submitted ahead of it, so cycle time on the claims that actually need a specialist degrades exactly when it matters most. Three secondary defects compound the problem at production volume:

• Priority inversion. A claim’s severity can escalate after it is already queued — a secondary inspection upgrades a MEDIUM water-damage report to CATASTROPHIC structural failure. A naive heap pins the stale priority and the upgraded claim never advances.
• Duplicate dispatch. The same loss arrives over a mobile app, an IoT telematics feed, and a third-party portal within milliseconds. Without deterministic deduplication the queue fans one event out to multiple adjusters, producing redundant reserves and conflicting audit records.
• Unbounded growth. A surge with no backpressure exhausts heap memory and triggers garbage-collection pauses that stall every consumer at once.

This guide resolves all four with versioned priority tokens, content-addressable idempotency, and bounded capacity with deterministic fallback routing.

• Python 3.11+ for StrEnum and timezone-correct datetime. The standard-library heapq module backs the in-process queue; nothing else is required to run the code below.
• A monotonic clock and sequence source. itertools.count() supplies stable FIFO tiebreaks within a single process; a distributed deployment substitutes a broker-assigned offset.
• An upstream tier signal. Each event must already carry the discrete tier and severity score produced by Automated Severity Scoring Models and gated by Coverage Validation Rules; this queue orders claims, it does not score or coverage-check them.
• An append-only audit sink (a WORM object store or an immutable log topic) for the dispatch records described in the compliance note.

For cross-node deployments, replace the in-process heap with a broker that preserves priority ordering — Redis sorted sets or RabbitMQ with the priority-queue plugin — but keep the envelope and dedup logic identical so behavior is reproducible across both.

The priority key must be a single integer where larger means more urgent, and the content address must be stable across every ingestion source so the same loss always hashes to the same value.

Copy
from __future__ import annotations

import hashlib
import logging
from dataclasses import dataclass
from datetime import datetime, timezone

logger = logging.getLogger("catastrophe.queue")

LIFE_SAFETY_BOOST = 1_000_000
HIGH_EXPOSURE_BOOST = 100_000
HIGH_EXPOSURE_FLOOR = 1_000_000.0


@dataclass(frozen=True)
class CatastropheClaim:
    claim_id: str
    policy_id: str
    loss_event_ts: datetime
    geohash: str
    severity_score: float          # 0.0–1.0 from the scoring model
    life_safety: bool
    estimated_exposure: float


def content_address(claim: CatastropheClaim) -> str:
    """Stable SHA-256 over the identity of the *loss*, not the submission."""
    raw = "|".join((
        claim.policy_id,
        str(int(claim.loss_event_ts.astimezone(timezone.utc).timestamp())),
        claim.geohash,
    )).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def compute_priority(claim: CatastropheClaim) -> int:
    """Larger integer == more urgent. Deterministic over the claim fields."""
    priority = int(round(claim.severity_score * 1000))
    if claim.life_safety:
        priority += LIFE_SAFETY_BOOST
    if claim.estimated_exposure >= HIGH_EXPOSURE_FLOOR:
        priority += HIGH_EXPOSURE_BOOST
    return priority

The hash deliberately excludes claim_id and submission timestamp: a duplicate that arrives over a different channel carries a different claim_id but the same policy, loss instant, and location, so it collapses onto one content address.

heapq is a min-heap, so the entry key is (-priority, sequence): the highest priority sorts first and the monotonic sequence gives a stable FIFO tiebreak between equal priorities. Escalation never mutates an entry in place — it tombstones the old one and pushes a new version, and poll lazily discards any entry the index no longer points at.

Copy
import heapq
import itertools
import threading


class QueueFull(RuntimeError):
    """Raised when a bounded queue rejects a genuinely new claim."""


class _Entry:
    __slots__ = ("neg_priority", "sequence", "version", "content_hash", "claim", "active")

    def __init__(self, neg_priority: int, sequence: int, version: int,
                 content_hash: str, claim: CatastropheClaim) -> None:
        self.neg_priority = neg_priority
        self.sequence = sequence
        self.version = version
        self.content_hash = content_hash
        self.claim = claim
        self.active = True

    def __lt__(self, other: "_Entry") -> bool:
        return (self.neg_priority, self.sequence) < (other.neg_priority, other.sequence)


class CatastrophePriorityQueue:
    def __init__(self, maxsize: int = 100_000) -> None:
        self._heap: list[_Entry] = []
        self._index: dict[str, _Entry] = {}   # content_hash -> live entry
        self._seq = itertools.count()
        self._ver = itertools.count()
        self._lock = threading.RLock()
        self._maxsize = maxsize

submit is the idempotency guard: a colliding hash with an equal-or-lower priority is dropped, a higher priority supersedes the queued version, and a genuinely new claim is rejected once the bounded capacity is reached so memory never grows without limit. poll returns the most urgent live claim and skips tombstones in O(log n) amortized time.

Copy
    def submit(self, claim: CatastropheClaim) -> str:
        chash = content_address(claim)
        priority = compute_priority(claim)
        with self._lock:
            existing = self._index.get(chash)
            if existing is not None:
                queued_priority = -existing.neg_priority
                if priority <= queued_priority:
                    logger.info("dedup_drop hash=%s queued=%d new=%d",
                                chash[:12], queued_priority, priority)
                    return chash
                existing.active = False  # escalation: tombstone the stale version
            elif len(self._index) >= self._maxsize:
                raise QueueFull(f"queue at capacity {self._maxsize}")

            entry = _Entry(
                neg_priority=-priority,
                sequence=next(self._seq),
                version=next(self._ver),
                content_hash=chash,
                claim=claim,
            )
            heapq.heappush(self._heap, entry)
            self._index[chash] = entry
            return chash

    def poll(self) -> CatastropheClaim | None:
        with self._lock:
            while self._heap:
                entry = heapq.heappop(self._heap)
                if not entry.active or self._index.get(entry.content_hash) is not entry:
                    continue  # tombstoned or superseded -> lazy delete
                del self._index[entry.content_hash]
                return entry.claim
            return None

    def depth(self) -> int:
        with self._lock:
            return len(self._index)

Holding self._lock for the whole of submit and poll makes priority computation, tombstoning, and the heap mutation a single atomic step, which is what keeps the queue deterministic under concurrent producers and consumers.

Determinism is the property under test: identical inputs must produce identical ordering, dedup must converge to one claim, and escalation must reorder. Assert all three directly.

Copy
def _claim(cid: str, score: float, life: bool = False, **kw) -> CatastropheClaim:
    base = dict(policy_id="POL-1", geohash="dqcjq",
                loss_event_ts=datetime(2026, 6, 1, tzinfo=timezone.utc),
                estimated_exposure=50_000.0)
    base.update(kw)
    return CatastropheClaim(claim_id=cid, severity_score=score, life_safety=life, **base)


def test_priority_ordering() -> None:
    q = CatastrophePriorityQueue()
    q.submit(_claim("low", 0.20, policy_id="A", geohash="a"))
    q.submit(_claim("crit", 0.90, life=True, policy_id="B", geohash="b"))
    q.submit(_claim("mid", 0.60, policy_id="C", geohash="c"))
    order = [q.poll().claim_id for _ in range(3)]
    assert order == ["crit", "mid", "low"], order


def test_dedup_is_idempotent() -> None:
    q = CatastrophePriorityQueue()
    h1 = q.submit(_claim("mobile", 0.55))
    h2 = q.submit(_claim("portal", 0.55))   # same loss, different channel
    assert h1 == h2 and q.depth() == 1


def test_escalation_supersedes() -> None:
    q = CatastrophePriorityQueue()
    q.submit(_claim("first", 0.30))
    q.submit(_claim("first", 0.95))         # re-scored upward, same loss
    assert q.depth() == 1
    assert q.poll().severity_score == 0.95
    assert q.poll() is None

test_dedup_is_idempotent proves two channels collapse to a single content address; test_escalation_supersedes proves a re-scored claim advances without leaving a stale duplicate behind. Run these on every commit so a refactor of compute_priority or the heap key cannot silently break ordering.

Every dispatch out of poll must emit an append-only record before the claim reaches Adjuster Assignment Algorithms, capturing the content_hash, the composite priority, the version that won, and the threshold version that classified the claim upstream. That lineage lets a compliance officer reconstruct why one catastrophic claim was worked before another — a direct requirement of state Department of Insurance fair-claims-handling rules and the audit-trail contract defined in Core Architecture & Compliance Mapping. Because the priority is a pure function of the claim fields and the queue records the winning version rather than mutating entries in place, the ordering decision is replayable years later. For log-integrity and retention controls, align the sink with NIST SP 800-92 Guide to Computer Security Log Management.

• Dynamic Threshold Tuning — the parent stage that emits the tier this queue prioritizes on
• Automated Severity Scoring Models — produces the severity score that feeds the composite priority key
• Routing high-severity claims to senior adjusters — the concurrency-safe stage that consumes what this queue dispatches
• Adjuster Assignment Algorithms — matches the dispatched claim to a licensed specialist
• Claims Triage & Routing Engines — the control plane this queue operates inside