Streaming Late-Event Repair For Clickstream Using Sql Temporal Windows

Why I got obsessed with “late events” in clickstreams

I built a real-time clickstream pipeline for a dashboard that updates every few seconds. Everything looked fine—until I noticed that some page-view counts briefly “jumped backward” minutes later.

What was happening: user events (like page views) don’t always arrive in order. Network delays, retries, and batching can cause late events—events with an event timestamp that fall “in the past” compared to when they show up in the stream. If you’re computing metrics in real time without accounting for that, you end up with wrong counts for recent windows.

In this post I’ll show the exact pattern I implemented: temporal windows with late-event repair—using SQL to re-aggregate the last N minutes every time new events arrive. I’ll keep it concrete with a small working example.

The setup: what we compute and what “late repair” means

I wanted a table that always has correct counts for a moving time window, even when events arrive late.

Incoming events have:
- user_id
- event_time (the timestamp the event actually happened)
- received_at (when the event hit the pipeline)
- event_name (like page_view)
Dashboard metric:
- page_views_per_10_minutes grouped by window_start

Late-event repair strategy

Every time events arrive, I recompute aggregates for a recent span that covers expected lateness.

If I expect events to be late by up to 5 minutes, and my dashboard uses 10-minute windows, then recomputing the last 15 minutes is usually enough:

last 10 minutes for the current window and one previous one
plus 5 minutes of “repair horizon”

In practice, the repair horizon should be based on observed data observability metrics (like a histogram of event-time lag).

A working example with SQLite (yes, really)

SQLite isn’t a streaming engine, but it’s perfect for demonstrating the logic. I’ll emulate streaming by inserting rows and re-running the SQL “materialization” step.

1) Create sample data tables

-- schema.sql (SQLite)
CREATE TABLE clickstream_events (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  user_id TEXT NOT NULL,
  event_name TEXT NOT NULL,
  event_time TEXT NOT NULL,   -- ISO-8601 timestamp
  received_at TEXT NOT NULL  -- ISO-8601 timestamp
);

-- This is our "materialized view" table:
-- it stores final aggregates per window.
CREATE TABLE page_view_window_agg (
  window_start TEXT NOT NULL,
  window_end TEXT NOT NULL,
  user_id TEXT NOT NULL,
  page_views INTEGER NOT NULL,
  PRIMARY KEY (window_start, window_end, user_id)
);

2) Seed events including late arrivals

I’ll define a single user with multiple page views. Some arrive late.

-- seed.sql (SQLite)
INSERT INTO clickstream_events (user_id, event_name, event_time, received_at) VALUES
('u1','page_view','2026-08-03T10:00:30','2026-08-03T10:00:30'), -- on time
('u1','page_view','2026-08-03T10:05:10','2026-08-03T10:05:10'), -- on time
('u1','page_view','2026-08-03T10:07:40','2026-08-03T10:20:00'); -- LATE (event_time is recent, but arrives much later)

That last one is the whole problem:

event_time = 10:07:40 (within a 10:00–10:09:59 window)
received_at = 10:20:00 (way later)

If you only aggregate based on received order, you’d miss that window count until the late event eventually shows up.

The core SQL: compute 10-minute windows + repair the last 15 minutes

How I do 10-minute bucketing in SQL

For each event_time, I compute:

window_start = floor to 10-minute boundary
window_end = start + 10 minutes

In SQLite, I can do this by converting timestamps to epoch seconds and bucketing.

1) A “repair” query

This query recomputes aggregates for a set of windows determined by a repair horizon.

Window size: 10 minutes
Late allowance: 5 minutes
Repair horizon: last 15 minutes relative to current processing time

Parameters I’m using

:now = current processing time (in ISO string)
:window_minutes = 10
:lateness_minutes = 5
:repair_horizon_minutes = 15 (10 + 5)

-- repair.sql (SQLite)
-- Assumes you have these parameters bound:
-- :now, :window_minutes, :lateness_minutes, :repair_horizon_minutes

