From 0463ba96022393fffde00fa666f2be4621b7f0d8 Mon Sep 17 00:00:00 2001 From: Julien Cornebise Date: Sat, 18 Jul 2026 02:15:06 +0100 Subject: [PATCH] python-math #3: fix(delphi): cold-start generator copies FULL revote history (no DISTINCT ON dedup) commit-id:970c79cf --- delphi/scripts/generate_cold_start_clojure.py | 115 +++++++--- delphi/tests/conftest.py | 127 +++++++++++ delphi/tests/test_generator_vote_copy.py | 205 ++++++++++++++++++ 3 files changed, 411 insertions(+), 36 deletions(-) create mode 100644 delphi/tests/test_generator_vote_copy.py diff --git a/delphi/scripts/generate_cold_start_clojure.py b/delphi/scripts/generate_cold_start_clojure.py index bc7444b19..56ee4e6a1 100755 --- a/delphi/scripts/generate_cold_start_clojure.py +++ b/delphi/scripts/generate_cold_start_clojure.py @@ -276,16 +276,19 @@ def copy_comments_with_fresh_timestamps(conn, source_zid: int, fake_zid: int) -> fail with 'nil has zero dimensionality'. The tid_auto trigger auto-assigns tids, so we disable triggers for this - session only (using session_replication_role) to preserve original tids. - This is safe for concurrent use — only affects the current DB session. + transaction only (SET LOCAL session_replication_role) to preserve original + tids. This is safe for concurrent use — only affects the current DB session, + and auto-reverts on commit/rollback. """ cursor = conn.cursor() now_ms = int(time.time() * 1000) - # Disable triggers for this session only (safe for concurrent use) - cursor.execute("SET session_replication_role = 'replica'") - try: + # SET LOCAL confines the override to THIS transaction: it reverts on + # commit AND on rollback, so a failed INSERT can never leave the + # session stuck in replica mode. (The first execute on a non-autocommit + # psycopg2 connection opens the transaction block SET LOCAL needs.) + cursor.execute("SET LOCAL session_replication_role = 'replica'") cursor.execute(""" INSERT INTO comments (zid, tid, pid, txt, created, velocity, mod, active, modified, uid, anon, is_seed, curation, is_meta) @@ -296,12 +299,13 @@ def copy_comments_with_fresh_timestamps(conn, source_zid: int, fake_zid: int) -> """, (fake_zid, now_ms, source_zid)) count = cursor.rowcount conn.commit() + except Exception: + # Clear the aborted transaction (which also reverts the SET LOCAL) so + # the original error propagates unmasked and the session stays usable. + conn.rollback() + raise finally: - # Restore normal trigger behavior for this session - cursor.execute("SET session_replication_role = 'origin'") - conn.commit() - - cursor.close() + cursor.close() return count @@ -309,11 +313,41 @@ def copy_votes_with_fresh_timestamps(conn, source_zid: int, fake_zid: int) -> in """ Copy votes from source conversation to fake conversation with fresh timestamps. - Preserves vote ORDER by using sequential timestamps starting from now. - The poller finds votes by `created > last_poll_timestamp`, so fresh - timestamps ensure these votes are picked up. - - Uses a single INSERT ... SELECT for efficiency (no Python roundtrips). + Copies the FULL vote history, including revotes (multiple rows for the same + (pid, tid) pair). An earlier version deduplicated with + ``DISTINCT ON (pid, tid) ... ORDER BY created DESC`` ("keep the latest"), + which silently dropped superseded revote rows (vw: 128 of 4683). That made + the Clojure reference consume a DIFFERENT input than the Python side (which + feeds every CSV row and lets the engine's later-vote-wins merge resolve + revotes), and it erases the revote dynamics that sequential replay + specifically needs (see REPLAY_HARNESS_DESIGN.md §5: "Do NOT dedup + revotes"). Both engines implement later-vote-wins internally, so the dedup + was never necessary for correctness of the final matrix — only harmful for + input parity. + + Preserves vote ORDER by using sequential timestamps starting from now + (10 ms apart, strictly increasing, so Clojure's later-vote-wins resolves + revotes in source order). Source order is ``created ASC`` with ``ctid`` as + a tiebreak: for revotes of the same (pid, tid) sharing the same source + millisecond, physical row order approximates insertion order (the table is + append-only); the true relative order of same-ms revotes is ambiguous in + the source data itself. + + The poller finds votes by ``created > last_poll_timestamp``, so fresh + timestamps ensure these votes are picked up. Uses a single + INSERT ... SELECT for efficiency (no Python roundtrips). + + The ``votes`` table carries the LIVE rule ``on_vote_insert_update_unique_table`` + (migration 000006): every INSERT DO-ALSO upserts ``votes_latest_unique`` with + ``ON CONFLICT (zid,pid,tid) DO UPDATE``. Because this single INSERT carries the + FULL history (revotes = duplicate (pid,tid) keys), the rule's upsert would hit + the same conflict key twice IN ONE STATEMENT, which Postgres rejects with + "ON CONFLICT DO UPDATE command cannot affect row a second time". We therefore + disable rules/triggers for this transaction only via ``SET LOCAL + session_replication_role`` (identical to ``copy_comments_with_fresh_timestamps`` + above), which auto-reverts on commit/rollback. Safe: the Clojure poller reads + only ``votes``, never + ``votes_latest_unique`` (postgres.clj:139,204,284); this is a throwaway copy. Returns the number of votes copied. """ @@ -322,30 +356,39 @@ def copy_votes_with_fresh_timestamps(conn, source_zid: int, fake_zid: int) -> in # Get current time in milliseconds (matching Polis schema) now_ms = int(time.time() * 1000) - # Single INSERT ... SELECT with ROW_NUMBER() to generate sequential timestamps - # This is much faster than executemany for large vote counts - # Use DISTINCT ON (pid, tid) to handle duplicate votes (keeps the latest) - cursor.execute(""" - INSERT INTO votes (zid, pid, tid, vote, weight_x_32767, created) - SELECT - %s, - pid, - tid, - vote, - weight_x_32767, - %s + (ROW_NUMBER() OVER (ORDER BY created ASC) - 1) * 10 - FROM ( - SELECT DISTINCT ON (pid, tid) pid, tid, vote, weight_x_32767, created + try: + # Disable rules/triggers for this transaction only: suppresses + # on_vote_insert_update_unique_table so the multi-row revote INSERT does + # not trip the single-statement ON CONFLICT cardinality check. SET LOCAL + # reverts on commit AND rollback, so a failed INSERT can never leave the + # session stuck in replica mode. (The first execute on a non-autocommit + # psycopg2 connection opens the transaction block SET LOCAL needs.) + cursor.execute("SET LOCAL session_replication_role = 'replica'") + # Single INSERT ... SELECT with ROW_NUMBER() to generate sequential + # timestamps. This is much faster than executemany for large vote counts. + cursor.execute(""" + INSERT INTO votes (zid, pid, tid, vote, weight_x_32767, created) + SELECT + %s, + pid, + tid, + vote, + weight_x_32767, + %s + (ROW_NUMBER() OVER (ORDER BY created ASC, ctid ASC) - 1) * 10 FROM votes WHERE zid = %s - ORDER BY pid, tid, created DESC - ) AS deduplicated - ORDER BY created ASC - """, (fake_zid, now_ms, source_zid)) + ORDER BY created ASC, ctid ASC + """, (fake_zid, now_ms, source_zid)) - copied_count = cursor.rowcount - conn.commit() - cursor.close() + copied_count = cursor.rowcount + conn.commit() + except Exception: + # Clear the aborted transaction (which also reverts the SET LOCAL) so + # the original error propagates unmasked and the session stays usable. + conn.rollback() + raise + finally: + cursor.close() return copied_count diff --git a/delphi/tests/conftest.py b/delphi/tests/conftest.py index 110a048ed..3632e438e 100644 --- a/delphi/tests/conftest.py +++ b/delphi/tests/conftest.py @@ -9,6 +9,8 @@ - Session-scoped conversation cache for efficient test execution """ +import contextlib +import os from copy import deepcopy import pytest @@ -119,6 +121,131 @@ def require_s3( pytest.skip(f"S3/MinIO is not available at {endpoint}: {exc}") +_POLIS_PG_MIGRATIONS_DIR = os.path.join( + os.path.dirname(__file__), "..", "..", "server", "postgres", "migrations", +) +# Migrations that establish the votes + votes_latest_unique schema and the +# on_vote_insert_update_unique_table RULE. 000006 holds the LIVE rule +# redefinition (idempotent DROP/CREATE) — apply both, in order. +_POLIS_PG_MIGRATIONS = ("000000_initial.sql", "000006_update_votes_rule.sql") + + +def _free_tcp_port() -> int: + """Grab an ephemeral free TCP port (avoids clashing on a fixed port under + xdist / when several integration modules run concurrently).""" + import socket + + with socket.socket() as s: + s.bind(("", 0)) + return s.getsockname()[1] + + +@contextlib.contextmanager +def require_polis_postgres(): + """Yield a Postgres URL with the polis votes schema applied — for opt-in + integration tests — or ``pytest.skip()`` if no Postgres is reachable. + + Resolution order: + + 1. **CI service** — if ``POLIS_TEST_POSTGRES_URL`` is set (a reachable + Postgres whose image already bakes the polis migrations, e.g. the + ``postgres`` service in ``docker-compose.test.yml`` which loads + ``server/postgres/migrations/*.sql`` via docker-entrypoint-initdb.d), + use it. The schema is verified; the caller skips loudly if it is + missing (a provisioned CI service is expected to have it). + 2. **Local throwaway docker** — a fresh ``postgres:17`` on an EPHEMERAL + port (NEVER the host's live 5432), with 000000 + 000006 applied via + ``psql``. + 3. Otherwise skip with a clear reason. + + Migrations applied: ``000000_initial.sql`` (votes + votes_latest_unique + + the ``on_vote_insert_update_unique_table`` rule) and + ``000006_update_votes_rule.sql`` (the LIVE rule redefinition). + + Shared by ``tests/poller/test_integration_postgres.py`` and + ``tests/test_generator_vote_copy.py``. + """ + import shutil + import subprocess + import time + import uuid + + import psycopg2 + + ci_url = os.environ.get("POLIS_TEST_POSTGRES_URL") + if ci_url: + try: + conn = psycopg2.connect(ci_url) + except Exception as exc: # pragma: no cover - infra guard + pytest.skip(f"POLIS_TEST_POSTGRES_URL set but unreachable: {exc}") + try: + with conn.cursor() as cur: + cur.execute("SELECT to_regclass('public.votes_latest_unique')") + present = cur.fetchone()[0] is not None + finally: + conn.close() + if not present: + pytest.skip( + "POLIS_TEST_POSTGRES_URL points at a Postgres without the polis " + "votes schema (expected the migrations baked into the service image)" + ) + yield ci_url + return + + docker = shutil.which("docker") + if not docker: + pytest.skip("no POLIS_TEST_POSTGRES_URL and docker not available") + + migrations = [ + os.path.abspath(os.path.join(_POLIS_PG_MIGRATIONS_DIR, m)) + for m in _POLIS_PG_MIGRATIONS + ] + for path in migrations: + if not os.path.exists(path): + pytest.skip(f"polis migration not found: {path}") + + port = _free_tcp_port() + name = f"delphi-polis-pg-it-{uuid.uuid4().hex[:8]}" + started = subprocess.run( + [docker, "run", "--rm", "-d", "--name", name, + "-p", f"{port}:5432", "-e", "POSTGRES_PASSWORD=test", "postgres:17"], + capture_output=True, text=True, + ) + if started.returncode != 0: + pytest.skip(f"could not start postgres container: {started.stderr.strip()}") + cid = started.stdout.strip() + try: + deadline = time.time() + 40 + ready = False + while time.time() < deadline: + if subprocess.run( + [docker, "exec", cid, "pg_isready", "-U", "postgres"], + capture_output=True, text=True, + ).returncode == 0: + ready = True + break + time.sleep(1) + if not ready: + pytest.skip("postgres container did not become ready in time") + + for path in migrations: + with open(path, "rb") as fh: + applied = subprocess.run( + [docker, "exec", "-i", cid, "psql", "-v", "ON_ERROR_STOP=1", + "-U", "postgres", "-d", "postgres"], + stdin=fh, capture_output=True, text=True, + ) + if applied.returncode != 0: + pytest.skip( + f"migration {os.path.basename(path)} failed to apply: " + f"{applied.stderr[-500:]}" + ) + + yield f"postgresql://postgres:test@localhost:{port}/postgres" + finally: + subprocess.run([docker, "stop", cid], capture_output=True, text=True) + + # ============================================================================= # Session-scoped Conversation Cache # ============================================================================= diff --git a/delphi/tests/test_generator_vote_copy.py b/delphi/tests/test_generator_vote_copy.py new file mode 100644 index 000000000..e2cef17e4 --- /dev/null +++ b/delphi/tests/test_generator_vote_copy.py @@ -0,0 +1,205 @@ +"""Integration test for the cold-start generator's full-history vote copy (T2). + +`copy_votes_with_fresh_timestamps` copies the FULL vote history (including +revotes — multiple rows for the same (pid, tid)) with a single multi-row +`INSERT ... SELECT`. The `votes` table carries the LIVE rule +`on_vote_insert_update_unique_table` (migration 000006), which DO-ALSO upserts +`votes_latest_unique` with `ON CONFLICT (zid,pid,tid) DO UPDATE`. A single INSERT +statement containing revote duplicates makes that upsert touch the same conflict +key twice IN ONE STATEMENT, which Postgres rejects with: + + ON CONFLICT DO UPDATE command cannot affect row a second time + +The fix wraps the copy in `session_replication_role = 'replica'` (mirroring +`copy_comments_with_fresh_timestamps`), suppressing the default-config rule for +the copy. This test seeds a source conversation WITH REVOTES and asserts the +copy round-trips every row, in source order, with the revote pairs preserved. + +OPT-IN / self-skipping: provisions a throwaway Postgres (or reuses the CI +service) via `require_polis_postgres` and applies the votes-schema migrations. +Skips cleanly when docker / a service is unavailable. +""" + +import importlib.util +import os + +import psycopg2 +import pytest + +from tests.conftest import require_polis_postgres + +pytestmark = pytest.mark.integration + + +def _load_generator(): + """Import the standalone generator script by path (it is not a package).""" + path = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "scripts", + "generate_cold_start_clojure.py") + ) + if not os.path.exists(path): + pytest.skip(f"generator script not found: {path}") + spec = importlib.util.spec_from_file_location("generate_cold_start_clojure", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +SOURCE_ZID = 990101 +FAKE_ZID = 990102 + +# (pid, tid) -> list of vote values, one row per revote in chronological order. +# Two keys have 3 revotes, two have 2, two have 1: 3+2+1+2+1+3 = 12 source rows, +# 6 distinct (pid,tid) keys. +_REVOTES = { + (0, 0): [-1, 1, -1], + (0, 1): [1, -1], + (1, 0): [-1], + (1, 1): [1, 1], + (2, 0): [-1], + (2, 1): [1, -1, 1], +} + + +@pytest.fixture(scope="module") +def pg_url(): + with require_polis_postgres() as url: + yield url + + +def _seed_source(url): + """Insert the source vote history one row at a time (so seeding does NOT + itself trip the single-statement rule), interleaving revotes across keys. + + `created` advances once per round-robin ROUND, not per row: rows within a + round share the same `created` (same-ms ties), so the copy's + `ORDER BY created ASC, ctid ASC` tiebreak is actually exercised — a copy + that dropped the ctid ordering could reorder tied rows and fail the + order-preservation assertion.""" + conn = psycopg2.connect(url) + conn.autocommit = True + created = 1_000_000 + n_rows = 0 + try: + with conn.cursor() as cur: + cur.execute("DELETE FROM votes_latest_unique WHERE zid IN (%s, %s)", + (SOURCE_ZID, FAKE_ZID)) + cur.execute("DELETE FROM votes WHERE zid IN (%s, %s)", + (SOURCE_ZID, FAKE_ZID)) + # Interleave: round-robin over keys by revote index so revotes are + # spread through the timeline rather than clustered per key. + max_revotes = max(len(v) for v in _REVOTES.values()) + for k in range(max_revotes): + created += 1 # ties WITHIN a round, distinct across rounds + for (pid, tid), votes in _REVOTES.items(): + if k < len(votes): + cur.execute( + "INSERT INTO votes (zid, pid, tid, vote, created) " + "VALUES (%s, %s, %s, %s, %s)", + (SOURCE_ZID, pid, tid, votes[k], created), + ) + n_rows += 1 + finally: + conn.close() + return n_rows + + +def _source_order(url): + """Source (pid,tid,vote) tuples in copy order: created ASC, ctid ASC.""" + conn = psycopg2.connect(url) + try: + with conn.cursor() as cur: + cur.execute( + "SELECT pid, tid, vote FROM votes WHERE zid = %s " + "ORDER BY created ASC, ctid ASC", + (SOURCE_ZID,), + ) + return cur.fetchall() + finally: + conn.close() + + +class TestCopyVotesFullHistory: + def test_copy_round_trips_all_revotes_in_order(self, pg_url): + gen = _load_generator() + n_source = _seed_source(pg_url) + assert n_source == sum(len(v) for v in _REVOTES.values()) == 12 + source_seq = _source_order(pg_url) + + # The call under test. Before the fix this raises CardinalityViolation + # ("ON CONFLICT DO UPDATE command cannot affect row a second time"); + # after the fix it copies every row with the rule suppressed. + conn = psycopg2.connect(pg_url) + try: + copied = gen.copy_votes_with_fresh_timestamps(conn, SOURCE_ZID, FAKE_ZID) + finally: + conn.close() + + assert copied == n_source # ALL rows copied, incl. revotes + + verify = psycopg2.connect(pg_url) + try: + with verify.cursor() as cur: + # (1) exact row count preserved (revotes not deduped). + cur.execute("SELECT COUNT(*) FROM votes WHERE zid = %s", (FAKE_ZID,)) + assert cur.fetchone()[0] == n_source + + # (2) created strictly increasing, 10 ms apart, in source order. + cur.execute( + "SELECT pid, tid, vote, created FROM votes WHERE zid = %s " + "ORDER BY created ASC", + (FAKE_ZID,), + ) + rows = cur.fetchall() + createds = [r[3] for r in rows] + assert len(createds) == n_source + assert all(b - a == 10 for a, b in zip(createds, createds[1:])) # strictly incr + assert len(set(createds)) == n_source + # order matches source (created ASC, ctid ASC) + assert [(r[0], r[1], r[2]) for r in rows] == source_seq + + # (3) every revote pair preserved with the right multiplicity. + cur.execute( + "SELECT pid, tid, COUNT(*) FROM votes WHERE zid = %s " + "GROUP BY pid, tid", + (FAKE_ZID,), + ) + counts = {(pid, tid): n for pid, tid, n in cur.fetchall()} + assert counts == {k: len(v) for k, v in _REVOTES.items()} + finally: + verify.close() + + # Cleanup (harmless on a throwaway container; keeps a shared CI service tidy). + cleanup = psycopg2.connect(pg_url) + cleanup.autocommit = True + try: + with cleanup.cursor() as cur: + cur.execute("DELETE FROM votes_latest_unique WHERE zid IN (%s, %s)", + (SOURCE_ZID, FAKE_ZID)) + cur.execute("DELETE FROM votes WHERE zid IN (%s, %s)", + (SOURCE_ZID, FAKE_ZID)) + finally: + cleanup.close() + + def test_failed_copy_propagates_original_error_and_leaves_session_clean(self, pg_url): + """If the bulk INSERT fails mid-copy, the ORIGINAL error must propagate + (not a follow-on InFailedSqlTransaction from cleanup running inside the + aborted transaction), and the session must come back clean: rule/trigger + suppression reverted, connection usable.""" + gen = _load_generator() + _seed_source(pg_url) + + conn = psycopg2.connect(pg_url) + try: + # Injected failure: NULL fake_zid violates votes.zid NOT NULL. + with pytest.raises(psycopg2.IntegrityError): + gen.copy_votes_with_fresh_timestamps(conn, SOURCE_ZID, None) + + # Same connection stays usable, with normal rule/trigger behavior. + with conn.cursor() as cur: + cur.execute("SELECT current_setting('session_replication_role')") + assert cur.fetchone()[0] == "origin" + cur.execute("SELECT COUNT(*) FROM votes WHERE zid = %s", (FAKE_ZID,)) + assert cur.fetchone()[0] == 0 # failed copy left nothing behind + finally: + conn.close()