Reconciling PTP and NTP Event Timestamps in Python #

A backbone fibre cut fires an interface-down event on a PTP-locked optical shelf and, milliseconds later, a BGP session drop on an NTP-disciplined route reflector. If the route reflector’s NTP daemon has drifted 70 ms, the BGP event carries a wall-clock time earlier than the optical event that actually caused it. Feed that into a correlation window and the engine names BGP as the root cause, routes the ticket to the IP team, and a NOC engineer burns twenty minutes chasing a symptom while the real optical fault keeps propagating. Reversed cross-source ordering is one of the most expensive silent failures in fault management: it does not throw an error, it just quietly inflates mean time to resolution (MTTR) by pointing responders at the wrong layer. Reconciling PTP and NTP timestamps onto one monotonic ordering — with an explicit lateness budget so a slow event is placed correctly rather than appended late — is what stops that failure before correlation ever runs.

Schema Alignment and Taxonomy Anchor #

This reference is the focused implementation of the stage described in Timestamp Normalization, which itself sits inside the Core Architecture & Log Taxonomy pipeline. That parent stage owns the contract — every event arrives already decoded and schema-validated, carrying a clock_source and a raw nanosecond timestamp — and this page owns the narrower job of turning those raw, disagreeing times into a single ordering key that later stages can sort on deterministically.

Anchoring the work to that documented contract keeps it interoperable with the rest of the taxonomy. A timestamp corrected here means the same thing as one attached to a normalized trap or an RFC 5424 syslog line validated under Event Schema Design, so the corrected ordering survives all the way to Cross-Source Event Linking without re-interpretation.

Production Code: Skew Estimation and Watermarked Reordering #

The implementation reconciles the two clock disciplines in three cooperating parts: a per-source skew estimator, a watermark that trails the newest corrected time by a fixed lateness budget, and an out-of-order buffer that releases events only once the watermark has moved past them. PTP is trusted tightly and corrected only for a small, stable offset; NTP is allowed a wider, faster-adapting band because its software discipline genuinely drifts. The event contract is Pydantic V2 (ConfigDict, field_validator classmethods) so an implausible timestamp is rejected before it can poison the estimate.

import asyncio
import heapq
import time
from enum import Enum
from pydantic import BaseModel, ConfigDict, Field, field_validator


class ClockSource(str, Enum):
    PTP = "ptp"
    NTP = "ntp"


class Event(BaseModel):
    model_config = ConfigDict(strict=True, frozen=True, extra="forbid")

    event_id: str
    ne_id: str
    clock_source: ClockSource
    raw_ts_ns: int = Field(gt=0)

    @field_validator("raw_ts_ns")
    @classmethod
    def _plausible(cls, v: int) -> int:
        # A 1970 reset or a default clock is corruption, not a late event.
        if abs(v - time.time_ns()) > 86_400_000_000_000:  # 24h in ns
            raise ValueError(f"implausible epoch: {v}")
        return v


# Per-source smoothing: PTP hardware time barely moves; NTP software time drifts.
_ALPHA = {ClockSource.PTP: 0.02, ClockSource.NTP: 0.15}


class SkewEstimator:
    """Exponentially-decaying offset estimate keyed by source and element."""

    def __init__(self) -> None:
        self._offset_ns: dict[str, float] = {}

    def update(self, ev: Event, reference_ns: int) -> None:
        key = f"{ev.clock_source.value}:{ev.ne_id}"
        sample = float(ev.raw_ts_ns - reference_ns)
        prior = self._offset_ns.get(key, sample)
        alpha = _ALPHA[ev.clock_source]
        self._offset_ns[key] = (1 - alpha) * prior + alpha * sample

    def corrected_ns(self, ev: Event) -> int:
        key = f"{ev.clock_source.value}:{ev.ne_id}"
        return int(ev.raw_ts_ns - self._offset_ns.get(key, 0.0))


class WatermarkReorderer:
    """Buffers corrected events and releases them in monotonic order.

    The watermark trails the highest corrected timestamp by lateness_budget.
    An event at or behind the watermark is safe to release; one arriving
    behind it afterwards is late and is reconciled, never reordered in place.
    """

    def __init__(self, out: asyncio.Queue, late: asyncio.Queue,
                 lateness_budget_ms: int = 500) -> None:
        self._out = out
        self._late = late
        self._heap: list[tuple[int, str]] = []
        self._budget_ns = lateness_budget_ms * 1_000_000
        self._high_ns = 0
        self._watermark_ns = 0

    async def add(self, corrected_ns: int, event_id: str) -> None:
        if corrected_ns < self._watermark_ns:
            await self._late.put((corrected_ns, event_id))
            return
        heapq.heappush(self._heap, (corrected_ns, event_id))
        self._high_ns = max(self._high_ns, corrected_ns)
        self._watermark_ns = self._high_ns - self._budget_ns
        while self._heap and self._heap[0][0] <= self._watermark_ns:
            ready_ts, ready_id = heapq.heappop(self._heap)
            await self._out.put((ready_ts, ready_id))

    async def flush(self) -> None:
        # Drain remaining events at shutdown or on an idle-source timeout.
        while self._heap:
            ready_ts, ready_id = heapq.heappop(self._heap)
            await self._out.put((ready_ts, ready_id))


class Reconciler:
    def __init__(self, out: asyncio.Queue, late: asyncio.Queue,
                 lateness_budget_ms: int = 500) -> None:
        self._skew = SkewEstimator()
        self._reorder = WatermarkReorderer(out, late, lateness_budget_ms)

    async def handle(self, ev: Event, reference_ns: int) -> None:
        self._skew.update(ev, reference_ns)
        corrected = self._skew.corrected_ns(ev)
        await self._reorder.add(corrected, ev.event_id)