WITH params AS (
  SELECT
    CAST(strftime('%s', :now) AS INTEGER) AS now_epoch,
    CAST(:window_minutes AS INTEGER) AS wmin,
    CAST(:lateness_minutes AS INTEGER) AS lmin,
    CAST(:repair_horizon_minutes AS INTEGER) AS rhmin
),
events_in_scope AS (
  -- We only need to recompute windows that could be affected
  -- by late events arriving up to now, within the repair horizon.
  SELECT
    e.user_id,
    e.event_name,
    e.event_time,
    CAST(strftime('%s', e.event_time) AS INTEGER) AS event_epoch
  FROM clickstream_events e
  JOIN params p
    ON 1=1
  WHERE e.event_name = 'page_view'
    AND CAST(strftime('%s', e.event_time) AS INTEGER)
        >= (p.now_epoch - (p.rhmin * 60))
),
bucketed AS (
  SELECT
    user_id,
    -- compute window start epoch: floor(event_epoch / (wmin*60)) * (wmin*60)
    (event_epoch / (p.wmin * 60)) * (p.wmin * 60) AS window_start_epoch,
    (event_epoch / (p.wmin * 60)) * (p.wmin * 60) + (p.wmin * 60) AS window_end_epoch
  FROM events_in_scope
  JOIN params p ON 1=1
),
agg AS (
  SELECT
    b.window_start_epoch,
    b.window_end_epoch,
    b.user_id,
    COUNT(*) AS page_views
  FROM bucketed b
  GROUP BY b.window_start_epoch, b.window_end_epoch, b.user_id
)
SELECT
  datetime(a.window_start_epoch, 'unixepoch') AS window_start,
  datetime(a.window_end_epoch, 'unixepoch') AS window_end,
  a.user_id,
  a.page_views
FROM agg a
ORDER BY window_start, user_id;

2) Turning that into an actual repair (delete + insert)

The repair step needs to:

delete existing aggregates for windows in scope
insert newly computed aggregates

-- materialize_repair.sql (SQLite)
-- This script assumes the same parameters:
-- :now, :window_minutes, :lateness_minutes, :repair_horizon_minutes

WITH params AS (
  SELECT
    CAST(strftime('%s', :now) AS INTEGER) AS now_epoch,
    CAST(:window_minutes AS INTEGER) AS wmin,
    CAST(:repair_horizon_minutes AS INTEGER) AS rhmin
),
windows_in_scope AS (
  -- Determine all window starts that fall within the repair horizon.
  -- We generate potential window starts by bucketing the lower bound.
  SELECT
    ((p.now_epoch - (p.rhmin * 60)) / (p.wmin * 60)) * (p.wmin * 60) AS scope_start_epoch,
    p.now_epoch AS now_epoch,
    p.wmin AS wmin
  FROM params p
),
candidate_windows AS (
  SELECT
    (scope_start_epoch + k.n * (wmin * 60)) AS window_start_epoch,
    (scope_start_epoch + (k.n + 1) * (wmin * 60)) AS window_end_epoch
  FROM windows_in_scope ws
  JOIN (
    -- generate a small range of k values; enough to cover scope
    -- Here: 0..10. For real systems, generate based on time difference.
    SELECT 0 AS n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 UNION ALL
    SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7 UNION ALL
    SELECT 8 UNION ALL SELECT 9 UNION ALL SELECT 10
  ) k
  ON 1=1
  WHERE (scope_start_epoch + k.n * (wmin * 60)) < ws.now_epoch
),
delete_scope AS (
  SELECT
    datetime(c.window_start_epoch, 'unixepoch') AS window_start,
    datetime(c.window_end_epoch, 'unixepoch') AS window_end
  FROM candidate_windows c
)
DELETE FROM page_view_window_agg
WHERE (window_start, window_end) IN (
  SELECT window_start, window_end FROM delete_scope
);

