Handling Late-Arriving Usage Events with Watermarks

Usage events arrive late — from retried batches, offline clients, and regional outages — and a metering pipeline that ignores this either undercounts revenue or double-counts it. This page is a focused part of Usage Metering Event Pipelines: separating event time from processing time, tracking a per-tenant watermark, and deciding what happens to whatever arrives after the deadline.

Problem Framing

Every usage event has two timestamps that are routinely conflated. Event time is when the metered thing happened; processing time is when your pipeline saw it. Billing is defined against event time — a customer's March invoice covers what they used in March — but aggregation naturally happens in processing time, because that is the order events arrive.

The gap between them is not small. A mobile SDK buffers offline for hours. A partner integration retries a failed batch the next morning. A regional outage delays a queue by forty minutes and then delivers everything at once. Aggregate by processing time and March's usage lands in April's invoice; ignore late events entirely and you undercount; accept them forever and no period can ever be closed.

A watermark resolves this. It is a per-tenant assertion of the form "no further events with event time before W are expected", derived from observed arrival patterns. Aggregation for a window can complete once the watermark passes its end, and anything arriving after that is handled by an explicit, published policy rather than by whatever the code happens to do.

The diagonal band is normal traffic. The outlier above it is the event that decides whether your billing is correct.

Step-by-Step Guide

1. Record both timestamps, and never aggregate on the wrong one

CREATE TABLE usage_event (
  event_id     text        PRIMARY KEY,          -- client-supplied, for idempotency
  tenant_id    uuid        NOT NULL,
  meter_key    text        NOT NULL,
  quantity     numeric     NOT NULL,
  occurred_at  timestamptz NOT NULL,             -- event time: what billing is defined against
  received_at  timestamptz NOT NULL DEFAULT now(),  -- processing time: what the pipeline saw
  lateness     interval GENERATED ALWAYS AS (received_at - occurred_at) STORED
);

CREATE INDEX ON usage_event (tenant_id, occurred_at);
CREATE INDEX ON usage_event (tenant_id, received_at);

The generated lateness column costs nothing and turns "how late are events, really" from an analysis project into a query — which is the input the watermark needs.

2. Derive the watermark from observed lateness, per tenant

-- A watermark that covers 99.5% of this tenant's observed lateness over the last 30 days.
CREATE OR REPLACE VIEW tenant_watermark AS
SELECT tenant_id,
       max(occurred_at) - coalesce(
         percentile_disc(0.995) WITHIN GROUP (ORDER BY lateness),
         interval '1 hour')                       AS watermark,
       percentile_disc(0.995) WITHIN GROUP (ORDER BY lateness) AS allowance
FROM   usage_event
WHERE  received_at > now() - interval '30 days'
GROUP  BY tenant_id;

Per tenant rather than global, because lateness distributions differ enormously: a server-side integration is seconds late, a mobile SDK is hours late, and a global watermark set for the worst tenant would delay everyone's invoice.

3. Aggregate only windows the watermark has passed

# aggregate.py
def closable_windows(conn, tenant_id: str) -> list[tuple]:
    return conn.execute("""
        SELECT w.window_start, w.window_end
        FROM   billing_window w
        JOIN   tenant_watermark m USING (tenant_id)
        WHERE  w.tenant_id = %s
          AND  w.aggregated_at IS NULL
          AND  m.watermark >= w.window_end        -- the watermark has moved past this window
        ORDER  BY w.window_start
    """, (tenant_id,)).fetchall()

def aggregate(conn, tenant_id: str, start, end):
    conn.execute("""
        INSERT INTO usage_rollup (tenant_id, meter_key, window_start, window_end, quantity, event_count)
        SELECT tenant_id, meter_key, %s, %s, sum(quantity), count(*)
        FROM   usage_event
        WHERE  tenant_id = %s AND occurred_at >= %s AND occurred_at < %s
        GROUP  BY tenant_id, meter_key
        ON CONFLICT (tenant_id, meter_key, window_start) DO UPDATE
          SET quantity = EXCLUDED.quantity, event_count = EXCLUDED.event_count
    """, (start, end, tenant_id, start, end))
    conn.execute("UPDATE billing_window SET aggregated_at = now() WHERE tenant_id=%s AND window_start=%s",
                 (tenant_id, start))

The upsert makes re-aggregation safe, which matters because a window may be recomputed if an event arrives during the grace period before the period seals.

4. Define exactly one policy for post-seal arrivals

# late_policy.py
from datetime import timedelta

def route_late_event(conn, ev) -> str:
    window = find_window(conn, ev.tenant_id, ev.occurred_at)

    if window is None or window.aggregated_at is None:
        return "in_window"                    # normal path, nothing special

    if window.sealed_at is None:
        recompute(conn, window)               # inside the grace period: fix the window
        return "recomputed"

    if ev.quantity * unit_price(ev.meter_key) < MATERIALITY_MINOR:
        record_dropped(conn, ev, reason="immaterial_after_seal")
        return "dropped"                      # logged, never silently discarded

    carry_forward(conn, ev, to_window=current_window(conn, ev.tenant_id),
                  reference=window.id)        # billed next period, with a reference back
    return "carried_forward"

