Alert Routing for Violations

A schema violation that never pages an engineer is indistinguishable from silent data corruption. In production JanusGraph deployments, out-of-type property writes, mixed-index lag, and backend replication drift all surface as the same thing at query time — wrong or missing results — long after the mutation that caused them has committed. This guide, part of Graph Schema Validation & Modeling Strategies, specifies how to classify those events at the write boundary and route them deterministically to the right channel, so a hard schema rejection escalates to on-call while transient index lag self-heals without waking anyone. Everything below is written for the engineer holding the pager: annotated configuration, a runnable dispatch pipeline, connection-lifecycle rules, and a symptom-to-resolution table for the failures you will actually route.

The flow below shows how a raw violation is classified by severity and fanned out to notification channels, with deduplication and bounded retry between the router and each destination.

Where Violations Originate in the Write Path

JanusGraph commits every mutation through a two-stage path, and each stage emits a distinct class of violation that must be routed differently. Understanding the seam is the difference between paging on a symptom and paging on a root cause.

• Pre-commit type enforcement. With schema.default=none and schema.constraints=true, JanusGraph rejects any write that references an unregistered property key or supplies an out-of-type value before it reaches storage. These raise synchronously, on the writer’s thread, and represent data the graph will never hold — the payload is malformed. Classify them as hard violations; they do not self-heal.
• Storage commit. Mutations land in the CQL backend (Cassandra or ScyllaDB) under a configured consistency level. A commit that fails quorum, times out, or reads stale schema during a node failure is a durability event, not a data-quality one. Route it against your Replication Strategies baseline — a single unavailable replica under LOCAL_QUORUM is a warning; loss of quorum is critical.
• Async index dispatch. After the storage commit succeeds, JanusGraph queues the mixed-index update to Elasticsearch or OpenSearch and returns. Everything downstream of this point is eventually consistent, as detailed under Eventual vs Strong Consistency. A growing index queue is a transient warning that drains on its own; a stalled queue or a mapping rejection is a hard violation that needs a reindex.

Routing correctly means never collapsing these three into one severity. A SCHEMA_VIOLATION and an INDEX_SYNC_LAG warning have opposite half-lives: the first is permanent until a human fixes the payload or the schema, the second disappears as the queue drains. The constraint boundaries that generate the pre-commit class are defined under Vertex and Edge Validation, and the index mapping rules that generate the dispatch class live in Property Indexing Rules.

Routing Configuration & Severity Tiers

Deterministic routing starts with graph-level configuration that converts silent failures into catchable signals, paired with a manifest that maps each signal to a severity tier and a set of destinations. The janusgraph.properties block below is a hardened baseline; every non-default value exists to make a violation observable rather than silent.

# janusgraph-production.properties

# --- Storage backend ---
storage.backend=cql
storage.hostname=cassandra-cluster-01.internal,cassandra-cluster-02.internal
storage.cql.keyspace=graph_prod
storage.cql.local-datacenter=dc1
storage.cql.read-consistency-level=LOCAL_QUORUM
storage.cql.write-consistency-level=LOCAL_QUORUM

# --- Schema enforcement (the source of pre-commit violations) ---
schema.default=none
schema.constraints=true

# --- Index backend (Elasticsearch value also serves OpenSearch) ---
index.search.backend=elasticsearch
index.search.hostname=es-cluster-01.internal
index.search.elasticsearch.client-only=true
index.search.elasticsearch.create.ext.number_of_replicas=1
index.search.elasticsearch.create.ext.refresh_interval=5s

Numbered operational constraints for this configuration:

1. schema.default=none is non-negotiable. It forces JanusGraph to reject unregistered property keys instead of auto-creating them. Without it, a typo becomes a permanent unindexed key and there is nothing to route — the write silently succeeds. This line is what makes hard violations exist as signals.
2. schema.constraints=true gates data types. It blocks writes whose value type does not match the registered PropertyKey, turning mixed-type corruption into a synchronous exception the router can classify as SCHEMA_VIOLATION.
3. *-consistency-level=LOCAL_QUORUM balances durability against latency and keeps schema reads consistent during single-node failure. ONE can let a validating writer read stale schema and reject a key that is in fact registered — a false-positive violation that pages on-call for nothing. Align these with your Cassandra Backend Setup; ScyllaDB cutovers should confirm the same levels per ScyllaDB Migration.
4. refresh_interval=5s sets the floor on how fast a mixed-index write becomes visible. Values below 2s under load thrash segment merges and increase apparent lag — do not shrink this to chase a lag alert.