The reference clock passed as reference_ns is a vetted source of truth — a PTP grandmaster capture or a hardened NTP peer — not the event’s own device. That is the whole point: the estimator measures how far each source sits from a trusted anchor and cancels it, so a drifting NTP daemon corrects toward the same timeline a PTP shelf already reports.

Async Ingestion Hook #

Skew estimation and heap operations are O(1) and O(log n) respectively, so the reconciler never needs a thread. It slots into the async pipeline as a coroutine that drains an input queue and feeds an ordered output queue, applying backpressure naturally when the downstream correlation stage is busy.

import asyncio
import logging

logger = logging.getLogger("timestamp_reconcile")


async def reconcile_worker(in_q: asyncio.Queue, out_q: asyncio.Queue,
                           late_q: asyncio.Queue, reference_clock) -> None:
    """Drain raw events, emit a monotonically ordered stream."""
    reconciler = Reconciler(out_q, late_q, lateness_budget_ms=500)
    while True:
        raw = await in_q.get()          # non-blocking wait on the bounded queue
        try:
            ev = Event.model_validate(raw)
        except Exception as exc:         # ValidationError -> quarantine, don't crash
            logger.warning("timestamp reject", extra={"raw": raw, "err": str(exc)})
            in_q.task_done()
            continue
        # A trusted, monotonic reference read; never the device's self-report.
        await reconciler.handle(ev, reference_clock.now_ns())
        in_q.task_done()

Because the only awaited calls are queue operations, a slow correlation consumer simply slows the out_q.put, which slows this worker, which lets the bounded input queue fill — end-to-end backpressure without dropping telemetry. An idle-source timer (not shown for brevity) should call flush() when the slowest source falls silent so a dead sensor cannot freeze the watermark.

Mitigation and Hardening #

When ordering cannot be established cleanly, the system must degrade deterministically rather than guess. These are the concrete failure paths a production deployment should implement:

1. Late-lane reconciliation, never silent drop. An event arriving behind the watermark is pushed to late_q with its corrected time and the window it missed, then handed to correlation as a supplementary signal that can reinforce or reopen an incident. Alert when the late-event ratio exceeds 1% over five minutes — that is the signal your lateness budget is too tight or a source has lost sync.
2. Plausibility rejection at the edge. A timestamp more than 24 hours from now is a default clock, an epoch reset, or a rollover, not a late event. Rejecting it in the validator keeps a single corrupt packet from dragging the skew estimate, and the rejection lands in the dead-letter queue as a measurable signal rather than silent corruption.
3. Bounded skew adaptation. The exponential smoothing coefficient caps how far one sample can move an offset, so a lone bad reading cannot swing the estimate. PTP’s tiny coefficient makes its estimate effectively rigid; only a sustained series of samples can move it, which is exactly the behaviour a hardware clock warrants.
4. Grandmaster-loss degradation. If the trusted reference disappears, freeze each affected source’s last-known offset and tag corrected events degraded_clock so downstream correlation can down-weight cross-source links involving that element until the reference recovers, rather than re-estimating against an untrusted peer.

Operational Hardening Notes #

Tuning this pattern is about balancing release latency against completeness, and the lateness budget is the single knob that governs it. A 500ms budget absorbs almost all real out-of-order arrival on a healthy carrier network while adding only half a second of hold latency; drop it to 150ms for latency-critical P1 correlation and accept a higher late-lane ratio, or raise it to 2000ms where NTP sources are bursty. Keep the reorder heap bounded with a per-partition high-water mark (50,000 events is a sound start) and shed on overflow by force-advancing the watermark, so a storm cannot exhaust memory. Partition by ne_id so each worker maintains its own small heap and skew table, keeping per-event cost flat and the p99 normalization latency under 2 ms even under alarm-storm load. Integrated this way, timestamp reconciliation becomes one deterministic gate that holds the same sub-2-second SLA ticket-creation budget as the rest of the taxonomy.

Frequently Asked Questions #

Why estimate skew instead of just converting everything to UTC? #

UTC conversion fixes timezone and format, but it does not fix a wrong clock. An NTP daemon can report a confident UTC time that is 70 ms off, and converting it changes nothing. Skew estimation measures how far each source sits from a trusted reference and subtracts that offset, so the corrected UTC instants actually agree instead of merely sharing a timezone.

How large a lateness budget should I use? #

Start at 500 milliseconds, which absorbs almost all real out-of-order arrival on a healthy carrier network while adding only half a second of hold latency. Tighten toward 150 milliseconds for latency-critical P1 correlation and accept a higher late-event ratio, or widen toward 2000 milliseconds where NTP sources are bursty. Watch the late-event ratio: if it climbs above 1 percent, the budget is too tight for current conditions.

Will a single bad timestamp corrupt the whole timeline? #

No, two guards prevent it. The plausibility validator rejects any timestamp more than 24 hours from now before it reaches the estimator, and the exponential smoothing coefficient bounds how far any one sample can move an offset. PTP uses a very small coefficient, so its estimate is effectively rigid and one bad packet cannot swing it.

What happens to an event that arrives after its window closed? #

It is sent to the late lane, corrected and tagged with the window it missed, then handed to correlation as a supplementary signal that can reinforce or reopen an incident. It is never dropped and never inserted out of order into a stream that has already advanced, so late arrivals stay honest without rewriting emitted ordering.

Related #

• Up to: Timestamp Normalization — the stage that owns skew correction and watermark policy
• Core Architecture & Log Taxonomy — the reference pipeline this correction feeds
• Event Schema Design — the canonical event contract every time field relies on
• Cross-Source Event Linking — the downstream consumer that needs correct cross-source ordering
• Temporal Windowing Strategies — how corrected timelines feed sliding and session windows