63 lines
3.7 KiB
Python
63 lines
3.7 KiB
Python
"""The repository's projection-coherence oracle (`recompute_checks` and
|
|
`expectProjectionsMatchRecompute` in src/storage/repositories/queries_repo.zig),
|
|
plus file-level invariants. Exits 1 on any violation. usage: check_coherence.py DATA_DIR"""
|
|
import sqlite3
|
|
import sys
|
|
|
|
DATA = sys.argv[1]
|
|
BUCKET = "(CASE WHEN timestamp >= 0 THEN timestamp / 1800 * 1800 ELSE (timestamp - 1799) / 1800 * 1800 END)"
|
|
CHECKS = [
|
|
("SELECT bucket, queries, blocked, cached, rt_sum, rt_count FROM bucket_totals",
|
|
f"SELECT {BUCKET}, count(*), coalesce(sum(blocked <> 0), 0), coalesce(sum(cache_hit = 1), 0),"
|
|
" coalesce(sum(response_time_us), 0), count(response_time_us) FROM query_log GROUP BY 1"),
|
|
("SELECT bucket, client_ip, queries FROM bucket_clients",
|
|
f"SELECT {BUCKET}, client_ip, count(*) FROM query_log GROUP BY 1, 2"),
|
|
("SELECT bucket, qtype, count FROM bucket_types",
|
|
f"SELECT {BUCKET}, coalesce(qtype, -1), count(*) FROM query_log GROUP BY 1, 2"),
|
|
("SELECT bucket, route_kind, source_present, source_text, count FROM bucket_routes",
|
|
f"SELECT {BUCKET}, route_kind, source IS NOT NULL, coalesce(source, ''), count(*)"
|
|
" FROM (SELECT timestamp, route_kind, CASE route_kind WHEN 'upstream' THEN upstream"
|
|
" WHEN 'forward_zone' THEN forward_zone END AS source FROM query_log) GROUP BY 1, 2, 3, 4"),
|
|
]
|
|
|
|
ql = sqlite3.connect(f"file:{DATA}/querylog.db?mode=ro", uri=True)
|
|
failures = 0
|
|
|
|
|
|
def expect(label, ok, detail=""):
|
|
global failures
|
|
print(("ok " if ok else "FAIL ") + label + (f" ({detail})" if detail else ""))
|
|
failures += not ok
|
|
|
|
|
|
for projection, recompute in CHECKS:
|
|
diff = ql.execute(
|
|
f"SELECT (SELECT count(*) FROM ({projection} EXCEPT {recompute}))"
|
|
f" + (SELECT count(*) FROM ({recompute} EXCEPT {projection}))"
|
|
).fetchone()[0]
|
|
expect(f"projection == recompute: {projection.split('FROM ')[1]}", diff == 0, f"{diff} differing rows")
|
|
|
|
integrity = ql.execute("PRAGMA integrity_check").fetchone()[0]
|
|
expect("querylog integrity_check", integrity == "ok", integrity)
|
|
expect("querylog user_version == 1", ql.execute("PRAGMA user_version").fetchone()[0] == 1)
|
|
orphans = ql.execute("SELECT count(*) FROM query_log WHERE domain_id NOT IN (SELECT id FROM domains)").fetchone()[0]
|
|
expect("every query_log.domain_id names a domain", orphans == 0, f"{orphans} orphans")
|
|
created, since, first = ql.execute(
|
|
"SELECT created_at, available_since, (SELECT min(timestamp) FROM query_log) FROM querylog_meta").fetchone()
|
|
expect("available_since = created_at + 1 <= first row", since == created + 1 and (first is None or since <= first),
|
|
f"created_at={created} available_since={since} first_row={first}")
|
|
disorder = ql.execute(
|
|
"SELECT count(*) FROM query_log a JOIN query_log b ON b.id = a.id + 1 WHERE b.timestamp < a.timestamp").fetchone()[0]
|
|
expect("query_log ids ascend with timestamps", disorder == 0, f"{disorder} inversions")
|
|
|
|
cfg = sqlite3.connect(f"file:{DATA}/config.db?mode=ro", uri=True)
|
|
cfg.execute(f"ATTACH 'file:{DATA}/querylog.db?mode=ro' AS ql")
|
|
bad = cfg.execute("""SELECT count(*) FROM clients c WHERE EXISTS (SELECT 1 FROM ql.query_log q WHERE q.client_ip = c.ip)
|
|
AND (c.first_seen > (SELECT min(timestamp) FROM ql.query_log WHERE client_ip = c.ip)
|
|
OR c.last_seen < (SELECT max(timestamp) FROM ql.query_log WHERE client_ip = c.ip))""").fetchone()[0]
|
|
expect("clients first_seen/last_seen bracket their logged rows", bad == 0, f"{bad} clients out of range")
|
|
unknown = cfg.execute("SELECT count(DISTINCT client_ip) FROM ql.query_log WHERE client_ip NOT IN (SELECT ip FROM clients)").fetchone()[0]
|
|
expect("every logged client is a named client row", unknown == 0, f"{unknown} unknown")
|
|
expect("config integrity_check", cfg.execute("PRAGMA integrity_check").fetchone()[0] == "ok")
|
|
sys.exit(1 if failures else 0)
|