momentum devandClaude Opus 4.8
78bdf26141
DB latency: every projection runs `WHERE event_type=? ORDER BY timestamp DESC
LIMIT n`. With only a plain timestamp index, the latest of a STALE type
backward-scans the whole events table — measured 373ms / ~52k buffers on 235k
rows. Add a composite index in event_store._ensure_schema:
CREATE INDEX idx_events_type_ts ON events(event_type, timestamp DESC)
-> same query is now a 0.018ms index seek (3 buffers). Applied to the live DB
CONCURRENTLY; also dropped two redundant legacy indexes (idx_events_type, now
covered by the composite prefix; idx_events_correlation, a dup of
idx_events_session_id) to cut write overhead.
tests/conftest.py: sweep the throwaway `mem_*` schemas EventStore(":memory:")
creates — 170 had leaked into the shared Postgres, bloating the catalog.
README: rewritten around the current system — PostgreSQL event store, a full
command reference (run_live_paper / momentum CLI / eod_replay / sweeps / tuning /
cron), config knobs, and a database & performance section (the index above,
extension guidance, retention).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
"""Shared pytest fixtures + session cleanup.
|
|
|
|
Drop the throwaway ``mem_*`` schemas that ``EventStore(":memory:")`` creates for
|
|
test isolation. The closing fixture drops them on ``.close()``, but tests that
|
|
build a store inline (without that fixture) leak one schema per run — 170 had
|
|
piled up in the shared Postgres instance, bloating the catalog and slowing
|
|
query planning. This sweeps them at session end.
|
|
"""
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def _drop_leftover_mem_schemas():
|
|
yield # run after the whole session
|
|
# Under xdist each worker shares the same Postgres, so a worker dropping at
|
|
# its own session-end could yank a schema another worker is still using.
|
|
# Only sweep in serial runs (the common case); parallel runs can sweep
|
|
# manually afterwards.
|
|
if os.environ.get("PYTEST_XDIST_WORKER"):
|
|
return
|
|
url = os.environ.get("DATABASE_URL")
|
|
if not url:
|
|
return
|
|
try:
|
|
import psycopg2
|
|
conn = psycopg2.connect(url)
|
|
conn.autocommit = True
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT schema_name FROM information_schema.schemata "
|
|
"WHERE schema_name LIKE 'mem_%'")
|
|
for (schema,) in cur.fetchall():
|
|
cur.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
|
|
conn.close()
|
|
except Exception: # noqa: BLE001 — cleanup must never fail the run
|
|
pass
|