-- Insert fresh aggregates for the same scope:
INSERT INTO page_view_window_agg (window_start, window_end, user_id, page_views)
WITH params AS (
  SELECT
    CAST(strftime('%s', :now) AS INTEGER) AS now_epoch,
    CAST(:window_minutes AS INTEGER) AS wmin,
    CAST(:lateness_minutes AS INTEGER) AS lmin,
    CAST(:repair_horizon_minutes AS INTEGER) AS rhmin
),
events_in_scope AS (
  SELECT
    e.user_id,
    e.event_name,
    e.event_time,
    CAST(strftime('%s', e.event_time) AS INTEGER) AS event_epoch
  FROM clickstream_events e
  JOIN params p
    ON 1=1
  WHERE e.event_name = 'page_view'
    AND CAST(strftime('%s', e.event_time) AS INTEGER)
        >= (p.now_epoch - (p.rhmin * 60))
),
bucketed AS (
  SELECT
    user_id,
    (event_epoch / (p.wmin * 60)) * (p.wmin * 60) AS window_start_epoch,
    (event_epoch / (p.wmin * 60)) * (p.wmin * 60) + (p.wmin * 60) AS window_end_epoch
  FROM events_in_scope
  JOIN params p ON 1=1
),
agg AS (
  SELECT
    b.window_start_epoch,
    b.window_end_epoch,
    b.user_id,
    COUNT(*) AS page_views
  FROM bucketed b
  GROUP BY b.window_start_epoch, b.window_end_epoch, b.user_id
)
SELECT
  datetime(a.window_start_epoch, 'unixepoch') AS window_start,
  datetime(a.window_end_epoch, 'unixepoch') AS window_end,
  a.user_id,
  a.page_views
FROM agg a;

3) What I observed when running it

If I run repair at :now = '2026-08-03T10:10:00':

The late event (event_time at 10:07:40) is already in the table in this demo, so it still gets counted.
In a real pipeline, the late event would not exist yet at 10:10; it arrives later, then a subsequent repair run re-computes the 10:00 window and updates the dashboard.

That’s the key property: repair runs are idempotent for windows in scope because I delete and re-insert deterministic aggregates.

Step-by-step: simulate “arrives late” and repair twice

Below is a simple sequence you could run from a SQLite client (or adapt to a small script).

First materialization (before late event exists)

Imagine the late event hasn’t arrived yet. I’ll simulate by deleting it temporarily.

-- Remove the late row to simulate an earlier point in time.
DELETE FROM clickstream_events
WHERE event_time = '2026-08-03T10:07:40';

Now repair at 10:10:00:

-- Bind:
-- :now='2026-08-03T10:10:00'
-- :window_minutes=10
-- :lateness_minutes=5
-- :repair_horizon_minutes=15

-- Run materialize_repair.sql here

Inspect results:

SELECT * FROM page_view_window_agg ORDER BY window_start, user_id;

Expected outcome:

The 10:00 window should have 2 page views (10:00:30 and 10:05:10)

Late event arrives

Now insert the late event and repair again at 10:20:00:

INSERT INTO clickstream_events (user_id, event_name, event_time, received_at) VALUES
('u1','page_view','2026-08-03T10:07:40','2026-08-03T10:20:00');

Run repair with:

:now='2026-08-03T10:20:00'
:window_minutes=10
:lateness_minutes=5
:repair_horizon_minutes=15

Inspect results again:

SELECT * FROM page_view_window_agg ORDER BY window_start, user_id;

Expected outcome:

The 10:00 window updates to 3 page views.
The key is that we didn’t try to “incrementally add” the late event to an old aggregation. We recomputed the window for the repair horizon, which guarantees correctness.

Data observability tie-in: choosing the repair horizon from real lag

The only way this approach breaks is if the lateness exceeds the repair horizon. That’s why I tied it to observability:

I tracked event-time lag = received_at - event_time
I computed percentiles (p50/p95/p99)
I set lateness_minutes to a value that covered, for example, p99 of lag (with a safety margin)

That way, late-event repair behaves like a controlled tradeoff:

larger horizon = more reprocessing cost
smaller horizon = higher risk of incorrect aggregates

Closing thoughts

I learned that late-event handling isn’t something I should “hope is rare”—it needs a deterministic mechanism. The late-event repair pattern (re-aggregating SQL windows over a measured repair horizon, using delete-and-reinsert for idempotence) turned my clickstream dashboard from “sometimes wrong” into “consistently correct,” even when events arrived out of order.