Watermark Alignment Strategies › Diagnosing Phantom Discrepancies in Watermark-Aligned Streams

Diagnosing Phantom Discrepancies in Watermark-Aligned Streams

This page answers one precise question: when a windowed reconciliation fires a discrepancy for a record that is present in the source but “missing” in the target, how do you decide — deterministically — whether that record was truly lost or merely arrived after its window closed? It sits under the watermark alignment strategies reference and assumes you already run a bounded-out-of-orderness watermark. A phantom discrepancy is the single most corrosive false alarm in continuous reconciliation: it looks exactly like data loss, resolves itself on the next full pass, and — if left undiagnosed — trains operators to mute the one channel that also reports real corruption.

Problem Framing

You reconcile a payments stream from PostgreSQL into a DynamoDB target with a five-second lateness bound. A dashboard fires: source has txn_9471, target does not, at event time T. By the time an operator opens the ticket the record is in the target and the discrepancy is gone. Was this a real loss that self-healed through a downstream retry, or a record that crossed the watermark late and was compared against a target that had not yet applied it? The two have opposite remediations — one demands a re-sync and an incident, the other demands widening the lateness bound — so guessing is not acceptable. The distinguishing signal is timing: a phantom’s target write timestamp is after the watermark that sealed its comparison window, while a genuine loss never lands at all.

Classifying a reported discrepancy as phantom or genuine loss A reported discrepancy enters a first decision: did the record later appear in the target? If no, it is a genuine loss routed to re-sync. If yes, a second decision asks whether the target write timestamp is after the sealing watermark. If yes, it is a phantom caused by lateness and the remedy is to widen the bound. If no, it is a genuine delayed write that still warrants investigation. yes yes no no Reported discrepancy Later present in target? Genuine loss → re-sync Target write after sealing watermark? Delayed write — investigate sink lag Phantom → widen bound
Two timing questions classify every reported discrepancy: whether the record ever reached the target, and whether it did so after the watermark that sealed its window — only the second yields a phantom.

Implementation

The classifier re-reads the target for each reported discrepancy and compares the target’s write timestamp against the watermark that sealed the original comparison. It is deliberately a separate, replayable pass so a verdict can be reproduced from the audit log rather than trusted from a live race.

python
import logging
from dataclasses import dataclass
from enum import Enum
from typing import Callable, Optional

logging.basicConfig(
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", level=logging.INFO
)
logger = logging.getLogger("recon.streaming.phantom")


class Verdict(Enum):
    PHANTOM = "phantom"            # late arrival across a sealed window
    GENUINE_LOSS = "genuine_loss"  # never landed
    DELAYED_WRITE = "delayed_write"  # landed on time but after the compare


@dataclass(frozen=True)
class Discrepancy:
    key: str
    event_time_ms: int
    sealing_watermark_ms: int      # the low watermark that closed the window


def classify(
    d: Discrepancy,
    target_write_ms: Callable[[str], Optional[int]],
) -> Verdict:
    """Classify a reported discrepancy using only timestamps, so it is reproducible."""
    landed = target_write_ms(d.key)
    if landed is None:
        logger.warning("key=%s never landed in target -> genuine loss", d.key)
        return Verdict.GENUINE_LOSS
    if landed > d.sealing_watermark_ms:
        # The target applied the row only after the window was sealed: a phantom.
        logger.info(
            "key=%s landed at %d, after watermark %d -> phantom",
            d.key, landed, d.sealing_watermark_ms,
        )
        return Verdict.PHANTOM
    logger.warning("key=%s landed at %d, before watermark -> delayed write", d.key, landed)
    return Verdict.DELAYED_WRITE

Key Implementation Notes

  • Timing is the only reliable discriminator. A value comparison cannot tell a phantom from a loss because a phantom’s value is correct once it lands. Only the relationship between the target write timestamp and the sealing watermark separates the two, which is why the sealing watermark must be persisted with every discrepancy.
  • Persist the sealing watermark, not just the verdict. The moment a discrepancy is recorded, capture the exact low watermark that closed its window. Without it, the classification is unreproducible and the audit trail cannot defend a “phantom” verdict — a requirement in regulated reconciliation.
  • A rising phantom rate is a tuning signal, not noise. Treat phantom_rate as a controller input: if it climbs, real-world lateness now exceeds the bound, and the fix is to widen the lateness window, following handling late-arriving events in reconciliation, not to raise alert thresholds.
  • Never auto-resolve a genuine loss. A phantom self-heals; a genuine loss must open an incident and trigger a targeted re-sync through the discrepancy routing and remediation path. Collapsing the two behind one “flaky, ignore” rule is exactly how silent corruption ships.

Verification

Assert the three verdicts with synthetic timings so the classifier’s boundaries are pinned. The phantom case must land strictly after the watermark; the delayed-write case strictly before.

python
wm = 1_000  # sealing watermark in ms

assert classify(Discrepancy("a", 900, wm), lambda k: None) is Verdict.GENUINE_LOSS
assert classify(Discrepancy("b", 900, wm), lambda k: 1_200) is Verdict.PHANTOM
assert classify(Discrepancy("c", 900, wm), lambda k: 950) is Verdict.DELAYED_WRITE
logger.info("phantom classifier boundaries verified")

Run this as a unit gate, then confirm in production that the measured phantom_rate correlates with observed source lateness: if phantoms cluster at a lateness just beyond the bound, the diagnosis is confirmed and the remedy is a wider window, not a re-sync.

Operational Considerations

Classification is cheap — one target point-read per discrepancy — so run it inline on every reported difference rather than in a nightly batch, and expose phantom_rate, genuine_loss_total, and lateness_p99_ms side by side so an operator sees the cause and effect in one panel. Bound the classifier’s target reads with the same connection pooling and least-privilege role the reconciler uses; a spike in discrepancies must not turn into a spike of uncontrolled point-reads against the production target. Retain every verdict with its sealing watermark in append-only audit storage, because a defensible “this was a phantom” claim during a compliance review depends on replaying the exact timing that produced it.