The routing manifest maps violation signatures to tiers and channels. Keep it in version control alongside the graph config so a severity change is reviewable.

# alert-routing.yaml
alert_routing:
  severity_map:
    SCHEMA_VIOLATION:       critical   # malformed payload, never self-heals
    QUORUM_UNAVAILABLE:     critical   # storage durability at risk
    INDEX_SYNC_STALL:       high       # queue not draining
    INDEX_REBUILD_REQUIRED: high       # mapping mismatch, needs REINDEX
    INDEX_SYNC_LAG:         warning    # transient, drains on its own
    CONSISTENCY_DRIFT:      warning    # replica divergence within tolerance
  destinations:
    critical:
      - pagerduty:graph-oncall
      - slack:#graph-incidents
    high:
      - slack:#graph-ops
      - webhook:https://internal-alerts/api/v1/route
    warning:
      - slack:#graph-monitoring
  dedup_window: 300s
  retry_policy:
    max_attempts: 3
    backoff: exponential
    jitter: true

Deduplication is what keeps the manifest from becoming a pager storm. A single index outage can emit thousands of INDEX_SYNC_LAG events per second; without a dedup window, the router amplifies one incident into an unreadable flood. With dedup_window: 300s, identical (violation_type, partition) keys collapse into one notification per five minutes plus an occurrence count. The worst-case time for a critical alert to reach the pager across n bounded retries with exponential backoff and cap t_max is:

$$T_{\text{page}} \le \sum_{k=0}^{n-1} \min\bigl(2^{k}\,t_{0},\; t_{\max}\bigr) + t_{\text{dispatch}}$$

With t_0 = 2s, t_max = 10s, and n = 3, worst-case dispatch stays under 15 seconds — inside a one-minute paging SLA even if the first two attempts fail. Size max_attempts so this sum never exceeds your escalation budget; unbounded retry silently converts a critical page into a lost one.

Index Synchronization Protocol & Lag Signals

The INDEX_SYNC_LAG and INDEX_SYNC_STALL classes both come from the asynchronous dispatch stage, but they demand opposite responses, so the router must distinguish them from a metric rather than from the exception text. JanusGraph writes graph mutations to CQL first, then a background worker drains an internal queue into the search backend. Lag is bounded and self-correcting; a stall is not.

Model the safe steady state exactly as the parent modeling guide does:

$$\lambda_{\text{ingest}} \le \frac{N_{\text{threads}} \cdot B_{\text{bulk}}}{t_{\text{flush}}}$$

where $\lambda_{\text{ingest}}$ is the mutation arrival rate, $N_{\text{threads}}$ the indexing worker count, $B_{\text{bulk}}$ the bulk batch size, and $t_{\text{flush}}$ the mean flush latency of the index backend. While the inequality holds, queue depth oscillates and any lag is transient — route it as a warning. When $\lambda_{\text{ingest}}$ exceeds the right-hand side, depth grows without bound: that is a high-severity INDEX_SYNC_STALL and the fix is capacity (more threads, larger bulk-size) or backpressure, not a retry.

The polling pattern that feeds the router: sample the mixed-index queue depth on a fixed interval and classify on the derivative, not the absolute value.

• Depth below threshold, first derivative ≈ 0 → healthy, emit nothing.
• Depth above threshold, derivative ≤ 0 → transient lag, emit INDEX_SYNC_LAG (warning); it is already draining.
• Depth above threshold, derivative > 0 for N consecutive samples → INDEX_SYNC_STALL (high); the queue is not keeping up.
• Depth flat and non-zero while ingest continues → worker wedged, emit INDEX_REBUILD_REQUIRED after confirming backend reachability.

Scrape depth and flush duration from the org.janusgraph.diskstorage.indexing.IndexProvider JMX bean. The precise dispatch model, queue-depth controls, and backpressure knobs belong to External Index Synchronization & Consistency Tuning; for OpenSearch-specific drift detection see OpenSearch Sync Patterns, and for how routed queries reach the right shards see Mixed-Index Routing.

