Skip to content

python-math #3: fix(delphi): cold-start generator copies FULL revote history (no DISTINCT ON dedup) #2638

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jucor wants to merge 1 commit into
base: spr/edge/a6494735
Choose a base branch
from
Draft

python-math #3: fix(delphi): cold-start generator copies FULL revote history (no DISTINCT ON dedup) #2638

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 79 additions & 36 deletions delphi/scripts/generate_cold_start_clojure.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -296,24 +299,55 @@ 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


def copy_votes_with_fresh_timestamps(conn, source_zid: int, fake_zid: int) -> int:
"""
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.
"""
Expand All @@ -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

Expand Down
127 changes: 127 additions & 0 deletions delphi/tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
- Session-scoped conversation cache for efficient test execution
"""

import contextlib
import os
from copy import deepcopy

import pytest
Expand Down Expand Up @@ -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
# =============================================================================
Expand Down
Loading
Loading