Add SQLite results storage with a resilient multiprocess writer (Stage 5)
- storage/schema.py: `runs` table recording every evaluated configuration, feasible or not, with an honest infeasible_reason instead of dropping failures - storage/database.py: WAL-mode SQLite, single-writer-over-a-queue pattern for safe concurrent writes from many sweep worker processes - Testing with real multiprocessing.Process (not mocks) caught a genuine bug: a duplicate run_id raised inside the writer loop and killed the writer process, silently halting queue drainage for the rest of a sweep. Fixed by catching the insert error per-record and logging to stderr instead of crashing; added a regression test for it. - Verified locally and inside the Docker image (43/43 tests both ways) Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
126
tests/test_storage_database.py
Normal file
126
tests/test_storage_database.py
Normal file
@@ -0,0 +1,126 @@
|
||||
import multiprocessing
|
||||
from pathlib import Path
|
||||
|
||||
from gausse.storage.database import (
|
||||
count_runs,
|
||||
fetch_runs,
|
||||
insert_run,
|
||||
insert_runs,
|
||||
open_connection,
|
||||
run_writer_process,
|
||||
)
|
||||
from gausse.storage.schema import RunRecord
|
||||
|
||||
|
||||
def _record(run_id: str, feasible: bool = True, efficiency: float | None = 0.1, reason=None) -> RunRecord:
|
||||
return RunRecord(
|
||||
run_id=run_id,
|
||||
timestamp="2026-07-06T00:00:00",
|
||||
search_mode="sweep",
|
||||
genome_json="{}",
|
||||
decoded_summary_json="{}",
|
||||
feasible=feasible,
|
||||
model_version="v1",
|
||||
infeasible_reason=reason,
|
||||
efficiency=efficiency,
|
||||
exit_velocity_mps=20.0 if feasible else None,
|
||||
cost_rub=500.0,
|
||||
)
|
||||
|
||||
|
||||
def test_insert_and_fetch_roundtrip(tmp_path):
|
||||
conn = open_connection(tmp_path / "runs.sqlite3")
|
||||
insert_run(conn, _record("r1"))
|
||||
rows = fetch_runs(conn)
|
||||
assert len(rows) == 1
|
||||
assert rows[0].run_id == "r1"
|
||||
assert rows[0].feasible is True
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_infeasible_runs_are_stored_not_dropped(tmp_path):
|
||||
conn = open_connection(tmp_path / "runs.sqlite3")
|
||||
insert_run(conn, _record("r-fail", feasible=False, efficiency=None, reason="датчик не сработал"))
|
||||
assert count_runs(conn) == 1
|
||||
assert count_runs(conn, feasible=False) == 1
|
||||
assert count_runs(conn, feasible=True) == 0
|
||||
row = fetch_runs(conn, feasible=False)[0]
|
||||
assert row.infeasible_reason == "датчик не сработал"
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_batch_insert_and_filters(tmp_path):
|
||||
conn = open_connection(tmp_path / "runs.sqlite3")
|
||||
records = [
|
||||
_record("r1", efficiency=0.05),
|
||||
_record("r2", efficiency=0.30),
|
||||
_record("r3", feasible=False, efficiency=None, reason="разряд не скоммутировался"),
|
||||
]
|
||||
insert_runs(conn, records)
|
||||
assert count_runs(conn) == 3
|
||||
best = fetch_runs(conn, feasible=True, min_efficiency=0.1)
|
||||
assert [r.run_id for r in best] == ["r2"]
|
||||
ranked = fetch_runs(conn, feasible=True, order_by_efficiency_desc=True)
|
||||
assert [r.run_id for r in ranked] == ["r2", "r1"]
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_reopening_database_preserves_previous_runs(tmp_path):
|
||||
db_path = tmp_path / "runs.sqlite3"
|
||||
conn = open_connection(db_path)
|
||||
insert_run(conn, _record("r1"))
|
||||
conn.close()
|
||||
|
||||
conn2 = open_connection(db_path)
|
||||
assert count_runs(conn2) == 1
|
||||
conn2.close()
|
||||
|
||||
|
||||
def test_writer_survives_duplicate_run_id_and_keeps_draining_queue(tmp_path):
|
||||
"""Регрессия: одна коллизия run_id раньше роняла writer-процесс и молча
|
||||
останавливала осушение очереди (см. историю разработки Этапа 5)."""
|
||||
db_path = tmp_path / "runs.sqlite3"
|
||||
ctx = multiprocessing.get_context("spawn")
|
||||
queue = ctx.Queue()
|
||||
|
||||
writer = ctx.Process(target=run_writer_process, args=(queue, db_path))
|
||||
writer.start()
|
||||
|
||||
queue.put(_record("dup"))
|
||||
queue.put(_record("dup")) # коллизия PRIMARY KEY
|
||||
queue.put(_record("after-collision"))
|
||||
queue.put(None)
|
||||
writer.join(timeout=10)
|
||||
|
||||
assert not writer.is_alive()
|
||||
conn = open_connection(db_path)
|
||||
assert count_runs(conn) == 2 # "dup" один раз + "after-collision"
|
||||
assert {r.run_id for r in fetch_runs(conn)} == {"dup", "after-collision"}
|
||||
conn.close()
|
||||
|
||||
|
||||
def _writer_worker(queue, worker_id, n):
|
||||
for i in range(n):
|
||||
queue.put(_record(f"proc-{worker_id}-run-{i}"))
|
||||
|
||||
|
||||
def test_multiprocess_writer_receives_all_records_from_multiple_workers(tmp_path):
|
||||
db_path = tmp_path / "runs.sqlite3"
|
||||
ctx = multiprocessing.get_context("spawn")
|
||||
queue = ctx.Queue()
|
||||
|
||||
writer = ctx.Process(target=run_writer_process, args=(queue, db_path))
|
||||
writer.start()
|
||||
|
||||
workers = [ctx.Process(target=_writer_worker, args=(queue, worker_id, 20)) for worker_id in range(3)]
|
||||
for w in workers:
|
||||
w.start()
|
||||
for w in workers:
|
||||
w.join()
|
||||
|
||||
queue.put(None)
|
||||
writer.join()
|
||||
|
||||
conn = open_connection(db_path)
|
||||
assert count_runs(conn) == 60
|
||||
conn.close()
|
||||
Reference in New Issue
Block a user