Carrying forward with a reference is the honest default: the customer is billed for what they used, in a period they can reconcile, and the invoice line can explain itself. Silently dropping material usage is revenue leakage; silently reopening a sealed period breaks every downstream report.

5. Publish the policy and monitor whether it is holding

-- Late-arrival profile per tenant: is the allowance still right?
SELECT tenant_id,
       count(*)                                                      AS events,
       percentile_disc(0.50) WITHIN GROUP (ORDER BY lateness)        AS p50,
       percentile_disc(0.995) WITHIN GROUP (ORDER BY lateness)       AS p995,
       max(lateness)                                                 AS worst,
       count(*) FILTER (WHERE lateness > interval '48 hours')        AS beyond_grace
FROM   usage_event
WHERE  received_at > now() - interval '7 days'
GROUP  BY tenant_id
HAVING count(*) FILTER (WHERE lateness > interval '48 hours') > 0
ORDER  BY beyond_grace DESC;

Four outcomes, all explicit. The failure to avoid is a fifth, unwritten one where the event is discarded by an exception handler.

Verification

# test_late_events.py
def test_event_inside_grace_reopens_the_rollup(pipe):
    pipe.ingest(event(tenant="acme", occurred="2026-06-30T23:50Z", qty=10))
    pipe.aggregate_window("acme", "2026-06")
    pipe.ingest(event(tenant="acme", occurred="2026-06-30T23:55Z", qty=5))   # still in grace
    assert pipe.rollup("acme", "2026-06").quantity == 15

def test_material_event_after_seal_is_carried_forward(pipe):
    pipe.seal_window("acme", "2026-06")
    pipe.ingest(event(tenant="acme", occurred="2026-06-30T23:59Z", qty=1000))
    assert pipe.rollup("acme", "2026-06").quantity == 15                     # sealed, unchanged
    line = pipe.rollup("acme", "2026-07").carried_forward[0]
    assert line.reference_window == "2026-06" and line.quantity == 1000

def test_duplicate_event_id_is_ignored(pipe):
    ev = event(tenant="acme", occurred="2026-07-01T10:00Z", qty=7, event_id="abc")
    pipe.ingest(ev); pipe.ingest(ev)
    assert pipe.rollup("acme", "2026-07").quantity == 7

Idempotency by event_id is what makes all of this safe: retries are the single largest source of late events, and a retry that double-counts is worse than one that arrives late.

One global watermark would either delay the server-integration tenant's invoice by hours or seal the mobile tenant's period before a third of its usage arrived.

Failure Modes & Gotchas

Symptom: usage appears in the wrong month. Cause: aggregation groups by received_at. Fix: aggregate by occurred_at and use received_at only to compute lateness and drive the watermark.
Symptom: an invoice changes after the customer downloaded it. Cause: late events reopen a sealed window. Fix: seal explicitly, and route post-seal arrivals to carry-forward with a reference rather than mutating history.
Symptom: one tenant's slow integration delays everyone's billing. Cause: a single global watermark set by the worst-case lateness. Fix: compute the watermark per tenant from that tenant's own distribution.
Symptom: a retry storm doubles a tenant's bill. Cause: no idempotency key, so retried batches are counted again. Fix: require a client-supplied event_id, make it the primary key, and ignore conflicts.

FAQ

How long should the grace window be? Long enough to cover the ninety-ninth-plus percentile of observed lateness for the slowest tenant you bill on that cycle, and short enough that finance can close the month — twenty-four to seventy-two hours is the usual landing zone. Derive it from data rather than choosing a round number, and re-derive it when a major integration changes, because a grace window that was right a year ago is often wrong after a new mobile client ships.

Should late events ever be dropped entirely? Only below an explicit materiality threshold, and never silently. Record every dropped event with its quantity and reason so the total can be reported and audited; if that total ever becomes material in aggregate, the threshold is wrong. Dropping without a record is indistinguishable from a pipeline bug, which is exactly the ambiguity reconciliation exists to remove — see Revenue Reconciliation & Billing Audits.

What if a customer disputes a carried-forward charge? Show them the reference. A carried-forward line that names the original window, the event count, and the reason it arrived late is a defensible answer; an unexplained line on the wrong invoice is not. This is the practical reason to carry forward with a reference rather than quietly folding the quantity into the current period's total.

Should the watermark ever move backwards? No — a watermark that regresses invalidates every window already closed behind it and makes the whole model unusable. Compute it as a monotonic maximum: the new value is the greater of the previous watermark and the newly derived one, so a quiet period or a burst of unusually late events narrows the advance rather than reversing it. If the derived value genuinely drops far below the current watermark, that is a signal about ingestion health worth alerting on, not a value to apply.

How does this interact with tenants that have almost no traffic? Low-volume tenants make percentile-based lateness unreliable, because a handful of events cannot describe a distribution. Fall back to a platform default allowance for any tenant below a minimum event count in the window, and only switch to the tenant-specific figure once there is enough data to justify it. The practical effect is that small tenants get a slightly more conservative watermark, which costs nothing — their periods close on the same schedule as everyone else's.