Python Routing Pipeline

The router must intercept violations at the ingestion boundary, classify each one, and dispatch it with bounded retry and a fallback path when the notification channel itself is down. The asynchronous processor below validates mutations against the registered schema contract, maps exceptions to severity tiers straight from the manifest, and never lets a failed dispatch drop a critical event silently.

import asyncio
import logging
from typing import Dict, List, Optional
import aiohttp
from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, RetryError,
)

logger = logging.getLogger("janusgraph.alert_router")

# Severity tiers mirror alert-routing.yaml exactly. Keep them in sync via CI.
SEVERITY = {
    "SCHEMA_VIOLATION": "critical",
    "QUORUM_UNAVAILABLE": "critical",
    "INDEX_SYNC_STALL": "high",
    "INDEX_REBUILD_REQUIRED": "high",
    "INDEX_SYNC_LAG": "warning",
    "CONSISTENCY_DRIFT": "warning",
}


class ViolationRouter:
    def __init__(self, webhook_url: str, dead_letter):
        self.webhook_url = webhook_url
        self.dead_letter = dead_letter          # fallback sink for un-deliverable alerts
        self.session: Optional[aiohttp.ClientSession] = None

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)),
        reraise=True,
    )
    async def _post(self, message: Dict) -> None:
        if not self.session:
            raise RuntimeError("HTTP session not initialized")
        async with self.session.post(
            self.webhook_url,
            json=message,
            timeout=aiohttp.ClientTimeout(total=5),
        ) as response:
            response.raise_for_status()

    async def dispatch(self, violation_type: str, payload: Dict) -> None:
        severity = SEVERITY.get(violation_type, "warning")
        message = {
            "severity": severity,
            "violation_type": violation_type,
            "timestamp": asyncio.get_event_loop().time(),
            "payload": payload,
        }
        try:
            await self._post(message)
            logger.info("Routed %s (%s)", violation_type, severity)
        except (RetryError, aiohttp.ClientError, asyncio.TimeoutError) as exc:
            # Never drop a critical event because the channel is down.
            logger.error("Dispatch failed for %s: %s", violation_type, exc)
            await self.dead_letter.put({**message, "dispatch_error": str(exc)})

    async def process_batch(self, mutations: List[Dict]) -> None:
        for mutation in mutations:
            try:
                self._validate_mutation(mutation)
            except ValueError as ve:
                await self.dispatch("SCHEMA_VIOLATION", {"error": str(ve), "mutation": mutation})
            except TimeoutError as te:
                await self.dispatch("INDEX_SYNC_STALL", {"error": str(te), "mutation": mutation})
            except Exception as exc:  # noqa: BLE001 - last-resort classification
                logger.exception("Unclassified pipeline error")
                await self.dispatch("CONSISTENCY_DRIFT", {"error": str(exc), "mutation": mutation})

    @staticmethod
    def _validate_mutation(mutation: Dict) -> None:
        # Mirrors schema.default=none + schema.constraints=true at the client edge,
        # so the whole batch is rejected before any write reaches storage.
        if "label" not in mutation or "properties" not in mutation:
            raise ValueError("Missing required mutation fields: label/properties")
        if not isinstance(mutation["properties"], dict):
            raise ValueError("Properties must be a dictionary")
        if not isinstance(mutation.get("id"), int):
            raise ValueError("Deterministic integer vertex id required for idempotent retry")

Two rules make this pipeline safe under load. First, validate the mutation at the client edge with the same contract the graph enforces, so a SCHEMA_VIOLATION is caught before it burns a storage round-trip — this mirrors the pre-flight checks in Vertex and Edge Validation. Second, a dispatch that exhausts its retries falls through to a dead-letter sink rather than raising and losing the event; a critical alert that fails to send must never disappear, and the same CI gate that blocks schema regressions — Schema Evolution and CI Gating — should assert that SEVERITY and alert-routing.yaml have not drifted apart.

Dispatcher Connection Lifecycle & Pool Management

The router is a long-lived service that fans out to PagerDuty, Slack, and internal webhooks, and its own connection handling determines whether alerts arrive during the incident they describe. A cold aiohttp.ClientSession per alert adds a full TLS handshake to every page; a shared, unbounded session leaks sockets when a downstream channel hangs.

