Skip to content

commit-id:b1f62d74 #2628

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

Closed
jucor wants to merge 1 commit into from
Closed

commit-id:b1f62d74 #2628

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
85 changes: 60 additions & 25 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 @@ -309,11 +309,40 @@ 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 session only via ``session_replication_role``
(identical to ``copy_comments_with_fresh_timestamps`` above), restoring it in a
``finally``. 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,29 +351,35 @@ 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
# Disable rules/triggers for this session only (safe for concurrent use):
# suppresses on_vote_insert_update_unique_table so the multi-row revote
# INSERT does not trip the single-statement ON CONFLICT cardinality check.
cursor.execute("SET session_replication_role = 'replica'")

try:
# 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()
finally:
# Restore normal rule/trigger behavior for this session.
cursor.execute("SET session_replication_role = 'origin'")
conn.commit()

copied_count = cursor.rowcount
conn.commit()
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