**Phase 14 `store_performance` — final verification pass (all 4 tasks already in `complete/`)** - Re-verified every task's work in the tree: read caches (`series_names`/`stats`), capped compound `pick_res_many`, `history_many` + `api._history_batch`, windowed `maintain()` + per-resolution prune + `PRAGMA optimize`/`automatic_analyze=0`/WAL pragmas, `store_perf.BUDGETS` (14 ceilings), probe, live watch. - **Completion criteria:** baseline+after side by side in `data/logs/NOTES-store-perf.md` ✅ · budget tests green (`uv run pytest -m store_perf --store-perf` → **4 passed**, 14/14 inside) ✅ · 25-min dev-server watch re-run **independently** (74 passes, `scripts/watch-store-perf.py`): **0 `store-error`** (total still 190, newest 2026‑10‑18 06:33:30), WAL 7.77 MB ≤ 67.1 MB, medians 1.7–3.2 ms, p95 720/613 ms = the documented cold-cache window; a 3rd 27-pass watch on a freshly restarted server reproduced it (3.1/708, 1.8/572) ✅ · `maintain()` 5,368 ms → 2.7–6.8 ms and WAL bounded across two cycles (test + notes) ✅ · no behaviour change (all four history routes pinned against `query()`) ✅ - **Full gates:** `uv run pytest` → **1267 passed, 15 skipped**; `bash .agents/validate.sh` → **exit 0**, coverage **98.75%** (floor 94), `ruff check src tests agent` clean, `pyright` 0 errors, `node --check` ok, repo hygiene ok; `uv run pytest -m live --live` → **11 passed**; dev server up, 9/9 sources `ok`, pages 200 in 1.5–3.3 ms. - **Docs fixed:** README/AGENTS still claimed a flat "20 s per pass" while `host_agent.py` (in this same uncommitted change) scales the budget to 5 s × disk count — corrected in both plus the two illustrative "20 s of wedged disk" sentences; broken bold (`` `driveripper** ``) repaired; stale gap-4 title + "tuning is its own phase" wording updated to say phase 14 ran; README's absolute "nothing … is answered by a whole-table scan" qualified to *per call* with the cold-window cost named; "third opt-in marker" → second; conftest `--store-perf` help named the real marker; **`data/logs/watch-store-perf.py` moved to `scripts/watch-store-perf.py`** (both docs told a clone to run a gitignored path) with root-anchored paths, `DASH_WATCH_OUT`, and dead code removed. - **New finding recorded:** gap 9's "~10 s process-wide stall is still gap 4's / `14_store_performance`'s problem" is now closed — coincident >8 s `llm` + >12 s `ups` raw-stream gaps went 393 (00:00–12:00) and 198 (12:00–21:36) to **0** across 110+ min of the phase-14 store; written into AGENTS.md gaps 4 and 9 and the notes table. - **Deviations:** did *not* pull either deferred cost knob (`STATS_CACHE_TTL_SECONDS`, warming caches from the maintenance tick) — AGENTS.md records that decision and its reasoning; `uv pip list --outdated` still reports certifi/fastapi/starlette/websockets (pre-existing advisory, no dependency touched this phase, gate treats it as advisory). Overwrote nothing: task 04's watch preserved as `store-perf-watch.task04.*`, my two runs as `.verify25.*` / `.current-code.*`. - **Next pending phase:** `15_agent_mtls`, then `16_agent_smart_pass_budget` (owns the scaled-budget tests + its doc criterion, now partly pre-satisfied by this pass), `17_installer_device_filter`.
1633 lines
87 KiB
Python
1633 lines
87 KiB
Python
"""History store: rollups, retention, and the honesty of an empty chart.
|
|
|
|
The rollup queries are plain SQL with positional bindings, and a mismatch there used to
|
|
raise on every maintenance tick while the poller swallowed it — so every window past the
|
|
raw retention was silently empty. These tests fail loudly if that ever happens again.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import math
|
|
import re
|
|
import sqlite3
|
|
import time
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
import store_perf
|
|
from dashboards.store import (
|
|
COMPOUND_CHUNK,
|
|
HOUR,
|
|
JOURNAL_SIZE_LIMIT_BYTES,
|
|
LAST_MAINTAIN_KEY,
|
|
MAINTAIN_SLACK_SECONDS,
|
|
MINUTE,
|
|
RES_HOUR,
|
|
RES_MIN,
|
|
RES_RAW,
|
|
ROLLUP_INSERT,
|
|
SERIES_CACHE_MAX_ENTRIES,
|
|
WAL_AUTOCHECKPOINT_PAGES,
|
|
Event,
|
|
Sample,
|
|
SampleStore,
|
|
samples_from_flat,
|
|
select_window,
|
|
)
|
|
|
|
|
|
def add_series(db, source: str, host: str, metric: str, values, *, start: float, step: float,
|
|
res: str = RES_RAW) -> None:
|
|
for i, value in enumerate(values):
|
|
# the private handle is the point: writing at an explicit resolution is what
|
|
# the resolution-choice tests need
|
|
db._db.execute(
|
|
"INSERT INTO sample (source, host, metric, res, ts, value) VALUES (?,?,?,?,?,?)",
|
|
(source, host, metric, res, start + i * step, float(value)),
|
|
)
|
|
db._db.commit()
|
|
|
|
|
|
def test_store_starts_in_wal_mode(db) -> None:
|
|
"""Readers poll while the writer commits; without WAL the API would hit locked errors."""
|
|
assert db._db.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
|
|
|
|
|
|
def test_add_and_query_roundtrip(db) -> None:
|
|
now = time.time()
|
|
db.add("host", "driveripper", "cpu.usage_pct", 42.5, ts=now)
|
|
points = db.query("host", "driveripper", "cpu.usage_pct", start=now - 60)
|
|
assert points == [(now, 42.5)]
|
|
|
|
|
|
def test_query_is_bounded_and_ascending(db) -> None:
|
|
base = 1_700_000_000.0
|
|
add_series(db, "host", "h", "m", [3, 1, 2], start=base, step=10)
|
|
assert db.query("host", "h", "m", start=base - 1) == [(base, 3.0), (base + 10, 1.0), (base + 20, 2.0)]
|
|
assert len(db.query("host", "h", "m", start=base - 1, limit=2)) == 2
|
|
|
|
|
|
def test_query_tail_keeps_the_newest_rows(db) -> None:
|
|
base = 1_700_000_000.0
|
|
add_series(db, "host", "h", "m", range(100), start=base, step=10)
|
|
head = db.query("host", "h", "m", start=base - 1, limit=3)
|
|
tail = db.query("host", "h", "m", start=base - 1, limit=3, tail=True)
|
|
assert [v for _, v in head] == [0.0, 1.0, 2.0]
|
|
assert [v for _, v in tail] == [97.0, 98.0, 99.0]
|
|
assert tail == sorted(tail), "tail must still come back oldest-first"
|
|
|
|
|
|
def test_samples_from_flat_drops_unknown_and_none() -> None:
|
|
payload: dict[str, Any] = {"a": 1.0, "b": None, "c": "nope"}
|
|
samples = samples_from_flat("host", "gw", payload, ts=5.0)
|
|
assert [(s.metric, s.value) for s in samples] == [("a", 1.0)]
|
|
|
|
|
|
def test_maintain_rolls_raw_into_minutes(db) -> None:
|
|
"""Regression: the rollup statement had 7 placeholders and 6 bindings."""
|
|
now = time.time()
|
|
# two full, aligned minute buckets — see the hour test for why alignment matters
|
|
minute = int(now // MINUTE) * MINUTE
|
|
add_series(db, "host", "h", "cpu.usage_pct", [10, 20, 30, 40], start=minute - 4 * MINUTE,
|
|
step=1)
|
|
add_series(db, "host", "h", "cpu.usage_pct", [1, 1, 1, 1], start=minute - 3 * MINUTE, step=1)
|
|
counts = db.maintain(now=now)
|
|
assert counts["to_minute"] > 0
|
|
minutes = db.query("host", "h", "cpu.usage_pct", res=RES_MIN, start=now - 3600)
|
|
assert minutes, "no 1m rollup was written"
|
|
assert [round(value) for _ts, value in minutes] == [25, 1]
|
|
|
|
|
|
def test_maintain_is_idempotent(db) -> None:
|
|
"""Rollups run on a timer: running twice must not double-count or duplicate rows.
|
|
|
|
The second pass is asked for the whole window again (the `last_maintain` row is deleted
|
|
first), because since phase 14 task 03 a pass that has recorded its own clock has nothing
|
|
left to do — the window is where the work got bounded, and this test is about the rollup being
|
|
an `INSERT OR REPLACE` rewrite rather than a duplicate row. `test_a_second_maintain_reads_
|
|
only_the_window_since_the_last_run` is the other half: the pass that does run again at a later
|
|
clock re-reads only what became due since.
|
|
"""
|
|
now = time.time()
|
|
minute = int(now // MINUTE) * MINUTE
|
|
add_series(db, "host", "h", "m", [10, 20], start=minute - 4 * MINUTE, step=1)
|
|
first = db.maintain(now=now)
|
|
rows = db.query("host", "h", "m", res=RES_MIN, start=now - 3600)
|
|
count = db.count()
|
|
assert db.delete_meta("last_maintain") == 1, "the whole-window re-run needs no previous pass"
|
|
second = db.maintain(now=now)
|
|
assert rows == db.query("host", "h", "m", res=RES_MIN, start=now - 3600)
|
|
assert first["to_minute"] == second["to_minute"] > 0
|
|
assert db.count() == count, "re-running maintenance duplicated samples"
|
|
|
|
|
|
def test_maintain_rolls_minutes_into_hours(db) -> None:
|
|
now = time.time()
|
|
# buckets are aligned to the epoch (`ts/3600*3600`), not to "hours from now", so the
|
|
# fixtures start on a boundary or the two hours mix into a third, averaged bucket
|
|
hour = int(now // HOUR) * HOUR
|
|
# two *complete* hours of minute buckets: the current and previous hour are still
|
|
# filling in, and maintenance must not touch them (see the partial-bucket test)
|
|
add_series(db, "host", "h", "m", [1] * 60, start=hour - 5 * HOUR, step=60, res=RES_MIN)
|
|
add_series(db, "host", "h", "m", [3] * 60, start=hour - 4 * HOUR, step=60, res=RES_MIN)
|
|
counts = db.maintain(now=now)
|
|
assert counts["to_hour"] > 0
|
|
hours = db.query("host", "h", "m", res=RES_HOUR, start=now - 86_400)
|
|
assert sorted(round(v) for _, v in hours) == [1.0, 3.0]
|
|
|
|
|
|
|
|
def test_maintain_prunes_expired_raw(db) -> None:
|
|
now = time.time()
|
|
old = now - (db.raw_keep_seconds + 7200)
|
|
add_series(db, "host", "h", "m", [1, 2], start=old, step=60)
|
|
add_series(db, "host", "h", "fresh", [1], start=now - 60, step=1)
|
|
counts = db.maintain(now=now)
|
|
assert counts["pruned"] >= 2
|
|
assert db.query("host", "h", "m", start=0) == []
|
|
|
|
|
|
def test_maintain_records_timestamp_in_stats(db) -> None:
|
|
db.maintain()
|
|
stats = db.stats()
|
|
assert stats["last_maintain"] is not None
|
|
assert "samples_by_res" in stats and "path" in stats
|
|
|
|
|
|
def test_pick_res_prefers_the_finest_resolution_that_fits(db) -> None:
|
|
now = time.time()
|
|
add_series(db, "host", "h", "m", list(range(500)), start=now - 3600, step=1)
|
|
assert db.pick_res("host", "h", "m", 3600, 600) == RES_RAW
|
|
add_series(db, "host", "h", "m", list(range(4000)), start=now - 7200, step=1)
|
|
assert db.pick_res("host", "h", "m", 7200, 600) == RES_RAW, "raw still fits"
|
|
add_series(db, "host", "h", "m", list(range(120)), start=now - 7200, step=60,
|
|
res=RES_MIN)
|
|
# 4500 raw points blow the 600 budget, so the minute rollup takes over
|
|
assert db.pick_res("host", "h", "m", 7200, 600) == RES_MIN
|
|
|
|
|
|
def test_pick_res_never_picks_an_empty_resolution(db) -> None:
|
|
"""A fresh install has raw only: choosing 1h because it 'fits' would chart nothing."""
|
|
now = time.time()
|
|
add_series(db, "host", "h", "m", [1] * 2000, start=now - 7200, step=1)
|
|
assert db.pick_res("host", "h", "m", 86_400, 600) == RES_RAW
|
|
|
|
|
|
def test_history_falls_back_to_raw_when_rollups_lag(db) -> None:
|
|
now = time.time()
|
|
add_series(db, "host", "h", "m", list(range(1500)), start=now - 1500, step=1)
|
|
points = db.history("host", "h", "m", span=3600, max_points=600)
|
|
assert len(points) > 0, "empty chart while raw data exists"
|
|
assert len(points) <= 600
|
|
|
|
|
|
def test_history_always_keeps_the_newest_point(db) -> None:
|
|
now = time.time()
|
|
values = list(range(2000))
|
|
add_series(db, "host", "h", "m", values, start=now - 2000, step=1)
|
|
points = db.history("host", "h", "m", span=3600, max_points=50)
|
|
assert points[-1][1] == pytest.approx(values[-1])
|
|
|
|
|
|
def test_series_names_and_hosts(db) -> None:
|
|
db.add("host", "a", "cpu.usage_pct", 1.0)
|
|
db.add("host", "a", "mem.used_pct", 2.0)
|
|
db.add("host", "b", "cpu.usage_pct", 3.0)
|
|
db.add("llm", "gufo", "llm.decode_tokens_per_second", 4.0)
|
|
assert db.series_names("host", "a") == ["cpu.usage_pct", "mem.used_pct"]
|
|
assert db.hosts("host") == ["a", "b"]
|
|
assert db.series_names("llm") == ["llm.decode_tokens_per_second"]
|
|
|
|
|
|
def test_delete_series_cleans_up(db) -> None:
|
|
db.add("host", "a", "gone.metric", 1.0)
|
|
db.add("host", "a", "keep.metric", 2.0)
|
|
assert db.delete_series("host", "a", "gone.metric") == 1
|
|
assert db.series_names("host", "a") == ["keep.metric"]
|
|
|
|
|
|
def test_events_roundtrip_and_filter(db) -> None:
|
|
now = time.time()
|
|
db.log_event(Event(ts=now, source="host", host="a", kind="down", severity="error",
|
|
message="a is not answering", data={"fails": 3}))
|
|
db.log_event(Event(ts=now + 1, source="nut", host="cyberpower", kind="state",
|
|
severity="warn", message="OL → OB"))
|
|
everything = db.recent_events(10)
|
|
assert [e.host for e in everything] == ["cyberpower", "a"]
|
|
assert everything[-1].data == {"fails": 3}
|
|
assert [e.source for e in db.recent_events(10, source="nut")] == ["nut"]
|
|
assert len(db.recent_events(1)) == 1
|
|
# the API hands out plain JSON, so the boundary has to be serialisable
|
|
assert everything[0].as_json()["message"] == "OL → OB"
|
|
|
|
|
|
def test_event_as_json_is_serialisable(db) -> None:
|
|
event = Event(ts=1.0, source="host", host="a", kind="up", severity="info", message="ok")
|
|
payload = event.as_json()
|
|
assert payload["severity"] == "info" and payload["data"] is None
|
|
|
|
|
|
def test_count_filters_by_resolution(db) -> None:
|
|
now = time.time()
|
|
db.add("host", "h", "m", 1.0, ts=now)
|
|
add_series(db, "host", "h", "m", [2.0], start=now, step=1, res=RES_MIN)
|
|
assert db.count() == 2
|
|
assert db.count(res=RES_RAW) == 1
|
|
assert db.count(res=RES_HOUR) == 0
|
|
|
|
|
|
def test_add_many_reports_inserted_rows(db) -> None:
|
|
now = time.time()
|
|
rows = [Sample("host", "h", f"m{i}", float(i), now) for i in range(5)]
|
|
assert db.add_many(rows) == 5
|
|
assert db.count() == 5
|
|
|
|
|
|
def test_store_reopens_without_losing_history(tmp_path) -> None:
|
|
path = tmp_path / "reopen.sqlite3"
|
|
first = SampleStore(path)
|
|
now = time.time()
|
|
first.add("host", "h", "m", 7.0, ts=now)
|
|
first.close()
|
|
second = SampleStore(path)
|
|
try:
|
|
assert second.query("host", "h", "m", start=now - 5) == [(now, 7.0)]
|
|
finally:
|
|
second.close()
|
|
|
|
|
|
def test_select_window_maps_range_onto_presets() -> None:
|
|
assert select_window(60, [300, 3600, 86_400]) == 300
|
|
assert select_window(3600, [300, 3600, 86_400]) == 3600
|
|
assert select_window(86_401, [300, 3600, 86_400]) == 86_400
|
|
|
|
|
|
def test_rollup_does_not_average_a_partial_minute(db) -> None:
|
|
"""A bucket is only rolled once its whole window is in the past, or it gets rewritten."""
|
|
now = time.time()
|
|
add_series(db, "host", "h", "m", [1, 1, 1], start=now - 5, step=1) # current minute
|
|
db.maintain(now=now)
|
|
assert db.query("host", "h", "m", res=RES_MIN, start=now - 600) == []
|
|
|
|
|
|
def test_hour_constant_is_used_in_retention(db) -> None:
|
|
assert db.raw_keep_seconds == pytest.approx(48 * HOUR)
|
|
|
|
|
|
# ---- a 300-second source (phase 09: `disks` samples SMART every 5 minutes) -- #
|
|
#
|
|
# The rollup SQL groups by minute and hour buckets, so a source slower than a minute
|
|
# puts one raw row in every fifth bucket and nothing in the other four. Two things had to
|
|
# be true and are asserted rather than assumed: an empty bucket produces *no* rollup row
|
|
# (inventing a value for a minute nothing was sampled is rule 2), and the hour bucket
|
|
# still averages the samples it did get.
|
|
|
|
SMART_STEP = 300.0
|
|
|
|
|
|
def test_a_300s_source_rolls_into_minutes_without_inventing_the_empty_ones(db) -> None:
|
|
now = time.time()
|
|
# five SMART polls at exactly the phase's cadence (-30, -25, -20, -15, -10 minutes),
|
|
# so every bucket is fully in the past (+60 s slack) and maintenance may roll all five
|
|
start = int(now // MINUTE) * MINUTE - 30 * MINUTE
|
|
add_series(db, "disks", "driveripper", "disk.temp_c@nvme0n1",
|
|
[38, 39, 38, 40, 37], start=start, step=SMART_STEP)
|
|
counts = db.maintain(now=now)
|
|
minutes = db.query("disks", "driveripper", "disk.temp_c@nvme0n1", res=RES_MIN, start=0)
|
|
assert counts["to_minute"] == 5
|
|
assert [value for _ts, value in minutes] == [38.0, 39.0, 38.0, 40.0, 37.0], (
|
|
"a bucket holding one sample rolls up that sample, unchanged")
|
|
assert [int(ts) for ts, _value in minutes] == [
|
|
int(start + index * SMART_STEP // MINUTE * MINUTE) for index in range(5)], (
|
|
"every fifth minute bucket has a row; the empty ones must have none")
|
|
assert db.count(res=RES_MIN) == 5, "empty buckets became fabricated rows"
|
|
|
|
|
|
def test_a_300s_source_averages_into_the_hour_bucket_it_lands_in(db) -> None:
|
|
now = time.time()
|
|
# aligned to the epoch, and old enough that the hour bucket is complete: 12 samples
|
|
# five minutes apart inside one hour average to their mean, not to a sample count
|
|
hour = int(now // HOUR) * HOUR - 3 * HOUR
|
|
add_series(db, "disks", "driveripper", "disk.power_on_hours@nvme0n1",
|
|
[100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 124],
|
|
start=hour, step=SMART_STEP)
|
|
db.maintain(now=now)
|
|
db.maintain(now=now) # the 1h pass reads 1m rows, so the first pass only makes them
|
|
hours = db.query("disks", "driveripper", "disk.power_on_hours@nvme0n1", res=RES_HOUR,
|
|
start=0)
|
|
assert len(hours) == 1
|
|
assert hours[0][0] == hour
|
|
assert hours[0][1] == pytest.approx((11 * 100 + 124) / 12)
|
|
|
|
|
|
def test_a_300s_series_is_queryable_at_every_resolution(db) -> None:
|
|
"""A 5-minute series over a 1h window is 12 points; `pick_res` must find them raw."""
|
|
now = time.time()
|
|
add_series(db, "disks", "driveripper", "disk.reallocated@nvme0n1",
|
|
[8] * 12, start=now - 3600 + 60, step=SMART_STEP)
|
|
assert db.pick_res("disks", "driveripper", "disk.reallocated@nvme0n1", 3600, 600) == RES_RAW
|
|
points = db.history("disks", "driveripper", "disk.reallocated@nvme0n1", span=3600)
|
|
assert len(points) == 12, "the chart is empty while raw history exists"
|
|
db.maintain(now=now)
|
|
db.maintain(now=now)
|
|
# with rollups present, a long window reads the rolled-up series, not 288 raw rows
|
|
assert db.pick_res("disks", "driveripper", "disk.reallocated@nvme0n1",
|
|
24 * 3600, 600) in (RES_MIN, RES_RAW)
|
|
|
|
|
|
def test_a_300s_source_is_pruned_by_the_existing_retention(db) -> None:
|
|
"""288 rows a day per series is fine for 48h and not fine forever — rule 9."""
|
|
now = time.time()
|
|
expired = now - db.raw_keep_seconds - 6 * SMART_STEP
|
|
add_series(db, "disks", "h", "disk.temp_c@sda", [30] * 4, start=expired, step=SMART_STEP)
|
|
add_series(db, "disks", "h", "disk.temp_c@sdb", [30], start=now - SMART_STEP,
|
|
step=SMART_STEP)
|
|
counts = db.maintain(now=now)
|
|
assert counts["pruned"] >= 4
|
|
assert db.query("disks", "h", "disk.temp_c@sda", start=0) == []
|
|
assert len(db.query("disks", "h", "disk.temp_c@sdb", start=0)) == 1
|
|
|
|
|
|
# ---- meta: the durable key/value state (phase 11) --------------------------- #
|
|
|
|
|
|
def test_get_meta_answers_none_for_a_key_nobody_wrote(db) -> None:
|
|
"""`None` and `""` are different sentences here.
|
|
|
|
`auth.verifier` missing means "this box has never been seeded" (fail closed at startup),
|
|
while a value means "the credential lives here". A store that returned `""` for an absent
|
|
row would read as a corrupt one instead.
|
|
"""
|
|
assert db.get_meta("no.such.key") is None
|
|
db.set_meta("empty.on.purpose", "")
|
|
assert db.get_meta("empty.on.purpose") == ""
|
|
|
|
|
|
def test_meta_round_trips_through_a_reopen(db) -> None:
|
|
db.set_meta("auth.session_secret", "ab" * 32)
|
|
assert db.get_meta("auth.session_secret") == "ab" * 32
|
|
db.close()
|
|
reopened = SampleStore(db.path)
|
|
try:
|
|
assert reopened.get_meta("auth.session_secret") == "ab" * 32
|
|
finally:
|
|
reopened.close()
|
|
|
|
|
|
def test_set_meta_replaces_instead_of_adding(db) -> None:
|
|
"""One row per key (`INSERT OR REPLACE`), because `key` is the primary key.
|
|
|
|
An `INSERT` alone would raise on the second write, which is the change-password path
|
|
(`auth.update_verifier`) failing after the user pressed Save.
|
|
"""
|
|
db.set_meta("auth.verifier", "first")
|
|
db.set_meta("auth.verifier", "second")
|
|
assert db.get_meta("auth.verifier") == "second"
|
|
(rows,) = db._db.execute( # the private handle is the point: exactly one row, not two
|
|
"SELECT COUNT(*) FROM meta WHERE key='auth.verifier'").fetchone()
|
|
assert rows == 1
|
|
|
|
|
|
def test_delete_meta_reports_whether_a_row_was_there(db) -> None:
|
|
"""The rowcount is the whole point: the password-reset procedure has to say what it did."""
|
|
db.set_meta("auth.verifier", "x")
|
|
assert db.delete_meta("auth.verifier") == 1
|
|
assert db.delete_meta("auth.verifier") == 0
|
|
assert db.get_meta("auth.verifier") is None
|
|
|
|
|
|
def test_delete_meta_touches_only_the_key_it_names(db) -> None:
|
|
"""A reset that deletes one row too many deletes the operator's only credential."""
|
|
db.set_meta("auth.verifier", "the-verifier")
|
|
db.set_meta("auth.session_secret", "the-secret")
|
|
db.set_meta("last_maintain", "1700000000.0")
|
|
|
|
assert db.delete_meta("auth.session_secret") == 1
|
|
|
|
assert db.get_meta("auth.session_secret") is None
|
|
assert db.get_meta("auth.verifier") == "the-verifier"
|
|
assert db.get_meta("last_maintain") == "1700000000.0"
|
|
|
|
|
|
def test_stats_reads_last_maintain_through_get_meta(db) -> None:
|
|
"""One path to the table.
|
|
|
|
`stats()` reported `None` before any maintenance ran and the timestamp afterwards; that
|
|
behaviour is pinned here while the read goes through the same helper everything else uses,
|
|
and it is why the helper is called outside `stats()`'s own lock (`threading.Lock` is not
|
|
reentrant — call it inside and the dashboard deadlocks on the first `/api/health`).
|
|
"""
|
|
assert db.stats()["last_maintain"] is None
|
|
db.maintain(now=1_700_000_000.0)
|
|
assert db.stats()["last_maintain"] == 1_700_000_000.0
|
|
|
|
|
|
# ---- phase 14 task 02: nothing on a per-request path may scan the table ---------------- #
|
|
#
|
|
# `TODO.md:7` is "pages take forever to load and seem to lock up the database", and the half of
|
|
# it a browser feels was two whole-table answers served on every request (`series_names`,
|
|
# `stats`) plus three `COUNT`s per series to pick a chart's resolution. The pins below assert the
|
|
# two things that make a cache trustworthy rather than fast: it is *invalidated* by the write that
|
|
# would make it a lie, and it *expires* so a missed invalidation is a 60-second bug rather than a
|
|
# permanent one.
|
|
#
|
|
# They count SQL statements rather than wall time. `sqlite3`'s trace callback is the query counter
|
|
# the task asks for, and it is the right instrument: "the second call did not scan" is a fact
|
|
# about statements, and a wall-time assertion here would be a budget that a noisy laptop fails for
|
|
# the wrong reason (task 04 owns the budgets).
|
|
|
|
|
|
@contextmanager
|
|
def query_log(store: SampleStore) -> Iterator[list[str]]:
|
|
"""Every statement the store executes on this handle while the context is open, in order."""
|
|
seen: list[str] = []
|
|
store._db.set_trace_callback(seen.append)
|
|
try:
|
|
yield seen
|
|
finally:
|
|
store._db.set_trace_callback(None)
|
|
|
|
|
|
def plan_of(store: SampleStore, sql: str, args: tuple[Any, ...] = ()) -> str:
|
|
"""`EXPLAIN QUERY PLAN` as one line, so a test can say "seek, not scan" in words."""
|
|
with store._lock:
|
|
return " | ".join(row[3] for row in store._db.execute("EXPLAIN QUERY PLAN " + sql, args))
|
|
|
|
|
|
def test_series_names_answers_from_cache_and_a_write_invalidates_it(db) -> None:
|
|
"""The `/api/metrics` answer is cached — and a write ends the cache, not the clock.
|
|
|
|
A cache that outlives a write answers "why is this chart empty" with yesterday's truth, so
|
|
the invalidation is the load-bearing half: a new metric name must be visible on the very next
|
|
call, and a deleted one must be gone.
|
|
"""
|
|
now = time.time()
|
|
db.add("host", "driveripper", "cpu.usage_pct", 1.0, ts=now)
|
|
with query_log(db) as seen:
|
|
first = db.series_names("host")
|
|
assert [s for s in seen if "DISTINCT metric" in s], "the first call pays the scan once"
|
|
|
|
seen.clear()
|
|
assert db.series_names("host") == first
|
|
assert seen == [], "a cached series_names() must not touch the table at all"
|
|
|
|
db.add("host", "driveripper", "gpu.busy_pct@0", 2.0, ts=now + 1)
|
|
seen.clear()
|
|
assert db.series_names("host") == ["cpu.usage_pct", "gpu.busy_pct@0"], (
|
|
"a metric that appeared after the cached read has to be in the next answer")
|
|
assert [s for s in seen if "DISTINCT metric" in s], "the write must have invalidated it"
|
|
|
|
|
|
def test_every_write_that_can_change_the_series_list_invalidates_it(db) -> None:
|
|
"""`add`, `add_many` and `delete_series` — the three writes that move the name set."""
|
|
db.add("host", "a", "cpu.usage_pct", 1.0)
|
|
assert db.series_names("host") == ["cpu.usage_pct"]
|
|
db.add_many([Sample("host", "a", "mem.used_pct", 2.0, time.time())])
|
|
assert "mem.used_pct" in db.series_names("host")
|
|
assert db.delete_series("host", "a", "cpu.usage_pct") == 1
|
|
assert db.series_names("host") == ["mem.used_pct"], "a deleted series must not linger in the list"
|
|
|
|
|
|
def test_maintaining_the_store_invalidates_the_series_list(db) -> None:
|
|
"""Pruning can delete the last row of a name; the list must not promise it afterwards."""
|
|
old = time.time() - (db.raw_keep_seconds + 7200)
|
|
add_series(db, "host", "h", "gone.soon", [1, 2], start=old, step=60)
|
|
assert db.series_names("host") == ["gone.soon"]
|
|
db.maintain(now=time.time())
|
|
assert db.series_names("host") == []
|
|
|
|
|
|
def test_a_write_of_an_already_written_series_leaves_the_series_cache_warm(db) -> None:
|
|
"""The live defect the task-04 watch caught, pinned so it cannot come back quietly.
|
|
|
|
Task 02 invalidated the series cache on *every* sample write. On a store the hub writes to
|
|
every 2-300 s that meant the cache never served: `/api/metrics` on the running dev server was
|
|
answering with its full `SELECT DISTINCT` scan (535 ms of store mutex) on nearly every call —
|
|
the same per-request scan the task existed to remove. A poll of a series the store has already
|
|
written cannot change the *set of names*, so it must not cost a scan; only a new name can.
|
|
"""
|
|
now = time.time()
|
|
db.add("host", "driveripper", "cpu.usage_pct", 1.0, ts=now)
|
|
assert db.series_names("host") == ["cpu.usage_pct"]
|
|
|
|
db.add("host", "deskwork", "cpu.usage_pct", 1.0, ts=now) # the second host of the same name
|
|
assert db.series_names("host") == ["cpu.usage_pct"]
|
|
|
|
# the next nine polls, of series the store already knows: the answer costs no statement
|
|
with query_log(db) as seen:
|
|
for poll in range(9):
|
|
db.add_many([Sample("host", "driveripper", "cpu.usage_pct", float(poll), now + poll),
|
|
Sample("host", "deskwork", "cpu.usage_pct", float(poll), now + poll)])
|
|
# the transaction's own BEGIN/COMMIT ride in the trace; what must not be in there is a
|
|
# read, and above all not the `SELECT DISTINCT` this whole task is about
|
|
assert not [s for s in seen if s.lstrip().upper().startswith("SELECT")], (
|
|
"a poll should cost its inserts and nothing else")
|
|
seen.clear()
|
|
assert db.series_names("host") == ["cpu.usage_pct"]
|
|
assert seen == [], f"a write of known series must not re-scan the name list: {seen}"
|
|
|
|
|
|
def test_a_write_of_a_series_nobody_wrote_yet_still_invalidates_immediately(db) -> None:
|
|
"""The other half of the rule, in the direction the cache must never be lazy about.
|
|
|
|
A new metric name is exactly the event `/api/metrics` exists to report, and a source gaining
|
|
one is a deploy — so "eventually" is not good enough and the next call has to see it.
|
|
"""
|
|
now = time.time()
|
|
db.add("host", "driveripper", "cpu.usage_pct", 1.0, ts=now)
|
|
assert db.series_names("host") == ["cpu.usage_pct"]
|
|
with query_log(db) as seen:
|
|
db.add("host", "deskwork", "a.brand.new.metric", 2.0, ts=now + 1)
|
|
seen.clear()
|
|
assert db.series_names("host") == ["a.brand.new.metric", "cpu.usage_pct"], (
|
|
"a metric nobody has written since the last read must be in the very next answer")
|
|
assert "a.brand.new.metric" in db.series_names("host")
|
|
# …and a per-device stream on a name the list already carries is a new triple too
|
|
db.series_names("host")
|
|
db.add_many([Sample("host", "driveripper", "gpu.busy_pct@3", 1.0, now + 2)])
|
|
assert db.series_names("host", "driveripper") == ["cpu.usage_pct", "gpu.busy_pct@3"]
|
|
|
|
|
|
def test_a_deleted_series_that_comes_back_is_a_new_series_again(db) -> None:
|
|
"""`delete_series` empties rows, so the triples it emptied have to be forgotten too.
|
|
|
|
Without that, re-creating a deleted series would leave `/api/metrics` answering without it for
|
|
a whole TTL window: the write matches a triple the store still believes exists.
|
|
"""
|
|
db.add("host", "deskwork", "gpu.busy_pct@0", 1.0)
|
|
assert db.series_names("host") == ["gpu.busy_pct@0"]
|
|
assert db.delete_series("host", "deskwork", "gpu.busy_pct@0") == 1
|
|
assert db.series_names("host") == []
|
|
db.add("host", "deskwork", "gpu.busy_pct@0", 2.0) # the same triple, now a live series again
|
|
assert db.series_names("host") == ["gpu.busy_pct@0"]
|
|
|
|
|
|
def test_the_series_write_tracking_is_capped(db, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""The existence set is memory, so it has a ceiling and a safe fallback below it.
|
|
|
|
Past the cap the store invalidates on every write — task 02's behaviour, correct and slower.
|
|
The cap is what keeps a namespace a caller can influence from becoming an unbounded structure.
|
|
"""
|
|
monkeypatch.setattr("dashboards.store.SERIES_TRACKED_SERIES_MAX", 4)
|
|
for index in range(30):
|
|
db.add("host", "deskwork", f"metric{index}", 1.0)
|
|
assert f"metric{index}" in db.series_names("host"), (
|
|
"giving up the tracking must never cost a name from the list")
|
|
assert len(db._known_series) <= 4, f"the tracking grew past its cap: {len(db._known_series)}"
|
|
with query_log(db) as seen:
|
|
db.add("host", "deskwork", "metric99", 1.0)
|
|
db.series_names("host")
|
|
assert [s for s in seen if "DISTINCT metric" in s], (
|
|
"past the cap the store must go back to deriving the list from the table")
|
|
|
|
|
|
def test_the_series_cache_is_bounded_however_many_names_a_caller_asks_for(db) -> None:
|
|
"""`/api/metrics?source=&host=` is free text, so the cache is capped, not grown.
|
|
|
|
The key space belongs to whoever sends the request; an unbounded dict keyed by it is a slow
|
|
memory leak with a query string on the front.
|
|
"""
|
|
for index in range(SERIES_CACHE_MAX_ENTRIES + 40):
|
|
db.series_names("host", f"host{index}")
|
|
assert len(db._series_cache) <= SERIES_CACHE_MAX_ENTRIES
|
|
|
|
|
|
def test_series_names_expires_on_the_clock(db) -> None:
|
|
"""The TTL is the ceiling on staleness when nothing wrote."""
|
|
db.add("host", "a", "cpu.usage_pct", 1.0)
|
|
db.series_names("host")
|
|
db.series_ttl_seconds = 0.0
|
|
with query_log(db) as seen:
|
|
db.series_names("host")
|
|
assert [s for s in seen if "DISTINCT metric" in s], "an expired entry must be re-derived"
|
|
|
|
|
|
def test_invalidate_read_caches_forces_both_answers_back_to_the_table(db) -> None:
|
|
"""The blunt instrument the cold measurement needs, pinned so it stays blunt.
|
|
|
|
`store_perf` asks for a cold `stats()`/`series_names()` against a store whose caches are warm;
|
|
if either survived that call, the phase's before/after would be timing a cache hit and calling
|
|
it a speed-up.
|
|
"""
|
|
db.add("host", "a", "cpu.usage_pct", 1.0)
|
|
assert db.series_names("host") == ["cpu.usage_pct"]
|
|
assert db.stats()["samples_total"] == 1
|
|
with query_log(db) as seen:
|
|
db.invalidate_read_caches()
|
|
assert seen == [], "invalidating is bookkeeping, not a query"
|
|
db.series_names("host")
|
|
db.stats()
|
|
assert [s for s in seen if "DISTINCT metric" in s], "the name list came back from the scan"
|
|
assert [s for s in seen if "COUNT(*) FROM sample" in s], "the counts did too"
|
|
|
|
|
|
def test_stats_answers_from_cache_until_the_ttl_expires(db) -> None:
|
|
"""`/api/health` is polled by every open tab; its two `COUNT` scans may run once per TTL.
|
|
|
|
The payload keys are a contract with deployment (`/api/health` is public), so this asserts
|
|
the same dict twice and the statements in between: nothing about the answer changes, only how
|
|
often the table is asked.
|
|
"""
|
|
db.add("host", "a", "cpu.usage_pct", 1.0)
|
|
db.log_event(Event(time.time(), "system", "a", "recording", "info", "up again"))
|
|
with query_log(db) as seen:
|
|
first = db.stats()
|
|
assert first["samples_total"] == 1 and first["events"] == 1
|
|
assert [s for s in seen if "COUNT(*) FROM sample" in s]
|
|
assert [s for s in seen if "COUNT(*) FROM event" in s]
|
|
|
|
seen.clear()
|
|
second = db.stats()
|
|
assert not [s for s in seen if "COUNT(*)" in s], (
|
|
"a cached stats() must not run a COUNT — that 1.3 s is the page load")
|
|
assert second == first
|
|
|
|
first["samples_by_res"][RES_RAW] = 999 # a caller must not be able to poison the cache
|
|
assert db.stats()["samples_by_res"] == second["samples_by_res"]
|
|
|
|
db.stats_ttl_seconds = 0.0
|
|
seen.clear()
|
|
db.add("host", "a", "cpu.usage_pct", 2.0)
|
|
third = db.stats()
|
|
assert [s for s in seen if "COUNT(*) FROM sample" in s], "the TTL, not a write, ages counts"
|
|
assert third["samples_total"] == 2
|
|
|
|
|
|
def test_stats_caches_the_counts_but_never_the_maintenance_timestamp(db) -> None:
|
|
"""`last_maintain` is how an operator spots a stalled rollup loop; caching it hides the bug.
|
|
|
|
The counts behind `/api/health` may be 30 s old. A `last_maintain` that never moves while the
|
|
page says `ok` is exactly the frozen-source failure AGENTS.md invariant 4 calls the one worse
|
|
than an outage, so it is read on every call and this test keeps it that way.
|
|
"""
|
|
db.add("host", "a", "cpu.usage_pct", 1.0)
|
|
assert db.stats()["last_maintain"] is None
|
|
with query_log(db) as seen:
|
|
db.maintain(now=1_700_000_000.0)
|
|
seen.clear()
|
|
stats = db.stats()
|
|
assert stats["last_maintain"] == 1_700_000_000.0
|
|
assert not [s for s in seen if "COUNT(*)" in s], "the counts stay cached; only meta is re-read"
|
|
assert [s for s in seen if "FROM meta" in s]
|
|
|
|
|
|
def test_pick_res_still_picks_the_finest_resolution_that_fits(db) -> None:
|
|
"""The capped count must decide exactly what a full `COUNT(*)` decided.
|
|
|
|
Cap at `max_points + 1` keeps both predicates — `count == 0` and `count <= max_points` —
|
|
identical, so this is a behaviour pin and not a tolerance.
|
|
"""
|
|
now = time.time()
|
|
# raw over budget, minutes under it: the answer is 1m whichever way the rows are counted
|
|
add_series(db, "host", "h", "m", range(900), start=now - 900, step=1)
|
|
add_series(db, "host", "h", "m", range(30), start=now - 3_600, step=60, res=RES_MIN)
|
|
assert db.pick_res("host", "h", "m", span=3_600.0, max_points=600) == RES_MIN
|
|
# nothing fits, so the finest resolution that has *any* row wins — an empty one reads as a
|
|
# dead sensor, which is the rule the pre-cap implementation had and the cap must not move
|
|
assert db.pick_res("host", "h", "m", span=3_600.0, max_points=20) == RES_RAW
|
|
assert db.pick_res("host", "h", "absent", span=3_600.0, max_points=600) == RES_HOUR
|
|
|
|
|
|
def test_the_resolution_count_is_one_statement_of_index_seeks(db) -> None:
|
|
"""One statement for all three resolutions, and its plan is a seek, not a scan."""
|
|
now = time.time()
|
|
for host in ("a", "b"):
|
|
add_series(db, "host", host, "m", range(50), start=now - 300, step=1)
|
|
with query_log(db) as seen:
|
|
db.pick_res_many("host", ["a", "b"], ["m", "absent"], span=3_600.0, max_points=600)
|
|
statements = [s for s in seen if "UNION ALL" in s]
|
|
assert len(statements) == 1, (
|
|
f"4 series x 3 resolutions must be one grouped round trip, saw {seen}")
|
|
# the trace callback hands back the statement with its bindings expanded, which is the exact
|
|
# text that ran — so the plan below is the plan the request paid, not a plan for a lookalike
|
|
plan = plan_of(db, statements[0])
|
|
assert "COVERING INDEX" in plan, plan
|
|
assert "SCAN sample" not in plan, f"the count fell back to a table scan: {plan}"
|
|
|
|
|
|
def test_pick_res_many_agrees_with_pick_res_for_every_pair(db) -> None:
|
|
now = time.time()
|
|
hosts = ["driveripper", "deskwork", "gateway"]
|
|
metrics = ["cpu.usage_pct", "mem.used_pct", "gpu.busy_pct@0", "never.written"]
|
|
for host in hosts:
|
|
if host == "gateway":
|
|
continue # the fleet's GPU-less box: a whole host with no gpu.* rows
|
|
for metric in metrics[:3]:
|
|
add_series(db, "host", host, metric, range(40), start=now - 400, step=1)
|
|
batched = db.pick_res_many("host", hosts, metrics, span=3_600.0, max_points=600)
|
|
assert batched == {(host, metric): db.pick_res("host", host, metric, 3_600.0, 600)
|
|
for host in hosts for metric in metrics}
|
|
assert batched[("gateway", "gpu.busy_pct@0")] == RES_HOUR, "no rows anywhere: the documented default"
|
|
assert db.pick_res_many("host", [], metrics, span=3_600.0, max_points=600) == {}
|
|
assert db.pick_res_many("host", hosts, [], span=3_600.0, max_points=600) == {}
|
|
|
|
|
|
def test_history_many_answers_exactly_what_history_answers_per_series(db) -> None:
|
|
"""The batched door and the single-series door are the same answer, series for series.
|
|
|
|
The shapes a chart depends on are in here too: the raw fallback while rollups lag, the
|
|
thinning of an over-budget window, and the newest point surviving both.
|
|
"""
|
|
now = time.time()
|
|
hosts = ["driveripper", "deskwork", "gateway"]
|
|
metrics = ["cpu.usage_pct", "mem.used_pct", "never.written"]
|
|
for host in hosts:
|
|
for index, metric in enumerate(metrics):
|
|
add_series(db, "host", host, metric, range(30 + index), start=now - 900, step=30)
|
|
# one series with rollups as well, so the resolution decision (not just the fetch) is compared
|
|
add_series(db, "host", "driveripper", "cpu.usage_pct", range(20), start=now - 7_200,
|
|
step=300, res=RES_MIN)
|
|
# one crowded series, for the thinning contract below
|
|
add_series(db, "host", "driveripper", "crowded.series", range(100), start=now - 1_000, step=10)
|
|
|
|
for span, max_points in ((3_600.0, 600), (3_600.0, 12), (30.0, 600), (90.0, 600)):
|
|
batched = db.history_many("host", hosts, [*metrics, "crowded.series"], span=span,
|
|
max_points=max_points)
|
|
per_series = {(host, metric): db.history("host", host, metric, span=span,
|
|
max_points=max_points)
|
|
for host in hosts for metric in [*metrics, "crowded.series"]}
|
|
assert batched == per_series, f"the batch changed a series (span={span}, points={max_points})"
|
|
|
|
# the tail contract survives batching: a thinned series is ascending, inside the budget, and
|
|
# still ends on the newest row the window holds — the point a chart cannot afford to lose
|
|
thinned = db.history_many("host", ["driveripper"], ["crowded.series"], span=3_600.0,
|
|
max_points=12)[("driveripper", "crowded.series")]
|
|
full = db.history("host", "driveripper", "crowded.series", span=3_600.0, max_points=600)
|
|
# `<= max_points + 1`, not `<= max_points`: the thinning has always been allowed to append the
|
|
# newest point back after striding, so a series can land one over budget. That quirk is part
|
|
# of the answer the endpoints must keep byte-identically, not something this task fixes.
|
|
assert len(full) == 100 and 12 <= len(thinned) <= 13
|
|
assert [ts for ts, _ in thinned] == sorted(ts for ts, _ in thinned)
|
|
assert thinned[-1] == full[-1], "the batch must not lose 'now' any more than the loop did"
|
|
assert thinned[0] > full[0], "the head is tail-truncated — the documented `tail=True` semantic"
|
|
|
|
assert db.history_many("host", [], ["cpu.usage_pct"]) == {}
|
|
# a series nobody ever wrote answers as an empty list, per pair — the contract every history
|
|
# endpoint shows the browser as an empty plot rather than a 404
|
|
assert db.history_many("host", ["gateway"], ["still.unwritten"]) == {
|
|
("gateway", "still.unwritten"): []}
|
|
# duplicate names in a request cost one query each, and still answer once per pair
|
|
assert db.history_many("host", ["gateway", "gateway"], ["cpu.usage_pct"]) == db.history_many(
|
|
"host", ["gateway"], ["cpu.usage_pct"])
|
|
|
|
|
|
def test_history_many_batches_the_decision_and_not_the_rows(db) -> None:
|
|
"""One grouped statement for the resolution of the whole batch, whatever its size.
|
|
|
|
Recorded deliberately: folding the *rows* into one statement (a window function over the
|
|
range, or a compound of the per-series tail selects) measured slower on the seeded store, so
|
|
the batch is the decision and the fetch stays a per-series index seek. This pins the half
|
|
that did get batched, so a future edit cannot quietly return to three counts per series.
|
|
"""
|
|
now = time.time()
|
|
hosts = [f"host{index}" for index in range(20)]
|
|
for host in hosts:
|
|
add_series(db, "host", host, "cpu.usage_pct", range(10), start=now - 300, step=30)
|
|
with query_log(db) as seen:
|
|
db.history_many("host", hosts, ["cpu.usage_pct", "mem.used_pct"], span=3_600.0,
|
|
max_points=600)
|
|
grouped = [s for s in seen if "UNION ALL" in s]
|
|
# 20 hosts x 2 metrics x 3 resolutions = 120 branches = two chunks of COMPOUND_CHUNK
|
|
assert len(grouped) == -(-120 // COMPOUND_CHUNK), (
|
|
f"expected one grouped count per {COMPOUND_CHUNK} branches, saw {grouped}")
|
|
|
|
|
|
# ---- phase 14 task 03: the rollup window, and the WAL it writes into ------------------- #
|
|
#
|
|
# `maintain()` used to average every row its source resolution retains, on every tick, inside one
|
|
# write transaction: 3.97 M rows of 1-minute history per minute. Task 01 measured that at 5.4 s,
|
|
# and at 5 of 9 concurrent poller inserts outliving `busy_timeout=5000` — the
|
|
# `store-error: database is locked` line in AGENTS.md known gap 4. The fix is a window per step,
|
|
# and the tests below are the two halves of "a window that is only cheaper": the bound is read out
|
|
# of the statements that actually ran (never recomputed from the same arithmetic under test), and
|
|
# the values come out identical to a whole-table rollup of the same rows.
|
|
|
|
#: The `(ceiling, floor)` a rollup statement ran with, pulled out of `sqlite3`'s trace text.
|
|
ROLLUP_WINDOW = re.compile(
|
|
r"FROM sample WHERE res='(?P<res>\w+)' AND ts < (?P<ceiling>[\d.]+) AND ts >= (?P<floor>[\d.]+)")
|
|
|
|
|
|
def rollup_windows(statements: list[str]) -> dict[str, tuple[float, float]]:
|
|
"""`(ceiling, floor)` per rollup step, keyed by the resolution it read from.
|
|
|
|
The trace callback hands back the statement with its bindings expanded, so this is the window
|
|
the statement ran with — the task's "assert via the EXPLAIN/row count hook, or a sqlite3
|
|
trace", taken literally. Keying on the source resolution is what tells the minute step
|
|
(`res='raw'`) from the hour step (`res='1m'`).
|
|
"""
|
|
found: dict[str, tuple[float, float]] = {}
|
|
for statement in statements:
|
|
match = ROLLUP_WINDOW.search(statement)
|
|
if match:
|
|
found[match["res"]] = (float(match["ceiling"]), float(match["floor"]))
|
|
return found
|
|
|
|
|
|
def bucket_start(ts: float, bucket: float = MINUTE) -> float:
|
|
"""The aligned bucket a clock falls in — the shape every window floor is snapped to."""
|
|
return math.floor(ts / bucket) * bucket
|
|
|
|
|
|
#: Wall clocks for the window-bound tests. `1_800_000_000` is exactly on an hour *and* a minute
|
|
#: boundary, so these walk the interesting cases: on it, one second past, half an hour past, and one
|
|
#: second short of the next hour. A minute-aligned bound and an hour-aligned one are the same number
|
|
#: at offset 0 and 1 980 s apart at offset 1 799, which is exactly the luck this test used to run on
|
|
#: (`pytest.approx`'s relative tolerance hid up to 1 800 s of misalignment and not more) — an
|
|
#: explicit clock makes every edge a case instead of a coin flip.
|
|
WINDOW_CLOCK_OFFSETS = (0, 1, 30, 59, 1_800, 3_539, 3_599)
|
|
|
|
|
|
def test_the_window_floor_is_always_aligned_to_its_own_bucket(db) -> None:
|
|
"""`_rollup_floor` snaps to the step's bucket, at every clock, and never above its bound.
|
|
|
|
The alignment is not cosmetic: the ceiling of a rollup step is a *row* bound, so the newest
|
|
bucket is a partial one that later passes re-roll, and a floor that lands inside a bucket makes
|
|
that re-roll average its own tail rows. Asserting the property over offsets — rather than one
|
|
number at one instant — is what keeps a future edit from re-fitting the test to whichever clock
|
|
it was written on.
|
|
"""
|
|
for bucket in (MINUTE, HOUR):
|
|
for offset in WINDOW_CLOCK_OFFSETS:
|
|
clock = 1_800_000_000.0 + offset
|
|
floor = db._rollup_floor(clock, 0.0, bucket)
|
|
assert floor % bucket == 0, f"bucket={bucket:g} offset={offset} floor={floor}"
|
|
assert floor <= clock - bucket - MAINTAIN_SLACK_SECONDS, "never above the bound"
|
|
assert floor > clock - (2 * bucket) - MAINTAIN_SLACK_SECONDS, "never a whole window back"
|
|
# …and the two steps really do use different alignments, or the test above proves nothing
|
|
clock = 1_800_000_000.0 + 1_799
|
|
assert db._rollup_floor(clock, 0.0, HOUR) != db._rollup_floor(clock, 0.0, MINUTE)
|
|
|
|
|
|
@pytest.mark.parametrize("hour_offset", WINDOW_CLOCK_OFFSETS)
|
|
def test_the_first_rollup_pass_scans_the_retention_it_has(db, hour_offset: int) -> None:
|
|
"""No `last_maintain` row means nothing is known, so the window is the whole retention.
|
|
|
|
This is the "a first run on a fresh DB scans the full retention window exactly as today"
|
|
half of the task, and it is also the restore-from-backup case: a store opened on history it
|
|
has never rolled has to be able to reach 47-hour-old raw rows, or its charts above the last
|
|
minute stay empty forever.
|
|
"""
|
|
now = 1_800_000_000.0 + hour_offset
|
|
deep = bucket_start(now - 47 * HOUR)
|
|
add_series(db, "host", "h", "deep.metric", [7], start=deep, step=1)
|
|
with query_log(db) as seen:
|
|
db.maintain(now=now)
|
|
windows = rollup_windows(seen)
|
|
assert windows[RES_RAW] == pytest.approx(
|
|
(now - MINUTE - MAINTAIN_SLACK_SECONDS, now - db.raw_keep_seconds))
|
|
assert windows[RES_MIN] == pytest.approx(
|
|
(now - HOUR - MAINTAIN_SLACK_SECONDS, now - db.minute_keep_seconds))
|
|
assert db.query("host", "h", "deep.metric", res=RES_MIN, start=deep - 1,
|
|
end=deep + MINUTE) == [(deep, 7.0)], (
|
|
"a first pass cannot reach the deep history it retains")
|
|
|
|
|
|
@pytest.mark.parametrize("hour_offset", WINDOW_CLOCK_OFFSETS)
|
|
def test_a_second_maintain_reads_only_the_window_since_the_last_run(db, hour_offset: int) -> None:
|
|
"""The cost change, pinned as a bound and as behaviour.
|
|
|
|
A bucket that filled after the previous pass is inside the next window and gets rolled. A raw
|
|
row that arrives for a bucket ten minutes behind it is not, and that bucket keeps the average
|
|
the earlier pass computed — the documented cost of a window, stated here as a passing test
|
|
rather than as a surprise. Reaching it in production costs a poller stall measured in minutes
|
|
(the poller stamps samples with its own wall clock); the alternative it replaced was
|
|
re-averaging 3.97 M rows every minute.
|
|
"""
|
|
now = 1_800_000_000.0 + hour_offset
|
|
bucket_a = bucket_start(now - 10 * MINUTE)
|
|
add_series(db, "host", "h", "m", [10, 20], start=bucket_a, step=1)
|
|
db.maintain(now=now)
|
|
assert db.query("host", "h", "m", res=RES_MIN, start=bucket_a - 1,
|
|
end=bucket_a + MINUTE) == [(bucket_a, 15.0)]
|
|
|
|
second = now + 60
|
|
# Two full minutes behind the ceiling, so both rows of the pair are always inside the next
|
|
# window: with the bucket only one step behind, a clock that lands within a second of the
|
|
# boundary puts the pair's second row above the ceiling and the bucket rolls 40.0 instead.
|
|
bucket_b = bucket_start(second - 3 * MINUTE)
|
|
add_series(db, "host", "h", "m", [40, 50], start=bucket_b, step=1)
|
|
add_series(db, "host", "h", "m", [1000], start=bucket_a + 30, step=1)
|
|
with query_log(db) as seen:
|
|
counts = db.maintain(now=second)
|
|
|
|
windows = rollup_windows(seen)
|
|
# The bound: one minute (one hour) of rows plus the overlap, not 48 h of raw (not 30 d of 1m).
|
|
# Each floor is snapped to *its own step's* bucket — the minute step to a minute, the hour step
|
|
# to an hour — which is what `hour_offset` walks, and both are far inside retention.
|
|
assert windows[RES_RAW] == pytest.approx(
|
|
(second - MINUTE - MAINTAIN_SLACK_SECONDS,
|
|
bucket_start(now - MINUTE - MAINTAIN_SLACK_SECONDS, MINUTE)))
|
|
assert windows[RES_MIN] == pytest.approx(
|
|
(second - HOUR - MAINTAIN_SLACK_SECONDS,
|
|
bucket_start(now - HOUR - MAINTAIN_SLACK_SECONDS, HOUR)))
|
|
assert windows[RES_RAW][1] > second - db.raw_keep_seconds
|
|
assert windows[RES_MIN][1] > second - db.minute_keep_seconds
|
|
# …and the behaviour that bound implies, in both directions
|
|
assert db.query("host", "h", "m", res=RES_MIN, start=bucket_b - 1,
|
|
end=bucket_b + MINUTE) == [(bucket_b, 45.0)], "the due bucket must be rolled"
|
|
assert db.query("host", "h", "m", res=RES_MIN, start=bucket_a - 1,
|
|
end=bucket_a + MINUTE) == [(bucket_a, 15.0)], (
|
|
"the row behind the window must not rewrite an older bucket")
|
|
assert counts["to_minute"] == 1, f"only the due bucket was written: {counts}"
|
|
assert db.get_meta(LAST_MAINTAIN_KEY) == str(second)
|
|
|
|
|
|
@pytest.mark.parametrize("hour_offset", WINDOW_CLOCK_OFFSETS)
|
|
def test_a_missed_interval_is_a_catch_up_scan_not_a_gap(db, hour_offset: int) -> None:
|
|
"""Twelve minutes with no pass, then one: every bucket in between is rolled.
|
|
|
|
The window reaches one bucket plus slack *behind* the previous pass's ceiling rather than
|
|
starting at it, and that overlap is the whole difference between "the app was stopped for
|
|
twelve minutes" and "twelve minutes of charts are silently empty". One row is placed
|
|
deliberately inside the overlap: above the first pass's ceiling, below the clock the first pass
|
|
recorded, so a bound of exactly `last_maintain` would skip it for ever.
|
|
"""
|
|
# An explicit clock, for the same reason as the test above: the overlap row's bucket and the
|
|
# window's floor are both snapped, and the count that proves the catch-up happened depends on
|
|
# which side of a boundary they land on.
|
|
first_at = 1_800_000_000.0 + hour_offset - 20 * MINUTE
|
|
db.maintain(now=first_at) # an otherwise empty store: this only records the clock
|
|
aligned = bucket_start(first_at)
|
|
for index in range(10):
|
|
add_series(db, "host", "h", "m", [float(index)], start=aligned + index * MINUTE + 5, step=1)
|
|
inside_the_overlap = first_at - 100
|
|
add_series(db, "host", "h", "m", [99], start=inside_the_overlap, step=1)
|
|
|
|
second = first_at + 12 * MINUTE
|
|
counts = db.maintain(now=second)
|
|
# `end` is explicit because `query` defaults it to the wall clock, and this test's clock is its
|
|
# own: a default would silently answer "no rows" for buckets that are in the store.
|
|
minutes = db.query("host", "h", "m", res=RES_MIN, start=0, end=second)
|
|
assert counts["to_minute"] == 11, f"ten buckets plus the overlap row: {counts}"
|
|
assert [value for _ts, value in minutes] == [99.0, *[float(step) for step in range(10)]], (
|
|
"a missed interval must be caught up, not skipped")
|
|
assert minutes[0][0] <= inside_the_overlap, "the overlap row has its bucket"
|
|
|
|
|
|
@pytest.mark.parametrize("sub_second", [0.0, 0.5, 30.0, 59.5])
|
|
@pytest.mark.parametrize("hour_offset", [0, 600, 1800, 3_540])
|
|
def test_windowed_rollups_answer_exactly_what_one_full_window_rollup_answers(
|
|
tmp_path, hour_offset: int, sub_second: float) -> None:
|
|
"""The equality the task asks for: the cost moves, not one averaged value.
|
|
|
|
Two stores, the same rows. One is maintained twice with an hour between the passes, so its
|
|
second pass reads a window; the other gets every row up front and is maintained once over the
|
|
whole retention at the later clock — the shape task 01 measured. Every row of every resolution
|
|
must match, `ts` and `value` exactly. `INSERT OR REPLACE` over a complete bucket is what makes
|
|
that safe, and the bucket-aligned floor (`_rollup_floor`) is what makes it true: a floor that
|
|
cut into a bucket would instead leave the *tail* rows of a split minute as its "average".
|
|
"""
|
|
# `first_at` is parametrised across an hour on purpose: the windows land on bucket boundaries
|
|
# or in the middle of them depending on where the clock fell, and the boundary cases are the
|
|
# ones a partial bucket can hide in.
|
|
first_at = bucket_start(time.time() - 6 * HOUR, HOUR) + hour_offset + sub_second
|
|
second_at = first_at + HOUR + 90.0 # an hour and a half: straddles an hour bucket boundary
|
|
|
|
def seed(store: SampleStore, *, late: bool) -> None:
|
|
for series, base in (("cpu.usage_pct", 0.0), ("mem.used_pct", 100.0)):
|
|
# three hours of raw at the fleet's 10 s cadence, all of it complete by `first_at`
|
|
add_series(store, "host", "h", series,
|
|
[base + step for step in range(1_080)],
|
|
start=first_at - 4 * HOUR, step=10)
|
|
if late:
|
|
# the buckets that fill in the hour between the two passes, unaligned on purpose
|
|
for series, base in (("cpu.usage_pct", 0.0), ("mem.used_pct", 100.0)):
|
|
add_series(store, "host", "h", series,
|
|
[base + step for step in range(180)],
|
|
start=first_at + 95.0, step=20)
|
|
|
|
def dump(store: SampleStore) -> list[tuple[str, str, str, str, float, float]]:
|
|
with store._lock, store._db:
|
|
return store._db.execute(
|
|
"SELECT source, host, metric, res, ts, value FROM sample"
|
|
" ORDER BY source, host, metric, res, ts").fetchall()
|
|
|
|
windowed = SampleStore(tmp_path / "windowed.sqlite3")
|
|
reference = SampleStore(tmp_path / "reference.sqlite3")
|
|
try:
|
|
seed(windowed, late=False)
|
|
windowed.maintain(now=first_at)
|
|
seed(windowed, late=True)
|
|
windowed.maintain(now=second_at)
|
|
windowed.maintain(now=second_at + 60) # a third tick, the steady state the app lives in
|
|
|
|
seed(reference, late=True)
|
|
assert reference.get_meta(LAST_MAINTAIN_KEY) is None, "a fresh store is a full window"
|
|
reference.maintain(now=second_at + 60)
|
|
|
|
windowed_rows, reference_rows = dump(windowed), dump(reference)
|
|
assert {row[3] for row in windowed_rows} == {RES_RAW, RES_MIN, RES_HOUR}, (
|
|
f"the comparison needs rolled rows on both steps: {sorted({r[3] for r in windowed_rows})}")
|
|
assert len([row for row in windowed_rows if row[3] == RES_MIN]) > 200, (
|
|
"the equality has to cover a lot of buckets, not a token few")
|
|
assert windowed_rows == reference_rows, (
|
|
"the window changed a rollup value (or a bucket a full scan would have written)")
|
|
finally:
|
|
windowed.close()
|
|
reference.close()
|
|
|
|
|
|
def test_a_re_rolled_bucket_only_ever_gains_rows(db) -> None:
|
|
"""The progressive correction the whole-table scan used to provide still happens.
|
|
|
|
A pass whose ceiling lands mid-bucket writes a partial average — as it always did — and the
|
|
next pass must re-read that bucket **from its own start** and publish the full one. A window
|
|
floor that cut into the bucket instead would re-read its tail and overwrite 2.5 with 4.5, which
|
|
is an invented number wearing a real chart's clothes (AGENTS.md rule 2).
|
|
"""
|
|
minute = bucket_start(time.time() - 10 * MINUTE)
|
|
add_series(db, "host", "h", "m", [0, 1, 2, 3, 4, 5], start=minute, step=10)
|
|
first = minute + 155.0 # ceiling = minute + 35: four of the six rows are in the window
|
|
db.maintain(now=first)
|
|
partial = db.query("host", "h", "m", res=RES_MIN, start=minute - 1, end=minute + MINUTE)
|
|
assert partial == [(minute, 1.5)], "a boundary bucket is rolled from its start, partially"
|
|
|
|
db.maintain(now=first + 60)
|
|
rolled = db.query("host", "h", "m", res=RES_MIN, start=minute - 1, end=minute + MINUTE)
|
|
assert rolled == [(minute, 2.5)], (
|
|
f"the re-roll must add the remaining rows, not replace them: {rolled}")
|
|
db.maintain(now=first + 120)
|
|
assert db.query("host", "h", "m", res=RES_MIN, start=minute - 1,
|
|
end=minute + MINUTE) == rolled, "a settled bucket keeps its value"
|
|
|
|
|
|
def test_a_corrupt_last_maintain_row_falls_back_to_the_full_window(db, caplog) -> None:
|
|
"""A bound the store cannot read is the expensive answer, said once, never a crash.
|
|
|
|
`maintain()` is the only writer of that row, so a non-number in it means something outside the
|
|
app edited the database. Rolling the whole window again rewrites identical rows; guessing a
|
|
window from a value that is not a timestamp is how a rollup starts skipping buckets in
|
|
silence, and the maintenance loop swallowing an exception is worse still (invariant 4).
|
|
"""
|
|
now = time.time()
|
|
db.set_meta(LAST_MAINTAIN_KEY, "not-a-timestamp")
|
|
with caplog.at_level(logging.WARNING, logger="dashboards.store"), query_log(db) as seen:
|
|
db.maintain(now=now)
|
|
assert rollup_windows(seen)[RES_RAW] == pytest.approx(
|
|
(now - MINUTE - MAINTAIN_SLACK_SECONDS, now - db.raw_keep_seconds))
|
|
assert [rec for rec in caplog.records if "last_maintain" in rec.message], (
|
|
"drift in the row that bounds every rollup is worth one line")
|
|
assert "not-a-timestamp" not in caplog.text, "the log quotes the key, not the stranger's value"
|
|
assert db.get_meta(LAST_MAINTAIN_KEY) == str(now), "the pass repairs the row it reads"
|
|
|
|
|
|
def test_a_failing_optimize_is_a_log_line_and_not_a_failed_tick(db, monkeypatch, caplog) -> None:
|
|
"""Statistics are bookkeeping: `maintain()` may not turn them into an outage.
|
|
|
|
`_optimize` runs outside the rollup transaction, so the only way it can hurt is by raising — and
|
|
a `store-error` event for a planner hint is worse than a slightly slower query (AGENTS.md
|
|
invariant 4: a source that stops answering is the failure, and the maintenance loop's health is
|
|
reported the same way). The stub quotes a message of its own; the log line names the pragma and
|
|
the consequence, nothing else.
|
|
"""
|
|
class _Boom:
|
|
def execute(self, _sql: str) -> None:
|
|
raise sqlite3.Error("database is locked")
|
|
|
|
monkeypatch.setattr(db, "_db", _Boom())
|
|
with caplog.at_level(logging.WARNING, logger="dashboards.store"):
|
|
db._optimize() # must not raise
|
|
assert "PRAGMA optimize failed" in caplog.text
|
|
assert "database is locked" not in caplog.text, "the tick's own words, not the driver's"
|
|
|
|
|
|
def test_the_perf_probe_explains_the_statement_that_runs(db) -> None:
|
|
"""The probe's rollup `EXPLAIN` is derived from `ROLLUP_INSERT`, so its bindings must match.
|
|
|
|
A hand-copied lookalike is how a NOTES table ends up describing a statement nobody runs — and
|
|
when the rollup grew its seventh binding (the target resolution), the copy in the probe was the
|
|
thing that broke. Preparing the derived statement with the probe's own arity is the pin: retyping
|
|
it, or letting the two drift, raises here instead of in the middle of a measurement run.
|
|
"""
|
|
whole_select = "SELECT" + ROLLUP_INSERT.split("SELECT", 1)[1]
|
|
assert whole_select == store_perf.ROLLUP_SELECT
|
|
assert ROLLUP_INSERT.count("?") == store_perf.ROLLUP_SELECT.count("?") == 7
|
|
with db._lock:
|
|
db._db.execute("EXPLAIN QUERY PLAN " + store_perf.ROLLUP_SELECT,
|
|
(RES_MIN, MINUTE, MINUTE, RES_RAW, 1.0, 0.0, MINUTE)).fetchall()
|
|
|
|
|
|
def test_the_wal_pragmas_are_set_on_every_connection(db) -> None:
|
|
"""The WAL half of task 03, pinned as the settings that ship.
|
|
|
|
`journal_size_limit` is the lever: a WAL that grows past it is truncated to it at the next
|
|
checkpoint instead of staying at a 140 MB high-water mark (AGENTS.md known gap 4).
|
|
`wal_autocheckpoint` is SQLite's own default and is set so the knob is visible beside it, and
|
|
`busy_timeout` stays at 5 000 ms because the fix is a shorter write, not a longer wait.
|
|
All the pragmas are per connection, so this asserts them on a handle built by `SampleStore` and
|
|
not on whatever SQLite happens to have guessed.
|
|
"""
|
|
with db._lock:
|
|
assert db._db.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
|
|
assert db._db.execute("PRAGMA synchronous").fetchone()[0] == 1
|
|
assert db._db.execute("PRAGMA busy_timeout").fetchone()[0] == 5000
|
|
assert db._db.execute("PRAGMA wal_autocheckpoint").fetchone()[0] == WAL_AUTOCHECKPOINT_PAGES
|
|
assert db._db.execute("PRAGMA journal_size_limit").fetchone()[0] == JOURNAL_SIZE_LIMIT_BYTES
|
|
assert JOURNAL_SIZE_LIMIT_BYTES == 64 * 1024 * 1024 == 67108864
|
|
# `automatic_analyze` (off, so no poller's commit pays for a statistics refresh) is set-only in
|
|
# this build — reading it back returns no row — so its effect is pinned where it is observable,
|
|
# in `test_maintain_leaves_query_statistics`.
|
|
|
|
|
|
def test_maintain_leaves_query_statistics(db) -> None:
|
|
"""`maintain()` is the app's only statistics refresh, and the windowed rollup needs them.
|
|
|
|
Without `sqlite_stat1` the planner cannot price a range and reads the table for every window:
|
|
on the seeded store the windowed rollup cost 328 ms of `SCAN sample` matching no rows at all,
|
|
and 1 ms with `SEARCH sample USING INDEX sqlite_autoindex_sample_1` after one analyse. So the
|
|
statistics are half of "cheaper for the same rows", and they are refreshed here rather than by
|
|
SQLite's own automatic ANALYZE — which would run inside whichever poller's commit tripped its
|
|
change counter, the exact path this task is shortening. On a database this size a scan is free,
|
|
so the assertion is that the statistics exist, not what plan they produce; the plan is pinned at
|
|
production scale in the `store_perf` report.
|
|
"""
|
|
def stat_rows() -> int:
|
|
with db._lock:
|
|
if db._db.execute("SELECT 1 FROM sqlite_master WHERE name='sqlite_stat1'").fetchone() is None:
|
|
return 0
|
|
return int(db._db.execute("SELECT COUNT(*) FROM sqlite_stat1").fetchone()[0])
|
|
|
|
assert stat_rows() == 0, "a store that has never been maintained has no statistics"
|
|
db.add("host", "gateway", "cpu.usage_pct", 1.0)
|
|
db.maintain(now=time.time())
|
|
assert stat_rows() > 0, "maintenance is the one place the planner's statistics get refreshed"
|
|
|
|
|
|
# ---- phase 14: the production-shaped store the performance work measures ---- #
|
|
#
|
|
# `TODO.md:7` is "pages take forever to load and seem to lock up the database", and a claim
|
|
# about speed needs a measured "before". The fixture below is that before: an on-disk store with
|
|
# the deployed retention semantics (`raw_keep_hours=48`, `minute_keep_days=30`,
|
|
# `hour_keep_days=400` — the `SampleStore` defaults, i.e. what a 90 d chart really reads is 1m
|
|
# and 1h, never raw beyond 48 h), the fleet's real metric names, the real 10 s cadence, ~2 000
|
|
# events and the `meta` rows the app writes. ~6.4 M rows in ~700 MB, which is the dev DB's order
|
|
# of magnitude (`AGENTS.md` gap 4 records 636 MB / 3.4 M rows for a store only days old).
|
|
#
|
|
# Why the fixture is not the dev DB: the testing policy keeps `uv run pytest` hermetic, and a
|
|
# fixture that read the operator's live history would change behaviour when that history does.
|
|
# Why it is not an in-memory store: `:memory:` hides the WAL and the lock behaviour, which are
|
|
# half of what this phase measures. Why it is not `tmp_path`: pytest's tmp dir on this box is
|
|
# `tmpfs`, i.e. RAM, and a database timed on RAM is a flattering number nobody deploys —
|
|
# `store_perf.cache_dir()` defaults to a real block device and says so in its report.
|
|
#
|
|
# The default-run tests here assert **shape and determinism**; the wall-time budgets task 04 pins
|
|
# are behind `-m store_perf --store-perf` (and in `scripts/store_perf_probe.py`, which writes the
|
|
# NOTES tables) because a budget asserted on first writing is a number fitted to the machine it was
|
|
# written on. The budgets themselves live in `store_perf.BUDGETS` — one table, priced by the probe,
|
|
# the report, and the two budget tests below, because three copies of a budget drift the moment one
|
|
# is relaxed. The `perf_store`/`perf_store_path` fixtures are in `tests/conftest.py`: the HTTP
|
|
# budgets in `tests/test_api.py` measure the same file.
|
|
|
|
|
|
|
|
|
|
def test_the_seed_seeds_the_retention_the_app_actually_keeps() -> None:
|
|
"""The fixture's retention must be `SampleStore`'s own defaults, not a story about them.
|
|
|
|
A budget measured on a store seeded to a *different* retention is a budget for a database
|
|
that does not exist: `pick_res`/`history` costs depend on how deep each resolution goes.
|
|
"""
|
|
fresh = SampleStore(":memory:")
|
|
try:
|
|
assert fresh.raw_keep_seconds == pytest.approx(store_perf.RAW_KEEP_HOURS * HOUR)
|
|
assert fresh.minute_keep_seconds == pytest.approx(store_perf.MINUTE_KEEP_DAYS * 24 * HOUR)
|
|
assert fresh.hour_keep_seconds == pytest.approx(store_perf.HOUR_KEEP_DAYS * 24 * HOUR)
|
|
finally:
|
|
fresh.close()
|
|
assert store_perf.CADENCE_SECONDS == 10.0, "the raw window's cadence is HOST_INTERVAL_SECONDS"
|
|
|
|
|
|
def test_the_seeded_store_is_the_shape_the_budget_tests_measure(perf_store: SampleStore) -> None:
|
|
"""Row counts, series, retention windows, events, meta — asserted, never timed."""
|
|
now = store_perf.seed_now(perf_store)
|
|
expected = store_perf.plan(now)
|
|
|
|
with perf_store._lock, perf_store._db: # one pass over the table, not three
|
|
rows = {row[0]: row for row in perf_store._db.execute(
|
|
"SELECT res, COUNT(*), MIN(ts), MAX(ts) FROM sample WHERE source='host' GROUP BY res")}
|
|
|
|
assert expected.raw_rows == expected.series_total * expected.raw_steps
|
|
assert expected.minute_rows_seeded == expected.series_total * expected.minute_steps
|
|
assert expected.hour_rows_seeded == expected.series_total * expected.hour_steps
|
|
assert sum(len(store_perf.HOST_SERIES[name]) for name in expected.hosts) == (
|
|
expected.series_total)
|
|
assert rows[RES_RAW][1] == expected.raw_rows, (
|
|
"the raw window is inserted, not rolled: its count is the one number the seed owns "
|
|
"outright, so a mismatch here means the shape drifted from the plan")
|
|
assert rows[RES_MIN][1] >= expected.minute_rows_seeded
|
|
assert rows[RES_HOUR][1] >= expected.hour_rows_seeded
|
|
# …and the rollups `maintain()` added inside the windows are there too, or the fixture
|
|
# would be a store whose charts draw nothing above the raw window
|
|
assert rows[RES_MIN][1] > expected.minute_rows_seeded
|
|
assert rows[RES_HOUR][1] > expected.hour_rows_seeded
|
|
|
|
# Retention: nothing lives past its resolution's window, and each window is deep enough to
|
|
# be the one the pages read (a 90 d chart reads 1m/1h; a 5 m chart reads raw).
|
|
assert rows[RES_RAW][2] >= now - store_perf.RAW_KEEP_HOURS * HOUR - 1.0
|
|
assert rows[RES_RAW][3] <= now + 1.0
|
|
assert rows[RES_MIN][2] >= now - store_perf.MINUTE_KEEP_DAYS * 24 * HOUR - 1.0
|
|
assert rows[RES_HOUR][2] >= now - store_perf.HOUR_KEEP_DAYS * 24 * HOUR - 1.0
|
|
|
|
assert perf_store.hosts("host") == list(expected.hosts)
|
|
assert len(perf_store.series_names("host")) == len(store_perf.BASE_SERIES) + len(
|
|
store_perf.GPU_SERIES), "the union: gateway has no gpu.* series, the other two do"
|
|
assert len(perf_store.series_names("host", "gateway")) == len(store_perf.BASE_SERIES)
|
|
assert len(perf_store.series_names("host", "driveripper")) == (
|
|
len(store_perf.BASE_SERIES) + len(store_perf.GPU_SERIES))
|
|
|
|
assert perf_store.count() == rows[RES_RAW][1] + rows[RES_MIN][1] + rows[RES_HOUR][1]
|
|
|
|
events = perf_store.recent_events(store_perf.EVENT_ROWS)
|
|
assert len(events) == store_perf.EVENT_ROWS
|
|
assert {event.source for event in events} == {"system"}
|
|
|
|
# The `meta` rows the app writes: `last_maintain` from the build's own `maintain()` call,
|
|
# plus this fixture's own provenance. Phase 11's auth rows are not seeded — the app writes
|
|
# them the first time a test boots a server on this store.
|
|
assert perf_store.get_meta("last_maintain") is not None
|
|
assert perf_store.get_meta(store_perf.META_SEED_VERSION) == store_perf.SEED_VERSION
|
|
|
|
# Magnitude, not exact bytes: a fixture two orders of magnitude smaller would not catch a
|
|
# full-table scan, and the number that matters is in the NOTES (dev DB 636 MB, this ~700 MB).
|
|
size = Path(perf_store.path).stat().st_size
|
|
assert size > 100 * 1024 * 1024, f"seeded store is {size / 1e6:.0f} MB — too small to measure"
|
|
|
|
|
|
def test_the_rollups_in_the_seed_came_from_the_raw_rows_and_run_twice_the_same(
|
|
perf_store: SampleStore,
|
|
) -> None:
|
|
"""Determinism, in both directions the phase depends on.
|
|
|
|
1. the seed is a pure function of `(shape, seed clock)`, so a rebuild is byte-identical and
|
|
a "before/after" comparison is comparing the same data — asserted by recomputing stored
|
|
raw values from `store_perf._value`;
|
|
2. the rollup rows are the real `maintain()` arithmetic over those raw values (the mean of
|
|
the raw samples in the bucket), and running maintenance again changes nothing.
|
|
"""
|
|
now = store_perf.seed_now(perf_store)
|
|
host, metric = "driveripper", "cpu.usage_pct"
|
|
|
|
first_rows = perf_store.query("host", host, metric, res=RES_RAW,
|
|
start=now - store_perf.RAW_KEEP_HOURS * HOUR,
|
|
limit=60, tail=False)
|
|
assert first_rows, "the raw window is empty"
|
|
offset = round((first_rows[0][0] - (now - store_perf.RAW_KEEP_HOURS * HOUR))
|
|
/ store_perf.CADENCE_SECONDS)
|
|
assert [value for _ts, value in first_rows] == [
|
|
store_perf.seed_value(metric, offset + index) for index in range(len(first_rows))], (
|
|
"stored raw values are not the seed function's output — the seed is no longer a pure "
|
|
"function of the shape, so before/after numbers would not be comparable")
|
|
|
|
# one complete bucket inside the raw window, recomputed by hand
|
|
bucket_ts = first_rows[0][0] + 30 * store_perf.CADENCE_SECONDS
|
|
bucket_ts = int(bucket_ts // MINUTE) * MINUTE
|
|
raw_in_bucket = perf_store.query("host", host, metric, res=RES_RAW, start=bucket_ts,
|
|
end=bucket_ts + MINUTE - 0.001)
|
|
assert len(raw_in_bucket) == 6, raw_in_bucket
|
|
rolled = perf_store.query("host", host, metric, res=RES_MIN, start=bucket_ts,
|
|
end=bucket_ts + MINUTE - 0.001)
|
|
assert rolled == [(float(bucket_ts), pytest.approx(
|
|
sum(value for _ts, value in raw_in_bucket) / len(raw_in_bucket)))]
|
|
|
|
def shape() -> dict[str, tuple[int, float, float]]:
|
|
"""Rows, and the time window each resolution holds — the exact half of the checksum.
|
|
|
|
The total of `value` is deliberately *not* in here. `maintain()` writes rollups with
|
|
`INSERT OR REPLACE`, which deletes and re-inserts the row, so a pass that changes no value
|
|
still moves every rowid it touched, and a `SUM(value)` scanned in rowid order accumulates
|
|
those doubles in a different sequence — a correctly idempotent pass then prints a
|
|
different total (2.8e5 out of 2.5e15 was how this test failed once the seeded file had
|
|
been maintained a few times). Counts and windows are exact; the sum is compared to a
|
|
tolerance below, and the *values* are compared exactly, in primary-key order, by
|
|
`values_slice()`.
|
|
"""
|
|
with perf_store._lock, perf_store._db:
|
|
return {row[0]: (row[1], row[2], row[3]) for row in perf_store._db.execute(
|
|
"SELECT res, COUNT(*), COALESCE(MIN(ts),0), COALESCE(MAX(ts),0)"
|
|
" FROM sample GROUP BY res")}
|
|
|
|
def totals() -> dict[str, float]:
|
|
with perf_store._lock, perf_store._db:
|
|
return {row[0]: row[1] for row in perf_store._db.execute(
|
|
"SELECT res, COALESCE(SUM(value),0) FROM sample GROUP BY res")}
|
|
|
|
def values_slice(res: str, limit: int = 4000) -> list[tuple[str, str, str, float, float]]:
|
|
"""A deterministic slice of the table, in primary-key order, compared exactly.
|
|
|
|
`ORDER BY source, host, metric, ts` walks the primary-key index rather than the rowids,
|
|
so this is the same list of rows before and after regardless of where maintenance moved
|
|
them — which is the assertion the aggregate cannot make: not one stored value rewritten.
|
|
"""
|
|
with perf_store._lock, perf_store._db:
|
|
return perf_store._db.execute(
|
|
"SELECT source, host, metric, ts, value FROM sample WHERE res=?"
|
|
" ORDER BY source, host, metric, ts LIMIT ?", (res, limit)).fetchall()
|
|
|
|
before, before_totals = shape(), totals()
|
|
slices = {res: values_slice(res) for res in (RES_RAW, RES_MIN, RES_HOUR)}
|
|
again = perf_store.maintain(now=now)
|
|
after, after_totals = shape(), totals()
|
|
|
|
assert after == before, (
|
|
"maintenance is on a timer: a second pass over the same data may not add or drop a row, "
|
|
"or move a resolution's time window")
|
|
for res, total in after_totals.items():
|
|
assert total == pytest.approx(before_totals[res], rel=1e-9), (
|
|
f"{res}: the stored total moved more than float accumulation can explain — a rollup "
|
|
f"re-averaged a bucket it had already rolled")
|
|
for res, rows in slices.items():
|
|
assert values_slice(res) == rows, (
|
|
f"{res}: a row's value changed on the second pass — the rollup is not idempotent")
|
|
assert again["pruned"] == 0, "nothing in this store is past its retention"
|
|
assert perf_store.count() == sum(row[0] for row in before.values())
|
|
|
|
|
|
def test_the_race_measurements_leave_the_seeded_shape_alone(perf_store: SampleStore) -> None:
|
|
"""A measurement that dirtied the baseline would move the next one — so this is a test.
|
|
|
|
The race inserts 30-sample polls (`store_perf.RACE_SOURCE`, never a seeded series) and the
|
|
task-04 comparison depends on the seeded counts being what they were.
|
|
"""
|
|
before = store_perf.counts(perf_store)
|
|
result = store_perf.race_single_connection(perf_store, iterations=5, max_seconds=15.0)
|
|
after = store_perf.counts(perf_store)
|
|
|
|
assert result.attempted >= 1 and result.inserts == result.attempted
|
|
assert after == before, f"the race left rows behind: {before} → {after}"
|
|
with perf_store._lock, perf_store._db:
|
|
leftovers = int(perf_store._db.execute(
|
|
"SELECT COUNT(*) FROM sample WHERE source=?",
|
|
(store_perf.RACE_SOURCE,)).fetchone()[0])
|
|
assert leftovers == 0
|
|
|
|
|
|
def test_the_batched_answers_match_the_per_series_answers_on_the_real_shaped_store(
|
|
perf_store: SampleStore) -> None:
|
|
"""The equality that lets the endpoints switch doors: same series, same order, same cost class.
|
|
|
|
On the store this phase measures its budgets on, at the two windows where the resolution
|
|
decision actually differs (24 h reads 1m, 90 d reads 1h), for the 27-series request
|
|
`/api/system/history` answers for a whole page. A capped count that decided a resolution
|
|
differently from a full `COUNT(*)` would move a chart onto the wrong resolution — the same
|
|
points averaged together, which is the kind of wrong that still renders.
|
|
"""
|
|
hosts = list(store_perf.HOST_SERIES)
|
|
metrics = list(store_perf.HISTORY_METRICS)
|
|
for span in (24 * 3_600.0, 90 * 24 * 3_600.0):
|
|
res_map = perf_store.pick_res_many("host", hosts, metrics, span, 600)
|
|
assert res_map == {(host, metric): perf_store.pick_res("host", host, metric, span, 600)
|
|
for host in hosts for metric in metrics}
|
|
batched = perf_store.history_many("host", hosts, metrics, span=span, max_points=600)
|
|
assert batched == {(host, metric): perf_store.history("host", host, metric,
|
|
span=span, max_points=600)
|
|
for host in hosts for metric in metrics}, (
|
|
f"the batched path changed a series at span={span:.0f}")
|
|
# and the series themselves are still the shape the charts parse: ascending, budgeted,
|
|
# ending at the newest row the window holds. `HISTORY_METRICS` is the *request* the page
|
|
# makes and the seed holds a subset of it (`cpu.temp_c` is not one of the seeded names),
|
|
# so the absent ones are asserted empty rather than skipped — an empty series is an answer.
|
|
held = {(host, metric) for host, names in store_perf.HOST_SERIES.items() for metric in names}
|
|
for (host, metric), points in batched.items():
|
|
if (host, metric) not in held:
|
|
assert points == [], f"{host}/{metric} is not in the seed: it must stay empty"
|
|
continue
|
|
assert points, f"{host}/{metric} at {span} came back empty on a store full of it"
|
|
assert 0 < len(points) <= 601, "the budget plus the point the thinning appends back"
|
|
assert [ts for ts, _value in points] == sorted(ts for ts, _value in points)
|
|
|
|
|
|
def test_two_maintain_cycles_leave_the_seeded_wal_under_the_cap(perf_store: SampleStore) -> None:
|
|
"""Criterion 3, on the store this phase measures its numbers on.
|
|
|
|
`journal_size_limit` is the ceiling; the windowed rollup is the reason the file has hardly
|
|
anything to grow with. The assertion is on the cap rather than a byte count because the cap is
|
|
the promise — a WAL that stays under 64 MB across ticks is what AGENTS.md known gap 4 needed,
|
|
and a run that left a bigger file behind would have to be explained by something other than
|
|
`maintain()`.
|
|
"""
|
|
path = Path(perf_store.path)
|
|
now = store_perf.seed_now(perf_store)
|
|
with perf_store._lock, perf_store._db: # start from a checkpointed file, so this measures the cycles
|
|
perf_store._db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
|
wal_before = store_perf.wal_bytes(path)
|
|
for _ in range(2):
|
|
perf_store.maintain(now=now)
|
|
wal_after = store_perf.wal_bytes(path)
|
|
# 8 MB of headroom is generous for a tick that writes one bucket per series: it is there so the
|
|
# test says something about the cycles rather than about the filesystem's rounding.
|
|
assert wal_after <= JOURNAL_SIZE_LIMIT_BYTES, (
|
|
f"WAL {wal_after / 1e6:,.1f} MB after two cycles, cap "
|
|
f"{JOURNAL_SIZE_LIMIT_BYTES / 1e6:,.0f} MB")
|
|
assert wal_after <= wal_before + 8 * 1024 * 1024, (
|
|
f"two steady ticks grew the WAL {wal_before:,} → {wal_after:,} bytes")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# task 04: the budgets
|
|
#
|
|
# Two kinds of test, because the phase's own rule is that a cost claim is pinned as a shape *and*
|
|
# priced, and the two need different stores. The shape half runs on the little `db` fixture in
|
|
# every `uv run pytest`: the budget table has to be internally sound and every budget has to name
|
|
# a measurement that still exists, so a budget cannot rot into a comment by pointing at a report
|
|
# key somebody renamed. The price half runs on the seeded store behind `--store-perf`, where the
|
|
# numbers mean something; a budget priced on a 40-row store would be a budget for a database
|
|
# nobody has, which is the failure AGENTS.md warns about.
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_the_budget_table_is_internally_coherent() -> None:
|
|
"""Every pinned budget is a checkable claim about a real report path.
|
|
|
|
Cheap, and it is the test that keeps the rest honest: a budget with a limit of 0 is always
|
|
crossed (or never, if the statistic is a count), a ratio pointing at a key in another section
|
|
silently measures nothing, and two rows on the same measurement with different ceilings is the
|
|
moment a table stops meaning one thing.
|
|
"""
|
|
assert store_perf.BUDGETS, "there are no budgets"
|
|
seen: set[tuple[str, str, str]] = set()
|
|
for budget in store_perf.BUDGETS:
|
|
assert budget.section in store_perf.BUDGET_SECTIONS, f"unknown section {budget.section}"
|
|
assert budget.key, "a budget must name the report key it prices"
|
|
assert budget.limit >= 0.0, f"{budget.label()} has a negative ceiling"
|
|
assert budget.unit in {"ms", "count"}, f"{budget.label()} has an unknown unit {budget.unit}"
|
|
assert budget.basis, f"{budget.label()} records no reasoning for its ceiling"
|
|
identity = (budget.section, budget.key, budget.statistic)
|
|
assert identity not in seen, f"{identity} is pinned twice, with different ceilings possible"
|
|
seen.add(identity)
|
|
# every budget's key has to be a key the pass that prices it actually produces
|
|
measured = (store_perf.READ_MEASURED_KEYS if budget.section == "paths"
|
|
else store_perf.HTTP_MEASURED_KEYS if budget.section == "http"
|
|
else store_perf.MAINTAIN_MEASURED_KEYS if budget.section == "maintain" else None)
|
|
if measured is not None:
|
|
assert budget.key in measured, (f"{budget.label()} names {budget.key}, which "
|
|
f"{budget.section} does not measure")
|
|
if budget.ratio_key is not None:
|
|
# `_budget_value` swaps only the key, so the partner has to live in the same section
|
|
# and be priced the same way, or the ratio compares two different things. The partner
|
|
# is usually *not* itself a budget — nobody promises the cold scan a ceiling; it is the
|
|
# denominator that says the cached answer is not a scan in disguise.
|
|
assert measured is not None and budget.ratio_key in measured, (
|
|
f"{budget.label()} is relative to {budget.ratio_key}, which {budget.section} "
|
|
"does not measure")
|
|
assert budget.ratio_limit is not None and 0.0 < budget.ratio_limit < 1.0, (
|
|
f"{budget.label()} must claim a *fraction* of its partner")
|
|
# the cached reads are the ones the phase claims are no longer scans: that claim is the ratio,
|
|
# so both cached paths carry one rather than a bare 10 ms that a fast laptop always beats
|
|
ratios = {budget.key for budget in store_perf.BUDGETS if budget.ratio_key is not None}
|
|
assert {"series_names_host_cached", "stats_cached"} <= ratios
|
|
|
|
|
|
def test_every_read_budget_names_a_measurement_that_still_exists(db) -> None:
|
|
"""The stale-pointer check, on the little store, in the default run.
|
|
|
|
`measure_budget_reads` is what the priced test calls; if a key there stops matching a key in
|
|
`BUDGETS` the priced test would quietly price nothing and pass. Crossings are not asserted
|
|
here — a 40-row store makes every ceiling trivially true and every ratio a coin flip.
|
|
"""
|
|
report = store_perf.measure_budget_reads(db, runs=2)
|
|
assert set(report["paths"]) == set(store_perf.READ_MEASURED_KEYS), (
|
|
"the read pass and the budget table no longer describe the same measurements")
|
|
assert store_perf.missing_budget_keys(report, ("paths",)) == []
|
|
rows = store_perf.budget_rows(report, ("paths",))
|
|
assert len(rows) == sum(1 for budget in store_perf.BUDGETS
|
|
if budget.section == "paths")
|
|
assert all(row.measured is not None and row.measured >= 0.0 for row in rows)
|
|
# the ratio partner is looked up by swapping the key, so it has to resolve on the same report
|
|
for row in rows:
|
|
if row.budget.ratio_key is not None:
|
|
partner = store_perf._budget_value(
|
|
report, replace(row.budget, key=row.budget.ratio_key))
|
|
assert partner is not None, f"{row.budget.label()} has an unpriceable partner"
|
|
|
|
|
|
@pytest.mark.store_perf
|
|
def test_the_seeded_store_reads_fit_their_pinned_budgets(perf_store: SampleStore) -> None:
|
|
"""The read budgets, priced on the store this phase's numbers were taken from.
|
|
|
|
Ceiling 2-3x the measured "after" (see each row's `basis`), so a slower deployment disk does
|
|
not turn red for the wrong reason — but the *scan* paths are what these catch: put the
|
|
whole-table `COUNT(*)` back in `stats()` and the cached row crosses on the ratio alone, which
|
|
is the half a generous absolute ceiling cannot see.
|
|
"""
|
|
report = store_perf.measure_budget_reads(perf_store)
|
|
print()
|
|
print(store_perf.render_budgets_markdown(store_perf.budget_rows(report, ("paths",))))
|
|
assert store_perf.budget_violations(report, ("paths",)) == []
|
|
|
|
|
|
@pytest.mark.store_perf
|
|
def test_the_write_side_fits_its_pinned_budgets(perf_store: SampleStore) -> None:
|
|
"""`maintain()`'s write transaction and the poller racing it — the gap-4 budgets.
|
|
|
|
Priced on a scratch copy (`measure_maintain`'s rule: a rollup really writes, and the cached
|
|
baseline has to come back unchanged), at the seed's own clock. The race needs two connections,
|
|
so it runs on its own copy too; the whole-window control is *not* budgeted here — it is what
|
|
`test_the_baseline_report_measures_every_hot_path` uses to prove the race can still see a
|
|
failure, and it is supposed to be slow.
|
|
"""
|
|
path = Path(perf_store.path)
|
|
now = store_perf.seed_now(perf_store)
|
|
maintain = store_perf.measure_maintain(path, now=now)
|
|
report = {"maintain": maintain}
|
|
print()
|
|
print(store_perf.render_budgets_markdown(store_perf.budget_rows(report, ("maintain",))))
|
|
assert store_perf.budget_violations(report, ("maintain",)) == []
|
|
|
|
with store_perf.scratch_copy(path, name="budget-race") as arena:
|
|
race = store_perf.race_two_connections(arena, iterations=100, now=now,
|
|
max_seconds=60.0,
|
|
maintain_gap_seconds=store_perf
|
|
.RACE_MAINTAIN_GAP_SECONDS)
|
|
report = {"race": {"poller_during_windowed_maintain": race.as_dict()}}
|
|
print()
|
|
print(store_perf.render_budgets_markdown(store_perf.budget_rows(report, ("race",))))
|
|
assert store_perf.budget_violations(report, ("race",)) == []
|
|
|
|
|
|
@pytest.mark.store_perf
|
|
def test_the_baseline_report_measures_every_hot_path(perf_store: SampleStore) -> None:
|
|
"""The measurement itself, inside pytest: every path task 01 names, with its EXPLAIN shape.
|
|
|
|
Opt-in (`-m store_perf --store-perf`) because it takes minutes, and skipped by default so
|
|
the shape tests above still run in every `uv run pytest`. The numbers that go in
|
|
`data/logs/NOTES-store-perf.md` come from `scripts/store_perf_probe.py` with 100 race
|
|
iterations; this runs the same code with a short race so the *report's shape* is guarded.
|
|
"""
|
|
report = store_perf.measure(perf_store, race_iterations=10, race_max_seconds=10.0,
|
|
with_http=True, log=lambda line: print(line, flush=True))
|
|
|
|
assert {"machine", "store", "paths", "plans", "race", "http", "maintain"} <= report.keys()
|
|
for key in ("series_names_all", "series_names_host", "stats", "pick_res_90d",
|
|
"history_5m", "history_24h", "history_90d",
|
|
# task 02: both halves of a cached answer, so the before/after compares cold to
|
|
# cold instead of calling a cache hit a 1 000x speed-up of a full scan
|
|
"series_names_all_cached", "series_names_host_cached", "stats_cached",
|
|
# task 02: the batched door against the loop it replaced, at both widths
|
|
"pick_res_x27_24h", "pick_res_many_x27_24h", "history_x27_24h",
|
|
"history_many_x27_24h", "pick_res_x27_90d", "pick_res_many_x27_90d",
|
|
"history_x27_90d", "history_many_x27_90d"):
|
|
assert report["paths"][key]["runs"] >= 1 and report["paths"][key]["max"] > 0.0
|
|
for key in store_perf.HTTP_MEASURED_KEYS:
|
|
assert key in report["http"], f"{key} is a hot path the operator's browser pays"
|
|
assert set(report["http"]) == set(store_perf.HTTP_MEASURED_KEYS)
|
|
# task 04: the report is the budget table's other customer. A budget crossed here is the phase
|
|
# saying "the regression is back", and a budget this report cannot price is a budget nobody
|
|
# checks any more — which is why `budget_rows` reports the missing ones as failures.
|
|
assert store_perf.missing_budget_keys(report) == []
|
|
assert report["budgets_ok"], "a pinned budget is crossed:\n" + "\n".join(
|
|
store_perf.budget_violations(report))
|
|
for key, plan_rows in report["plans"].items():
|
|
assert plan_rows, f"{key}: no EXPLAIN QUERY PLAN row — the statement was not explained"
|
|
|
|
# one connection: the store lock serialises, so an insert never sees SQLITE_BUSY here
|
|
assert report["race"]["single_connection"]["busy_errors"] == 0
|
|
# two connections: the gap-4 failure mode is a *raised* error, and a short-race pass here still
|
|
# has to have attempted inserts for that to mean anything. The saturated rows (task 01's shape)
|
|
# are reported for comparability only; the pass line is the cadence-shaped one, and the
|
|
# whole-window control beside it is what proves the race can still see the failure.
|
|
for key in ("two_connections_prod_timeout", "two_connections_no_timeout",
|
|
"poller_during_windowed_maintain", "poller_during_full_window_maintain"):
|
|
entry = report["race"][key]
|
|
assert entry["attempted"] >= 1 and entry["attempted"] <= 10
|
|
assert entry["inserts"] + entry["busy_errors"] == entry["attempted"]
|
|
assert entry["notes"], "every race says what it actually ran"
|
|
windowed = report["race"]["poller_during_windowed_maintain"]
|
|
assert windowed["busy_errors"] == 0, (
|
|
f"a poller insert lost to a windowed rollup: {windowed}")
|
|
assert report["race"]["poller_during_full_window_maintain"]["waiters"] >= 1, (
|
|
"the control pass must contend, or the windowed zero proves nothing")
|
|
# the write transaction, per window, on a scratch copy — windowing has to show up as a shorter
|
|
# write for the same rows, which is the whole point of task 03
|
|
passes = report["maintain"]["passes"]
|
|
assert set(passes) == {label for label, _offset, _reset in store_perf.MAINTAIN_PASSES}
|
|
assert all(entry["seconds"] > 0.0 for entry in passes.values())
|
|
assert passes["full_window"]["seconds"] > passes["steady"]["seconds"], (
|
|
"the control measured no slower than the windowed pass — the copy or the window is wrong")
|
|
assert passes["full_window"]["rows_written"] > passes["steady"]["rows_written"]
|
|
# criterion 1's cost bound. Ten minutes of missed buckets is ten times the rollup work of one
|
|
# minute and costs under twice the wall clock at this size because the fixed costs dominate; the
|
|
# slack is the prune, which is deliberately **not** windowed and therefore deletes as many rows
|
|
# as aged out during the jump whatever the rollup window was.
|
|
assert passes["catch_up_10m"]["seconds"] <= 2 * passes["steady"]["seconds"] + 0.10, (
|
|
f"catch-up {passes['catch_up_10m']['seconds']:.3f} s vs steady "
|
|
f"{passes['steady']['seconds']:.3f} s")
|
|
# criterion 2's shape, at production scale: a window the planner can seek. Without statistics it
|
|
# prices a range as a fraction of the table and reads the table for it (`SCAN sample`) — which is
|
|
# how a "windowed" rollup came to cost 328 ms while matching nothing.
|
|
for key, plan_rows in report["maintain"]["plans"].items():
|
|
assert plan_rows, f"{key}: nothing was explained"
|
|
assert not any(row.startswith("SCAN") for row in plan_rows), (
|
|
f"{key} still reads the whole table: {plan_rows}")
|
|
assert report["maintain"]["wal_within_cap"], (
|
|
f"the WAL escaped journal_size_limit across two ticks: {report['maintain']['two_steady_cycles']}")
|
|
assert report["store_after"]["integrity"] == "unchanged", (
|
|
f"the measurements changed the baseline: {report['store_after']['integrity']}")
|
|
|
|
print()
|
|
print(store_perf.render_markdown(report))
|