Add live top requests table and request detail overlay

README.md

@@ -21,13 +21,21 @@ Bulma and Chart.js are vendored under `app/static/vendor`, so the dashboard does
 Add the dedicated log (the application does not change Nginx configuration):
 
 ```nginx
-log_format traffic '$msec $host $request_method $status '
+log_format traffic_json escape=json
-                   '$request_length $bytes_sent '
+    '{"timestamp":"$msec","host":"$host",'
-                   '$request_time $upstream_response_time';
+    '"method":"$request_method","status":"$status",'
+    '"incoming":"$request_length","outgoing":"$bytes_sent",'
+    '"request_time":"$request_time",'
+    '"upstream_response_time":"$upstream_response_time",'
+    '"client_address":"$remote_addr","path":"$uri"}';
 
-access_log /var/log/nginx-traffic/traffic.log traffic;
+access_log /var/log/nginx-traffic/traffic.log traffic_json;
 ```
 
+For an existing installation, deploy the updated dashboard first, then replace the dedicated log format and its `access_log` directive, run `sudo nginx -t`, and reload with `sudo systemctl reload nginx`. The parser supports mixed JSON and original whitespace records during migration. Original records still contribute to the table but display “Unavailable” for client address and path. No historical logs are replayed.
+
+The client address is Nginx's `$remote_addr`. Behind a proxy, it may be the proxy address unless trusted real-IP handling is already configured in Nginx. The path is normalized `$uri`, which can reflect rewrites; query parameters, headers, and request bodies are not captured. All JSON values in this format are strings, including numeric fields.
+
 The follower tolerates malformed lines, `-` upstream times, multiple upstream values, temporary disappearance, and rename/replacement rotation. It starts at the end of an existing file, like `tail -F`, so old log history is not mistaken for live traffic. The application keeps only configured rolling buckets (and bounded latency samples); totals are from application startup.
 
 ## Running
@@ -54,6 +62,16 @@ TRAFFIC_LOG_PATH=/var/log/nginx-traffic/traffic.log LISTEN_ADDRESS=0.0.0.0 LISTE
 
 The visible window can be set to 1 minute, 5 minutes, 15 minutes, or 1 hour. A new browser receives enough current history to fill its selected graph window. Host names are discovered automatically and can be selected for breakout graphs. Theme selection (system, light, dark) is stored in browser local storage.
 
+## Top 10 Requests
+
+The table updates every second and follows the selected chart window. Each completed request appears once, ranked by the larger of its inbound request size and outbound response size. Its arrow and size indicate that direction; ties use inbound. Equal sizes sort by newest completion timestamp and then newest ingestion ID. Sizes use decimal units, with exact byte counts on hover.
+
+Click a row (or focus it and press Enter/Space) to open request details, including both byte counts. Details are a frozen snapshot: table updates, expiration, window changes, and reconnections do not change an open overlay. Close it with the close button, Escape, or the backdrop. “Copy all details” copies labeled text including exact bytes; if clipboard access is unavailable, selectable text appears for manual copying.
+
+Requests appear only after Nginx writes their completion log entry. This is not a view of uploads/downloads currently in progress. Request sizes include the bytes counted by `$request_length` and `$bytes_sent`; existing throughput metrics still attribute those bytes to the completion second.
+
+Request tracking retains at most ten records per second for up to one hour, limited by configured retention; restarts clear it. Both `/api/telemetry` and `/api/events` include `top_requests_by_window`, mapping `"60"`, `"300"`, `"900"`, and `"3600"` to up to ten full request records. Each record contains a process-local `id`, `timestamp`, `host`, `method`, `client_address`, `path`, `status`, `incoming`, `outgoing`, `request_time`, `upstream_time` (average), `direction`, and `size` (larger byte count). Missing client address, path, or upstream timing is `null`.
+
 ## tmpfs and permissions
 
 One possible tmpfs setup is:

app/main.py

@@ -8,7 +8,7 @@ import os
 import statistics
 import threading
 import time
