2b266caab5
harness/compact.py + cli.py compact [--dry-run|--vacuum|--verify-archives]: шум сек-уровня (CycleTicked sec, ReflexObserved без алерта/команд/каналов) старше часа → архив jsonl.gz (0600, строки целиком со sig) → мост цепочки (подписи унесённых строк, на которые ссылаются остающиеся, + маркеры boundary:<id> для legacy-границ) → подписанный StoreCompacted с sha256 архива/строк/моста → удаление пачками. audit.verify_events принимает мост только при совпадении хеша с подписанным событием; проекции подгружают мост лениво, если хвост перекрыл сжатый диапазон (rebuild с нуля). Репетиция на копии прода: 1389 → 200 МБ, verify 484 с/3.4 ГБ → 29 с/420 МБ, архив 246 МБ — все 2.47 млн подписей проверены; rebuild_projections OK. Найдено репетицией: legacy-граница (id 466633) и коллизия id сжатия в одну секунду — оба покрыты тестом. scripts/compact-store.sh — процедура с VACUUM под остановленными писателями. мин-уровень: если до ответа нужна работа >30 с — сначала короткий tg-reply, итог — cli.py tg-reply <id> --followup (в чат входящего: группа → группа; идемпотентно по тексту). p90 «старт LLM → ответ» был 223 с. hour: лимит 300 вызовов оставлен — реальный пик 205 (p90 169), 100 оборвал бы 8 из 19 тиков. Тесты: test_compact 25 PASS, test_followup 10 PASS; полный набор — без регрессий. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
148 lines
7.7 KiB
Python
148 lines
7.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Сжатие стора с подписью (harness/compact.py) — hermetic: свой стор, свой ключ.
|
||
|
||
Запуск: ./venv/bin/python tests/test_compact.py
|
||
|
||
Проверяет: шум сек-уровня уходит в архив, значимое остаётся (алерт/команды
|
||
сек-уровня, задачи, сообщения, голова стора); полный verify до и после —
|
||
OK, мост сверен с подписанным StoreCompacted; подмена моста (убрать/
|
||
добавить подпись, снести таблицу) → FAIL; архив: sha256 и подпись каждой
|
||
строки, правка строки → FAIL; повторное сжатие; VACUUM ужимает файл;
|
||
проекции после сжатия живы.
|
||
"""
|
||
import gzip
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
|
||
sys.path.insert(0, "/root/simbiot-harness")
|
||
_tmp = tempfile.mkdtemp(prefix="simbiot_compact_")
|
||
os.environ["SIMBIOT_DB"] = os.path.join(_tmp, "harness.db")
|
||
os.environ["SIMBIOT_AUDIT_KEY_DIR"] = os.path.join(_tmp, "audit")
|
||
os.environ["SIMBIOT_ARCHIVE_DIR"] = os.path.join(_tmp, "archive")
|
||
|
||
from harness import audit, compact, es
|
||
|
||
es.init_db()
|
||
fails = []
|
||
|
||
|
||
def check(name, cond, extra=""):
|
||
print(("PASS" if cond else "FAIL"), name, ("| " + str(extra)) if not cond else "")
|
||
if not cond:
|
||
fails.append(name)
|
||
|
||
|
||
def fill(seq0, n):
|
||
for i in range(seq0, seq0 + n):
|
||
es.append_tick("sec", i)
|
||
es.append_event("cycle", f"reflex:{i}", "ReflexObserved", {"level": "sec", "seq": i, "load1": 0.5})
|
||
if i % 50 == 0:
|
||
es.append_event("cycle", f"reflex-alert:{i}", "ReflexObserved",
|
||
{"level": "sec", "seq": i, "alert": "load 20 >= 12"})
|
||
if i % 40 == 0:
|
||
es.append_event("tg", f"m-{i}", "MessageReceived", {"update_id": i, "text": "привет"})
|
||
es.append_tick("min", i)
|
||
|
||
|
||
def types():
|
||
db = sqlite3.connect(os.environ["SIMBIOT_DB"])
|
||
out = dict(db.execute("SELECT event_type, COUNT(*) FROM events GROUP BY 1").fetchall())
|
||
db.close()
|
||
return out
|
||
|
||
|
||
fill(1000, 300)
|
||
# legacy-граница: неподписанная шумовая строка (как писал старый код до
|
||
# рестарта демонов) и за ней подписанная с пустым prev — кейс id 466633
|
||
_db = sqlite3.connect(os.environ["SIMBIOT_DB"])
|
||
_db.execute("INSERT INTO events (aggregate,aggregate_id,event_type,data,timestamp,version) "
|
||
"VALUES ('cycle','reflex:legacy','ReflexObserved','{\"level\":\"sec\"}',?,1)",
|
||
(es.datetime.now(es.timezone.utc).isoformat(),))
|
||
_db.commit(); _db.close()
|
||
es.append_event("tg", "m-after-legacy", "MessageReceived", {"update_id": 1, "text": "после legacy"})
|
||
fill(2000, 60)
|
||
before = types()
|
||
v0 = audit.verify_events()
|
||
check("verify до сжатия — OK", v0["ok"] and v0["checked"] > 600, v0)
|
||
plan = compact.plan(older_than_hours=0)
|
||
check("plan: кандидаты — шум сек-уровня", plan["candidates"] >= 590, plan)
|
||
|
||
r = compact.compact(older_than_hours=0, log=lambda *_: None)
|
||
after = types()
|
||
check("сжатие: строки ушли в архив и удалены", r["rows"] == r["removed"] >= 590, r)
|
||
check("значимое осталось: MessageReceived", after.get("MessageReceived") == before["MessageReceived"])
|
||
check("значимое осталось: мин-тики",
|
||
sqlite3.connect(os.environ["SIMBIOT_DB"]).execute(
|
||
"SELECT COUNT(*) FROM events WHERE event_type='CycleTicked' "
|
||
"AND json_extract(data,'$.level')='min'").fetchone()[0] == 10)
|
||
check("маркер legacy-границы в мосте",
|
||
sqlite3.connect(os.environ["SIMBIOT_DB"]).execute(
|
||
"SELECT COUNT(*) FROM compaction_bridge WHERE sig LIKE 'boundary:%'").fetchone()[0] >= 1)
|
||
check("значимое осталось: ReflexObserved с алертом",
|
||
sqlite3.connect(os.environ["SIMBIOT_DB"]).execute(
|
||
"SELECT COUNT(*) FROM events WHERE event_type='ReflexObserved' "
|
||
"AND json_extract(data,'$.alert') IS NOT NULL").fetchone()[0] == 8)
|
||
check("StoreCompacted записан", after.get("StoreCompacted") == 1)
|
||
check("мост не пуст (ссылки остающихся на удалённых)", r["bridge_count"] > 0, r["bridge_count"])
|
||
|
||
v1 = audit.verify_events()
|
||
check("verify после сжатия — OK", v1["ok"] and v1.get("compactions") == 1, v1)
|
||
check("в цепочке меньше строк", v1["total"] < v0["total"] - 500, (v0["total"], v1["total"]))
|
||
|
||
arch = os.path.join(os.environ["SIMBIOT_ARCHIVE_DIR"], r["archive"])
|
||
va = compact.verify_archive(arch, r["archive_sha256"])
|
||
check("архив: sha256 и подписи всех строк", va["ok"] and va["rows"] == r["rows"], va)
|
||
check("архив 0600", oct(os.stat(arch).st_mode & 0o777) == "0o600")
|
||
|
||
# подделка моста
|
||
db = sqlite3.connect(os.environ["SIMBIOT_DB"])
|
||
victim = db.execute("SELECT sig FROM compaction_bridge LIMIT 1").fetchone()[0]
|
||
db.execute("DELETE FROM compaction_bridge WHERE sig=?", (victim,)); db.commit()
|
||
check("мост без подписи → FAIL", not audit.verify_events()["ok"])
|
||
db.execute("INSERT INTO compaction_bridge VALUES (?, ?)", (victim, r["compaction"]))
|
||
db.execute("INSERT INTO compaction_bridge VALUES ('forged', ?)", (r["compaction"],)); db.commit()
|
||
check("мост с лишней подписью → FAIL", not audit.verify_events()["ok"])
|
||
db.execute("DELETE FROM compaction_bridge WHERE sig='forged'"); db.commit()
|
||
check("мост восстановлен → OK", audit.verify_events()["ok"])
|
||
saved = db.execute("SELECT sig, compaction FROM compaction_bridge").fetchall()
|
||
db.execute("DROP TABLE compaction_bridge"); db.commit()
|
||
check("таблица моста снесена → FAIL", not audit.verify_events()["ok"])
|
||
db.execute(compact.BRIDGE_DDL)
|
||
db.executemany("INSERT INTO compaction_bridge VALUES (?, ?)", saved); db.commit()
|
||
db.close()
|
||
check("мост на месте → OK", audit.verify_events()["ok"])
|
||
|
||
# подделка архива
|
||
bad = os.path.join(_tmp, "bad.jsonl.gz")
|
||
with gzip.open(arch, "rt", encoding="utf-8") as f:
|
||
lines = f.read().splitlines()
|
||
row = json.loads(lines[5]); row["data"] = row["data"].replace("0.5", "9.9")
|
||
lines[5] = json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||
with gzip.open(bad, "wt", encoding="utf-8") as f:
|
||
f.write("\n".join(lines) + "\n")
|
||
vb = compact.verify_archive(bad, r["archive_sha256"])
|
||
check("правка строки архива → подпись и sha256 не сходятся",
|
||
not vb["ok"] and vb["first_bad"] == row["id"] and not vb["sha_ok"], vb)
|
||
|
||
# проекции живы, повторное сжатие
|
||
es.run_projections()
|
||
fill(5000, 120)
|
||
r2 = compact.compact(older_than_hours=0, log=lambda *_: None)
|
||
v2 = audit.verify_events()
|
||
check("второе сжатие — verify OK, сжатий 2", r2["removed"] > 200 and v2["ok"] and v2["compactions"] == 2, v2)
|
||
check("проекции после сжатий", isinstance(es.run_projections(), dict))
|
||
# бывшая голова (шум, которую второй проход не тронул) — законный кандидат
|
||
r3 = compact.compact(older_than_hours=0, log=lambda *_: None)
|
||
check("третий проход: только бывшая голова", r3["removed"] <= 1, r3)
|
||
v3 = audit.verify_events()
|
||
check("verify после третьего сжатия — OK", v3["ok"], v3)
|
||
|
||
vac = compact.vacuum()
|
||
check("VACUUM ужал файл", vac["after"] < vac["before"], vac)
|
||
check("verify после VACUUM — OK", audit.verify_events()["ok"])
|
||
|
||
print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)")
|
||
sys.exit(1 if fails else 0)
|