Sizing and lifecycle rules:

• One session per process, opened at startup. The async with ViolationRouter(...) context owns a single ClientSession for the router’s lifetime. Reusing it keeps HTTP keep-alive connections warm so the first alert of an incident is not delayed by a handshake.
• Bound the connector pool. Configure aiohttp.TCPConnector(limit=20, limit_per_host=8, keepalive_timeout=30). The per-host cap stops a single wedged webhook from starving connections to PagerDuty; keepalive_timeout recycles idle sockets before the far side silently drops them.
• Hard per-request timeout. ClientTimeout(total=5) guarantees a dispatch cannot block a batch. A notification channel that is slower than five seconds is treated as down and routed to the dead-letter sink — an alert delivered late is worse than one delivered to fallback.
• Retry with jitter, capped. Exponential backoff (min=2, max=10) with stop_after_attempt(3) bounds worst-case dispatch to the $T_{\text{page}}$ budget above. Add jitter so a shared outage does not synchronize every worker’s retries into a thundering herd against the recovering channel.
• Separate the alert pool from the ingestion pool. The graph driver’s connection pool — sized per Connection Pooling — must not share limits with the alert dispatcher. If a storage incident saturates the ingestion pool, the router still needs live sockets to page about it.

The invariant: the alerting path must stay healthy precisely when the data path is failing. Give it dedicated, bounded, pre-warmed connections so it never competes with the workload it is monitoring.

Diagnostics & Operational Fallbacks

When a violation routes, the on-call response depends entirely on its class. The table pairs each routed signal with the command that confirms the diagnosis and the resolution — including the fallback when the first remediation does not clear it.

Two escalation rules keep the pager honest. First, never let a self-healing warning wake anyone — INDEX_SYNC_LAG and CONSISTENCY_DRIFT go to the monitoring channel, never to PagerDuty, because paging on transient lag trains on-call to ignore the pager. Second, tie the router’s dead-letter sink into the same observability stack you use for the graph, so a failed dispatch is itself a visible, alertable condition — the one failure mode that a naive router hides. For upstream references on the guarantees these signals encode, see the Apache Cassandra consistency levels documentation for quorum trade-offs and the Elasticsearch refresh API for tuning the visibility window that governs mixed-index lag. Align every threshold with observed p99 latency so routing fires on genuine regressions, not on normal ingestion peaks.

Frequently Asked Questions

Should a transient index lag ever page on-call? No. INDEX_SYNC_LAG and CONSISTENCY_DRIFT are self-healing — the mixed-index queue drains on its own and replica divergence stays within tolerance — so they route to the monitoring channel and never to PagerDuty. Paging on transient lag trains on-call to ignore the pager, which is how a genuine SCHEMA_VIOLATION gets missed.

How does the router tell INDEX_SYNC_LAG from INDEX_SYNC_STALL? From the derivative of queue depth, not the absolute value. Depth above threshold with a first derivative at or below zero is transient lag that is already draining (warning); depth rising for N consecutive samples is a stall the queue cannot clear (high), and the fix is capacity or backpressure, not a retry.

What happens if the notification channel itself is down? A dispatch that exhausts its bounded retries falls through to a dead-letter sink rather than raising and dropping the event. Wire that sink into the same observability stack as the graph so a failed dispatch is itself alertable — that is the one failure mode a naive router hides.

Why read schema at LOCAL_QUORUM instead of ONE for routing? A validating writer that reads schema at ONE can hit a replica with stale schema during a node failure and reject a property key that is in fact registered — a false-positive SCHEMA_VIOLATION that pages on-call for nothing. LOCAL_QUORUM keeps schema reads consistent so the router fires only on real violations.

Related

• Up a level: Graph Schema Validation & Modeling Strategies — the parent guide that defines the schema contract these alerts enforce.
• Vertex and Edge Validation — the constraint boundaries that generate pre-commit violations.
• Property Indexing Rules — composite vs mixed index mappings behind index-sync signals.
• Schema Evolution and CI Gating — block severity-map drift and schema regressions before deploy.
• External Index Synchronization & Consistency Tuning — the async index layer where lag and stall violations originate.