-from collections import deque
+from heapq import nlargest
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, AsyncIterator
@@ -58,8 +58,23 @@ def percentile(values: list[float], p: float) -> float | None:
 
 
 def parse_line(line: str) -> dict[str, Any] | None:
-    """Parse the configured, whitespace-delimited Nginx traffic format."""
+    """Parse JSON traffic records or the original whitespace format."""
-    fields = line.strip().split()
+    line = line.strip()
+    if line.startswith("{"):
+        try:
+            record = json.loads(line)
+            keys = ("timestamp", "host", "method", "status", "incoming", "outgoing", "request_time")
+            if not all(isinstance(record[key], str) for key in (*keys, "client_address", "path", "upstream_response_time")):
+                return None
+            fields = [record[key] for key in keys]
+            fields.append(record["upstream_response_time"])
+            client_address = record["client_address"] or None
+            path = record["path"].split("?", 1)[0] or None
+        except (ValueError, KeyError, TypeError):
+            return None
+    else:
+        fields = line.split()
+        client_address = path = None
     # A trailing empty $upstream_response_time disappears when splitting on
     # whitespace, so seven fields is valid and means "no upstream timing".
     if len(fields) < 7:
@@ -82,6 +97,9 @@ def parse_line(line: str) -> dict[str, Any] | None:
     return {
         "timestamp": timestamp,
         "host": fields[1] or "(unknown)",
+        "method": fields[2],
+        "client_address": client_address,
+        "path": path,
         "incoming": request_length,
         "outgoing": bytes_sent,
         "status": status,
@@ -100,6 +118,27 @@ class Telemetry:
         self.total_outgoing = 0
         self.lock = threading.RLock()
         self.subscribers: set[asyncio.Queue[str]] = set()
+        self.request_buckets: dict[int, list[dict[str, Any]]] = {}
+        self.next_request_id = 0
+
+    @staticmethod
+    def _request_rank(record: dict[str, Any]) -> tuple[int, float, int]:
+        return record["size"], record["timestamp"], record["id"]
+
+    def _prune_requests(self, now: int) -> None:
+        cutoff = now - min(self.retention, 3600) + 1
+        for second in [second for second in self.request_buckets if second < cutoff]:
+            del self.request_buckets[second]
+
+    def _top_requests(self, now: int) -> dict[str, list[dict[str, Any]]]:
+        self._prune_requests(now)
+        return {
+            str(window): nlargest(10, (
+                record for second, records in self.request_buckets.items()
+                if now - window + 1 <= second <= now for record in records
+            ), key=self._request_rank)
+            for window in (60, 300, 900, 3600)
+        }
 
     def ingest(self, event: dict[str, Any]) -> None:
         now = int(time.time())
@@ -110,6 +149,15 @@ class Telemetry:
             return
         second = min(second, now)
         with self.lock:
+            self._prune_requests(now)
+            self.next_request_id += 1
+            record = dict(event, id=self.next_request_id,
+                          direction="inbound" if event["incoming"] >= event["outgoing"] else "outbound",
+                          size=max(event["incoming"], event["outgoing"]))
+            if second >= now - min(self.retention, 3600) + 1:
+                records = self.request_buckets.setdefault(second, [])
+                records.append(record)
+                self.request_buckets[second] = nlargest(10, records, key=self._request_rank)
             self._add(self.global_series.bucket(second), event)
             host = event["host"]
             if host not in self.hosts:
@@ -162,7 +210,8 @@ class Telemetry:
             current["total_outgoing"] = self.total_outgoing
             current["active_hosts"] = len(active)
             current["busiest_host"] = max(active, key=lambda host: sum(p["incoming"] + p["outgoing"] for p in host_points[host][-60:]), default=None)
-            return {"type": "telemetry", "now": now, "points": points, "hosts": hosts, "host_points": host_points, "summary": current}
+            return {"type": "telemetry", "now": now, "points": points, "hosts": hosts, "host_points": host_points, "summary": current,
+                    "top_requests_by_window": self._top_requests(now)}
 
     @staticmethod
     def _point(bucket: Bucket | None, second: int) -> dict[str, Any]:

app/static/index.html

@@ -14,6 +14,16 @@
     .metric-label { color: var(--bulma-text-weak); font-size: .75rem; text-transform: uppercase; letter-spacing: .05em; }
     .host-list { max-height: 185px; overflow-y: auto; }
     .legend-dot { width: .65rem; height: .65rem; display: inline-block; border-radius: 50%; margin-right: .4rem; }
+    #request-rows tr { cursor: pointer; }
+    #request-rows tr:focus-visible { outline: 2px solid var(--bulma-link); outline-offset: -2px; }
+    #request-dialog { width: min(640px, calc(100% - 2rem)); max-height: calc(100vh - 2rem); overflow: auto; border: 0; border-radius: 12px; padding: 1.5rem; color: var(--bulma-text); background: var(--bulma-scheme-main); }
+    #request-dialog::backdrop { background: rgb(0 0 0 / .55); }
+    #request-details { display: grid; grid-template-columns: max-content minmax(0, 1fr); gap: .65rem 1.25rem; }
+    #request-details dt { font-weight: 600; }
+    #request-details dd { overflow-wrap: anywhere; }
+    #request-copy-text { width: 100%; min-height: 220px; }
+    #request-copy-text[hidden] { display: none; }
+    .request-size { white-space: nowrap; }
     @media (max-width: 768px) { .chart-wrap { height: 280px; } }
   </style>
 </head>
@@ -29,8 +39,24 @@
   </div>
   <div class="columns is-multiline mb-2" id="metrics"></div>
   <div class="card mb-5"><div class="card-content"><div class="level mb-3"><div class="level-left"><h2 class="title is-5 mb-0">Throughput</h2></div><div class="level-right"><span class="tag is-light" id="updated">Waiting for data</span></div></div><div class="chart-wrap"><canvas id="traffic-chart"></canvas></div></div></div>
+  <div class="card mb-5"><div class="card-content">
+    <h2 class="title is-5 mb-2" id="top-requests-title" tabindex="-1">Top 10 Requests</h2>
+    <p class="help mb-4">Completed requests in the selected window, ranked by their larger inbound or outbound size. Select a row for details.</p>
+    <div class="table-container"><table class="table is-fullwidth is-hoverable" aria-labelledby="top-requests-title">
+      <thead><tr><th scope="col">Rank</th><th scope="col">Direction</th><th scope="col">Host</th><th scope="col">Client address</th><th scope="col">Size</th></tr></thead>
+      <tbody id="request-rows"></tbody>
+    </table></div>
+    <p id="requests-empty" class="has-text-grey">No completed requests in this window.</p>
+  </div></div>
   <div class="card"><div class="card-content"><div class="columns"><div class="column is-one-quarter"><h2 class="title is-5">Host breakouts</h2><p class="help mb-3">Select hosts to compare their recent traffic.</p><div id="hosts" class="host-list"></div></div><div class="column"><div class="chart-wrap"><canvas id="host-chart"></canvas></div></div></div></div></div>
 </div></main>
+<dialog id="request-dialog" aria-labelledby="request-dialog-title">
+  <div class="is-flex is-justify-content-space-between is-align-items-center mb-4"><h2 class="title is-5 mb-0" id="request-dialog-title">Request details</h2><button class="delete" id="request-close" aria-label="Close request details" autofocus></button></div>
+  <dl id="request-details" class="mb-5"></dl>
+  <button class="button is-link" id="request-copy">Copy all details</button>
+  <p id="request-copy-status" class="help mt-2" role="status"></p>
+  <textarea id="request-copy-text" class="textarea mt-3" aria-label="Request details to copy manually" readonly hidden></textarea>
+</dialog>
 <script>
 window.addEventListener('error', event => {
   const box = document.getElementById('frontend-error');
@@ -49,7 +75,7 @@ function bytesPerSecond(value) { const bits = value * 8; if (bits >= 1e9) return
 function duration(v) { return v == null ? '—' : v.toFixed(3)+' s'; }
 function applyTheme(value) { const actual = value === 'system' ? (matchMedia('(prefers-color-scheme: dark)').matches ? 'dark' : 'light') : value; document.documentElement.dataset.theme = actual; document.documentElement.style.colorScheme = actual; if (typeof trafficChart !== 'undefined') { trafficChart.options.scales.x.grid.color = getComputedStyle(document.documentElement).getPropertyValue('--chart-grid'); hostChart.options.scales.x.grid.color = trafficChart.options.scales.x.grid.color; trafficChart.update(); hostChart.update(); } }
 function metric(label, value) { return `<div class="column is-half-mobile is-one-quarter-tablet"><div class="card"><div class="card-content py-4"><div class="metric-label">${label}</div><div class="metric-value">${value}</div></div></div></div>`; }
-function renderMetrics(s) { const status = s.status.reduce((a,b)=>a+b,0); const errors = status ? ((s.status[2]+s.status[3])/status*100).toFixed(1)+'%' : '0.0%'; $('metrics').innerHTML = metric('Inbound', bytesPerSecond(s.incoming))+metric('Outbound', bytesPerSecond(s.outgoing))+metric('Requests', s.requests.toFixed(1)+'/sec')+metric('Transferred', formatBytes(s.total_incoming+s.total_outgoing))+metric('Avg latency', duration(s.latency))+metric('p95 latency', duration(s.p95))+metric('Upstream avg', duration(s.upstream))+metric('Errors (4xx/5xx)', errors)+metric('Active hosts', s.active_hosts)+metric('Busiest host', s.busiest_host || '—'); }
+function renderMetrics(s) { const status = s.status.reduce((a,b)=>a+b,0); const errors = status ? ((s.status[2]+s.status[3])/status*100).toFixed(1)+'%' : '0.0%'; $('metrics').innerHTML = metric('Inbound', bytesPerSecond(s.incoming))+metric('Outbound', bytesPerSecond(s.outgoing))+metric('Requests', s.requests.toFixed(1)+'/sec')+metric('Transferred', `<span title="${s.total_incoming+s.total_outgoing} bytes">${formatBytes(s.total_incoming+s.total_outgoing)}</span>`)+metric('Avg latency', duration(s.latency))+metric('p95 latency', duration(s.p95))+metric('Upstream avg', duration(s.upstream))+metric('Errors (4xx/5xx)', errors)+metric('Active hosts', s.active_hosts)+metric('Busiest host', escapeHtml(s.busiest_host || '—')); }
 function formatBytes(n) { const units=['B','KB','MB','GB','TB']; let i=0; while(n>=1000 && i<units.length-1){n/=1000;i++;} return (i? n.toFixed(1):Math.round(n))+' '+units[i]; }
 function makeChart(id, datasets) { return new Chart($(id), {type:'line', data:{labels:[],datasets}, options:{animation:false, maintainAspectRatio:false, interaction:{mode:'index',intersect:false}, scales:{x:{grid:{color:getComputedStyle(document.documentElement).getPropertyValue('--chart-grid')},ticks:{maxTicksLimit:8}},y:{beginAtZero:true,ticks:{callback:v=>bytesPerSecond(v)}}}, plugins:{legend:{display:true}}}}); }
 const trafficChart = makeChart('traffic-chart',[{label:'Inbound',data:[],borderColor:'#3273dc',backgroundColor:'#3273dc',tension:.25,pointRadius:0},{label:'Outbound',data:[],borderColor:'#23d160',backgroundColor:'#23d160',tension:.25,pointRadius:0}]);
@@ -58,7 +84,93 @@ function render(data) { latest=data; const windowSize=Number($('window').value);
 function connect() { if(source) source.close(); source=new EventSource('/api/events?window='+$('window').value); source.onopen=()=>{$('connection').textContent='Live · updates every second';}; source.onmessage=e=>render(JSON.parse(e.data)); source.onerror=()=>{$('connection').textContent='Reconnecting…';}; }
 function renderHosts(hosts) { $('hosts').innerHTML=hosts.map((host,i)=>`<label class="checkbox is-block mb-2"><input type="checkbox" value="${escapeHtml(host)}" ${selected.has(host)?'checked':''}> <span class="legend-dot" style="background:${palette[i%palette.length]}"></span>${escapeHtml(host)}</label>`).join('') || '<p class="has-text-grey">No hosts observed yet.</p>'; document.querySelectorAll('#hosts input').forEach(input=>input.onchange=()=>{input.checked?selected.add(input.value):selected.delete(input.value); if(latest) render(latest);}); }
 function escapeHtml(s) { return s.replace(/[&<>'"]/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;',"'":'&#39;','"':'&quot;'}[c])); }
-const oldRender=render; render=data=>{oldRender(data);renderHosts(data.hosts);};
+let openedRequest = null, openerId = null, modalGeneration = 0;
-$('theme').value=localStorage.getItem('nginx-theme')||'system'; applyTheme($('theme').value); $('theme').onchange=()=>{localStorage.setItem('nginx-theme',$('theme').value);applyTheme($('theme').value);}; $('window').onchange=connect; connect();
+function requestFields(record) {
+  return [
+    ['Request ID', String(record.id)], ['Completed at', new Date(record.timestamp*1000).toISOString()],
+    ['Host', record.host], ['Client address', record.client_address || 'Unavailable'],
+    ['Method', record.method], ['Path', record.path || 'Unavailable'], ['Status', String(record.status)],
+    ['Direction', record.direction === 'inbound' ? 'Inbound' : 'Outbound'],
+    ['Inbound size', formatBytes(record.incoming), record.incoming],
+    ['Outbound size', formatBytes(record.outgoing), record.outgoing],
+    ['Request duration', duration(record.request_time)], ['Upstream average', duration(record.upstream_time)]
+  ];
+}
+function requestCopyText(record) {
+  return requestFields(record).map(([label, value, bytes]) => `${label}: ${value}${bytes === undefined ? '' : ` (${bytes} bytes)`}`).join('\n');
+}
+function openRequest(record) {
+  openedRequest = {...record};
+  openerId = String(record.id);
+  modalGeneration++;
+  $('request-details').replaceChildren();
+  for (const [label, value, bytes] of requestFields(openedRequest)) {
+    const term = document.createElement('dt'), detail = document.createElement('dd');
+    term.textContent = label;
+    detail.textContent = value;
+    if (bytes !== undefined) detail.title = `${bytes} bytes`;
+    $('request-details').append(term, detail);
+  }
+  $('request-copy-status').textContent = '';
+  $('request-copy-text').hidden = true;
+  $('request-copy-text').value = '';
+  $('request-dialog').showModal();
+}
+function renderRequests(data) {
+  const focusedId = document.activeElement?.dataset.requestId;
+  const records = data.top_requests_by_window?.[$('window').value] || [];
+  const rows = records.map((record, index) => {
+    const row = document.createElement('tr');
+    row.tabIndex = 0;
+    row.dataset.requestId = String(record.id);
+    const inbound = record.direction === 'inbound';
+    for (const value of [String(index+1), `${inbound ? '↓ Inbound' : '↑ Outbound'}`, record.host, record.client_address || 'Unavailable', formatBytes(record.size)]) {
+      const cell = document.createElement('td');
+      cell.textContent = value;
+      row.append(cell);
+    }
+    row.lastElementChild.title = `${record.size} bytes`;
+    row.lastElementChild.className = 'request-size';
+    row.onclick = () => openRequest(record);
+    row.onkeydown = event => { if (event.key === 'Enter' || event.key === ' ') { event.preventDefault(); openRequest(record); } };
+    return row;
+  });
+  $('request-rows').replaceChildren(...rows);
+  $('requests-empty').hidden = records.length > 0;
+  if (focusedId && !$('request-dialog').open) {
+    (rows.find(row => row.dataset.requestId === focusedId) || $('top-requests-title')).focus();
+  }
+}
+$('request-close').onclick = () => $('request-dialog').close();
+$('request-dialog').addEventListener('click', event => {
+  const bounds = $('request-dialog').getBoundingClientRect();
+  if (event.target === $('request-dialog') && (event.clientX < bounds.left || event.clientX > bounds.right || event.clientY < bounds.top || event.clientY > bounds.bottom)) $('request-dialog').close();
+});
+$('request-dialog').addEventListener('close', () => {
+  openedRequest = null;
+  modalGeneration++;
+  $('request-details').replaceChildren();
+  $('request-copy-text').value = '';
+  const opener = [...$('request-rows').children].find(row => row.dataset.requestId === openerId);
+  (opener || $('top-requests-title')).focus();
+  openerId = null;
+});
+$('request-copy').onclick = async () => {
+  if (!openedRequest) return;
+  const text = requestCopyText(openedRequest), generation = modalGeneration;
+  try {
+    await navigator.clipboard.writeText(text);
+    if (generation === modalGeneration) $('request-copy-status').textContent = 'Copied all request details.';
+  } catch {
+    if (generation !== modalGeneration) return;
+    $('request-copy-status').textContent = 'Clipboard unavailable. Select and copy the details below.';
+    $('request-copy-text').hidden = false;
+    $('request-copy-text').value = text;
+    $('request-copy-text').focus();
+    $('request-copy-text').select();
+  }
+};
+const oldRender=render; render=data=>{oldRender(data);renderHosts(data.hosts);renderRequests(data);};
+$('theme').value=localStorage.getItem('nginx-theme')||'system'; applyTheme($('theme').value); $('theme').onchange=()=>{localStorage.setItem('nginx-theme',$('theme').value);applyTheme($('theme').value);}; $('window').onchange=()=>{if(latest) render(latest); connect();}; connect();
 </script>
 </body></html>

tests/test_main.py

@@ -1,13 +1,135 @@
 import time
+import json
+import random
+import asyncio
 
-from app.main import Telemetry, parse_line
+import pytest
+
+from app.main import Telemetry, create_app, parse_line
+
+
+def json_record(**changes):
+    record = dict(timestamp="1720000000.123", host="example.test", method="POST",
+                  status="201", incoming="6460500000", outgoing="240", request_time="180.0",
+                  upstream_response_time="0.100, 0.200", client_address="2001:db8::1",
+                  path='/upload/"file"/<script>')
+    record.update(changes)
+    return json.dumps(record)
+
+
+def test_json_details_and_legacy_compatibility():
+    event = parse_line(json_record())
+    assert event["client_address"] == "2001:db8::1"
+    assert event["path"] == '/upload/"file"/<script>'
+    assert event["method"] == "POST"
+    assert event["incoming"] == 6460500000
+    assert event["upstream_time"] == pytest.approx(.15)
+    assert parse_line(json_record(path="/upload?token=secret"))["path"] == "/upload"
+    for upstream in ("", "-"):
+        assert parse_line(json_record(upstream_response_time=upstream))["upstream_time"] is None
+    legacy = parse_line("1720000000.123 example.test GET 200 120 240 .1 -")
+    assert legacy["client_address"] is None
+    assert legacy["path"] is None
+    assert legacy["method"] == "GET"
+
+
+@pytest.mark.parametrize("line", [
+    '{"timestamp":', '{}', '{"timestamp":null}',
+    json_record(incoming="-1"), json_record(outgoing="2.5"),
+    json_record(timestamp="NaN"), json_record(request_time="Infinity"),
+    json_record(status="600"), json_record(client_address=[]),
+    json_record(upstream_response_time="not-a-time"),
+])
+def test_invalid_json_records_are_ignored(line):
+    assert parse_line(line) is None
+
+
+def test_top_requests_matches_full_sort_and_window_boundaries(monkeypatch):
+    now = 1720004000
+    monkeypatch.setattr("app.main.time.time", lambda: now)
+    telemetry = Telemetry(3600)
+    randomizer = random.Random(42)
+    all_records = []
+    # Dense seconds exercise per-bucket truncation, while boundary seconds
+    # exercise membership in every selectable window.
+    ages = (0, 59, 60, 299, 300, 899, 900, 3599, 3600)
+    for age in ages:
+        for _ in range(30):
+            event = parse_line(json_record(timestamp=str(now-age),
+                               incoming=str(randomizer.randrange(100)),
+                               outgoing=str(randomizer.randrange(100))))
+            telemetry.ingest(event)
+            all_records.append(dict(event, id=telemetry.next_request_id,
+                                    size=max(event["incoming"], event["outgoing"])))
+    snapshot = telemetry.snapshot(60)
+    for window in (60, 300, 900, 3600):
+        expected = sorted((record for record in all_records if now-window+1 <= record["timestamp"] <= now),
+                          key=lambda record: (record["size"], record["timestamp"], record["id"]), reverse=True)[:10]
+        actual = snapshot["top_requests_by_window"][str(window)]
+        assert [record["id"] for record in actual] == [record["id"] for record in expected]
+    assert all(len(records) <= 10 for records in telemetry.request_buckets.values())
+    assert len(telemetry.request_buckets) == len(ages)-1
+    assert snapshot["summary"]["incoming"] == sum(r["incoming"] for r in all_records if r["timestamp"] >= now-59)/60
+    assert telemetry.total_incoming == sum(r["incoming"] for r in all_records)
+
+
+def test_directions_ties_unique_ids_and_idle_expiration(monkeypatch):
+    now = 1720004000
+    monkeypatch.setattr("app.main.time.time", lambda: now)
+    telemetry = Telemetry(60)
+    for incoming, outgoing in ((100, 100), (100, 100), (1, 200), (300, 1)):
+        telemetry.ingest(parse_line(json_record(timestamp=str(now), incoming=str(incoming), outgoing=str(outgoing))))
+    records = telemetry.snapshot(60)["top_requests_by_window"]["60"]
+    assert [record["id"] for record in records] == [4, 3, 2, 1]
+    assert [record["direction"] for record in records] == ["inbound", "outbound", "inbound", "inbound"]
+    now += 59
+    assert len(telemetry.snapshot(60)["top_requests_by_window"]["60"]) == 4
+    now += 1
+    assert all(not records for records in telemetry.snapshot(3600)["top_requests_by_window"].values())
+    assert telemetry.request_buckets == {}
+
+
+def test_out_of_order_old_and_future_requests(monkeypatch):
+    now = 1720004000
+    monkeypatch.setattr("app.main.time.time", lambda: now)
+    telemetry = Telemetry(60)
+    for age in (0, 20, 59, 61, -100):
+        telemetry.ingest(parse_line(json_record(timestamp=str(now-age))))
+    records = telemetry.snapshot(60)["top_requests_by_window"]["60"]
+    assert len(records) == 4
+    assert records[0]["timestamp"] == now+100
+    assert len(telemetry.request_buckets) == 3
+
+
+def test_rest_and_sse_include_request_details(monkeypatch):
+    now = 1720004000
+    monkeypatch.setattr("app.main.time.time", lambda: now)
+    app = create_app()
+    app.state.telemetry.ingest(parse_line(json_record(timestamp=str(now))))
+    endpoints = {route.path: route.endpoint for route in app.routes if hasattr(route, "endpoint")}
+
+    async def check():
+        rest = await endpoints["/api/telemetry"](window=60)
+        response = await endpoints["/api/events"](request=None, window=60)
+        try:
+            first = await anext(response.body_iterator)
+            streamed = json.loads(first.removeprefix("data: ").strip())
+            assert streamed["top_requests_by_window"] == rest["top_requests_by_window"]
+            record = streamed["top_requests_by_window"]["60"][0]
+            assert record["client_address"] == "2001:db8::1"
+            assert record["size"] == 6460500000
+        finally:
+            await response.body_iterator.aclose()
+        assert not app.state.telemetry.subscribers
+
+    asyncio.run(check())
 
 
 def test_parse_line_and_multiple_upstreams():
     event = parse_line("1720000000.123 example.test GET 200 120 240 0.150 0.100, 0.200")
     assert event["host"] == "example.test"
     assert event["incoming"] == 120
-    assert event["upstream_time"] == 0.15
+    assert event["upstream_time"] == pytest.approx(0.15)
 
 
 def test_parse_line_with_empty_trailing_upstream_time():