diff --git a/delphi/polismath/poller/__init__.py b/delphi/polismath/poller/__init__.py new file mode 100644 index 000000000..302634ddf --- /dev/null +++ b/delphi/polismath/poller/__init__.py @@ -0,0 +1,153 @@ +"""polismath.poller — Python replacement for the Clojure math poller (phase 1). + +A service that polls Postgres for votes/moderation, maintains per-conversation +math state in-memory, and writes the same Postgres tables the TS server + legacy +clients consume. See ``delphi/docs/MATH_POLLER_DESIGN.md`` for the full recon +and cutover plan. + +Architecture +------------ +:: + + scripts/math_poller.py (CLI) + └─ service.MathPollerService + ├─ vote loop (thread): poll_votes_since(wm) cadence VOTE_POLLING_INTERVAL + ├─ mod loop (thread): poll_moderation_since(wm) cadence MOD_POLLING_INTERVAL + │ both group-by zid, allow/block filter, advance watermark to max(ts) + ├─ worker_pool.ConversationWorkerPool + │ one FIFO queue + single-owner flag per zid -> strict per-zid + │ serialization; drains+coalesces queued batches (votes-before- + │ moderation); bounded concurrency across zids + ├─ engine: Conversation held in memory per zid + │ update_votes(recompute=False) -> update_moderation(recompute=False) + │ -> recompute() (POLISMATH_ENGINE_MODE honored) + ├─ load-or-init (first message per zid): from_dict(math_main) warm + │ restore + full-history rating-matrix rebuild + ├─ math_writer.MathWriter: math_main (caching_tick=MAX+1), math_bidtopid + │ (derived from base_clusters), math_ptptstats — one shared math_tick + └─ error path: dump conv+batch JSON -> retry once -> park zid (breaker) + +Clojure provenance for every duty is cited inline in each module. + +bidToPid shape (VERIFIED against both ends of the contract) +----------------------------------------------------------- +``math_bidtopid.data`` = ``{"zid", "bidToPid", "lastVoteTimestamp"}`` where +``bidToPid`` is a LIST OF PID-LISTS, positionally aligned with +``math_main.base-clusters.id`` (ascending by base-cluster id): + +* Clojure ``prep-bidToPid`` (math/src/polismath/conv_man.clj:35-40) wraps + ``:bid-to-pid`` = ``(mapv :members (sort-by :id base-clusters))`` + (math/src/polismath/math/conversation.clj:585-586) — "a vector of member + vectors, sorted by base cluster id". +* The TS server (server/src/utils/participants.ts:33-51 with pca.ts:20-27 + ``base-clusters.members: number[][]``) indexes ``data.bidToPid`` by the + position of a bid inside ``base-clusters.id``: + ``bidToIndex[base_clusters.id[i]] = i`` then ``indexToPids[bidToIndex[bid]]``. + +Python's ``Conversation.base_clusters`` is sorted ascending by ``id`` +(conversation.py:789) and ``_fold_base_clusters`` writes ``base-clusters.id`` / +``.members`` in that order (conversation.py:1643-1649), so +``[c['members'] for c in conv.base_clusters]`` is the exact alignment the server +needs. ``derive_bidtopid`` (math_writer.py) implements this. Pids are strings +Python-side (poll_votes casts ``str(pid)``) vs ints Clojure-side; the server +parseInt()s them (participants.ts:53-55), so a parity comparer needs int/str +tolerance on this one field. + +load-or-init finding (from_dict restoration is PARTIAL) +------------------------------------------------------- +``Conversation.from_dict`` (conversation.py:2249-2303) restores from a dict with +underscore/nested keys: ``last_updated, participant_count, comment_count, +vote_stats, moderation{...}, pca{center,comps}, proj, group_clusters, repness, +participant_info, comment_priorities``. ``Conversation.to_dict`` (used as the +math_main ``data`` blob) is a SUPERSET that carries those same underscore keys +alongside the hyphenated Clojure keys, so ``from_dict(to_dict(conv))`` round-trips +the listed fields — notably the PCA warm-start vectors and prior moderation. + +But ``from_dict`` does NOT restore: ``raw_rating_mat`` / ``rating_mat`` (the vote +matrices), ``base_clusters``, ``subgroup_clusters``, ``group_clusterings`` / +``group_k_smoother`` (warm smoother state), ``consensus`` or ``group_votes``. +Therefore load-or-init ALWAYS rebuilds the rating matrices from the full vote +history (``poll_votes(zid)`` ordered by zid,tid,pid,created — parity with +conv-poll offset 0) and recomputes base_clusters; the non-persisted smoother +state cold-starts. This is CLOSE TO — but not byte-identical with — a Clojure +worker restart: on restart Clojure ``restructure-json-conv`` RESTORES +``base-clusters`` (and the PCA) from the persisted blob before its ``:reboot`` +recompute (conv_man.clj:173 keeps ``:base-clusters`` in the subset, :180 unfolds +them), whereas Python re-derives base_clusters cold +from the vote matrices. The rating-matrix rebuild itself matches +(conv_man.clj:188-207 rebuilds ``raw-rating-mat`` the same way), and we +opportunistically seed the warm PCA start from ``from_dict`` when a row exists +(low-risk, literally what ``restructure-json-conv`` does). The base-cluster +lineage difference is a KNOWN divergence to trace against Clojure's ``:reboot`` +semantics before the parity gate; a full cold rebuild is otherwise correct — +just without Clojure's restored-lineage warm start. + +Config var mapping (config.py names PREFERRED, design aliases accepted) +----------------------------------------------------------------------- +====================== ============================================== ======= +PollerConfig field Env var(s) (first set wins) Default +====================== ============================================== ======= +database_url DATABASE_URL — +math_env MATH_ENV dev +vote_interval_ms POLL_VOTE_INTERVAL_MS | VOTE_POLLING_INTERVAL | + POLL_INTERVAL_MS 1000 +mod_interval_ms POLL_MOD_INTERVAL_MS | MOD_POLLING_INTERVAL | + POLL_INTERVAL_MS 1000 +poll_from_days_ago POLL_FROM_DAYS_AGO 10 +allowlist POLL_ALLOWLIST | MATH_ZID_ALLOWLIST [] +blocklist POLL_BLOCKLIST | MATH_ZID_BLOCKLIST [] +engine_mode POLISMATH_ENGINE_MODE improved +worker_pool_size MATH_WORKER_POOL_SIZE 4 +dump_dir MATH_POLLER_DUMP_DIR scratch/errorconv +retry_cap MATH_POLLER_RETRY_CAP 1 +====================== ============================================== ======= + +``POLL_VOTE_INTERVAL_MS`` / ``POLL_MOD_INTERVAL_MS`` / ``POLL_ALLOWLIST`` / +``POLL_BLOCKLIST`` are the names already present in +``polismath.components.config.py`` (:216-269, previously unwired); +``VOTE_POLLING_INTERVAL`` / ``MOD_POLLING_INTERVAL`` / ``MATH_ZID_ALLOWLIST`` / +``MATH_ZID_BLOCKLIST`` are the design-doc aliases. (config.py's example default +for the mod interval was 5000ms; the binding design §3 uses 1000ms, adopted here.) + +Usage +----- +:: + + # Run the service (blocks; SIGTERM/SIGINT -> graceful stop) + uv run python scripts/math_poller.py + + # Single poll cycle then exit (smoke test / cron-style) + uv run python scripts/math_poller.py --once + +Shadow-mode deployment writes under a DISTINCT ``MATH_ENV`` (e.g. ``delphi``) +next to the Clojure ``math`` container; ``UNIQUE(zid, math_env)`` keeps the rows +invisible to the prod server until cutover. +""" + +from polismath.poller.service import ( + MathPollerService, + PollerConfig, + advance_watermark, + initial_watermark, + should_process_zid, +) +from polismath.poller.worker_pool import ( + ConversationWorkerPool, + CoalescedBatch, + coalesce_messages, +) +from polismath.poller.math_writer import MathWriter, derive_bidtopid, dump_error + +__all__ = [ + "MathPollerService", + "PollerConfig", + "advance_watermark", + "initial_watermark", + "should_process_zid", + "ConversationWorkerPool", + "CoalescedBatch", + "coalesce_messages", + "MathWriter", + "derive_bidtopid", + "dump_error", +] diff --git a/delphi/polismath/poller/math_writer.py b/delphi/polismath/poller/math_writer.py new file mode 100644 index 000000000..2e05d4a41 --- /dev/null +++ b/delphi/polismath/poller/math_writer.py @@ -0,0 +1,172 @@ +"""Postgres writer for the math poller. + +Writes the three data tables the TS server + legacy clients consume, all under +one math_env string and ONE shared math_tick per cycle, exactly like Clojure's +write-conv-updates! (conv_man.clj:158-169): + + math-tick = inc-math-tick(zid) ; atomic, postgres.clj:292-295 + upload-math-main zid math-tick ... ; postgres.clj:323-338 + upload-math-bidtopid zid math-tick ... ; postgres.clj:369-380 + upload-math-ptptstats zid math-tick ... ; postgres.clj:350-361 + +The Clojure-exact SQL (caching_tick = MAX+1 subquery, atomic tick upsert) lives +in polismath.database.postgres.PostgresClient; this module orchestrates the +per-cycle write and derives the bidToPid blob. +""" + +import json +import logging +import os +import time +import traceback +import uuid +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + + +def derive_bidtopid(conv: Any, zid: int) -> Dict[str, Any]: + """Derive the prep-bidToPid blob from a computed Conversation. + + Shape (verified against BOTH sides of the contract): + + * Clojure ``prep-bidToPid`` (conv_man.clj:35-40) emits + ``{:zid :bidToPid :lastVoteTimestamp}`` where ``:bidToPid`` is + ``(mapv :members (sort-by :id base-clusters))`` (conversation.clj:585-586) + — "a vector of member vectors, sorted by base cluster id". + + * The TS server (server/src/utils/participants.ts:33-51, + pca.ts:20-27 ``members: number[][]``) indexes ``data.bidToPid`` by the + POSITION of a base-cluster id inside ``base-clusters.id``: + ``bidToIndex[base_clusters.id[i]] = i`` then ``bidToPid[i]``. + + Python's ``Conversation.base_clusters`` is already sorted ascending by ``id`` + (conversation.py:789) and ``_fold_base_clusters`` (conversation.py:1643-1649) + writes ``base-clusters.id`` / ``base-clusters.members`` in that same order, so + ``[c['members'] for c in conv.base_clusters]`` is positionally aligned with + ``base-clusters.id`` — the exact alignment the server relies on. + + Note on element type: Python pids are strings (poll_votes casts ``str(pid)``), + whereas Clojure emits integer pids. The server parseInt()s them + (participants.ts:53-55) so both work; a parity comparer needs int/str + tolerance on this field. Members are left as-is so that + math_bidtopid.bidToPid and math_main.base-clusters.members stay identical. + + Args: + conv: A computed Conversation (public attrs only). + zid: Conversation id. + + Returns: + ``{"zid": int, "bidToPid": [[pid, ...], ...], "lastVoteTimestamp": int}`` + """ + base_clusters = getattr(conv, "base_clusters", None) or [] + # Defensive: never rely on caller having sorted; sort by id here too + # (idempotent since conversation.py already keeps them sorted). + ordered = sorted(base_clusters, key=lambda c: c["id"]) + bid_to_pid: List[List[Any]] = [list(c.get("members", [])) for c in ordered] + return { + "zid": zid, + "bidToPid": bid_to_pid, + "lastVoteTimestamp": getattr(conv, "last_updated", None), + } + + +def derive_ptptstats(conv: Any, zid: int) -> Dict[str, Any]: + """Derive the prep-ptpt-stats blob (conv_man.clj:90-94). + + ptptstats is a secondary consumer (scoped "replace", not fidelity-critical + like math_main / math_bidtopid). We wrap the conversation's public + ``participant_info`` under the same envelope keys Clojure uses. + """ + return { + "zid": zid, + "ptptstats": getattr(conv, "participant_info", {}) or {}, + "lastVoteTimestamp": getattr(conv, "last_updated", None), + } + + +class MathWriter: + """Writes a computed conversation's results to Postgres for one cycle.""" + + def __init__(self, pg_client: Any): + self._pg = pg_client + + def write_conv_updates(self, zid: int, conv: Any) -> int: + """Mint one math_tick and write all three data tables with it. + + Returns the math_tick used (handy for logging / tests). + """ + math_tick = self._pg.increment_math_tick(zid) + + data = conv.to_dict() + last_vote_timestamp = data.get("lastVoteTimestamp") + if last_vote_timestamp is None: + last_vote_timestamp = getattr(conv, "last_updated", None) + + # 1. math_main — client-facing PCA/cluster/repness blob (fidelity-critical) + self._pg.write_math_main( + zid, + data, + last_vote_timestamp=last_vote_timestamp, + math_tick=math_tick, + ) + # 2. math_bidtopid — server bid->pid mapping (fidelity-critical) + self._pg.write_math_bidtopid( + zid, data=derive_bidtopid(conv, zid), math_tick=math_tick + ) + # 3. math_ptptstats — participant stats + self._pg.write_participant_stats( + zid, data=derive_ptptstats(conv, zid), math_tick=math_tick + ) + + logger.info( + "Wrote math results for zid=%s math_tick=%s (main+bidtopid+ptptstats)", + zid, + math_tick, + ) + return math_tick + + +def dump_error( + zid: int, + conv: Any, + coalesced: Any, + error: BaseException, + dump_dir: str, +) -> str: + """Dump conversation state + failing batch + traceback to an errorconv JSON. + + Mirrors Clojure's conv-update-dump on failure (conv_man.clj:319-323): a + debugging artefact written before the batch is retried / the zid is parked. + + Returns the path written (best-effort; never raises). + """ + try: + os.makedirs(dump_dir, exist_ok=True) + # ms + short uuid so back-to-back dumps within the same millisecond never + # collide (the retry-then-park path dumps twice in quick succession). + stamp = f"{int(time.time() * 1000)}-{uuid.uuid4().hex[:8]}" + path = os.path.join(dump_dir, f"errorconv-zid{zid}-{stamp}.json") + try: + conv_dump = conv.to_dict() if conv is not None else None + except Exception: # pragma: no cover - defensive + conv_dump = {"_dump_error": "conv.to_dict() failed"} + payload = { + "zid": zid, + "error": str(error), + "traceback": "".join( + traceback.format_exception(type(error), error, error.__traceback__) + ), + "batch": { + "votes": getattr(coalesced, "votes", None), + "moderation": getattr(coalesced, "moderation", None), + }, + "conv": conv_dump, + } + with open(path, "w") as fh: + json.dump(payload, fh, default=str) + logger.error("Dumped failed conversation state for zid=%s to %s", zid, path) + return path + except Exception: # pragma: no cover - dump must never mask the real error + logger.exception("Unable to write errorconv dump for zid=%s", zid) + return "" diff --git a/delphi/polismath/poller/service.py b/delphi/polismath/poller/service.py new file mode 100644 index 000000000..1c1b1fe88 --- /dev/null +++ b/delphi/polismath/poller/service.py @@ -0,0 +1,499 @@ +"""MathPollerService — the Python replacement for the Clojure math poller. + +Two watermark loops (votes, moderation) poll Postgres, group results by zid, and +dispatch per-zid batches to a serialized worker pool. Each zid keeps a +Conversation in memory; the engine chain is +``update_votes(recompute=False) -> update_moderation(recompute=False) -> recompute()`` +and the results are written back to math_main / math_bidtopid / math_ptptstats +under one math_env and one shared math_tick. + +Recon anchors (Clojure): poller.clj:12-37 (poll loop + watermark + allow/block), +conv_man.clj:188-207 (load-or-init), :291-388 (actor + error handling). +""" + +import logging +import os +import threading +import time +from collections import OrderedDict +from dataclasses import dataclass, field +from typing import Any, Dict, Iterable, List, Optional + +from polismath.conversation.conversation import Conversation +from polismath.utils.engine_mode import ( + ENGINE_MODE_ENV_VAR, + ENGINE_MODE_CHOICES, + resolve_engine_mode, +) +from polismath.poller.math_writer import MathWriter, dump_error +from polismath.poller.worker_pool import ( + ConversationWorkerPool, + CoalescedBatch, + VOTES, + MODERATION, +) + +logger = logging.getLogger(__name__) + +_MS_PER_DAY = 24 * 60 * 60 * 1000 + + +# --------------------------------------------------------------------------- # +# Pure poll-loop helpers (unit-tested in isolation) +# --------------------------------------------------------------------------- # +def advance_watermark(current: int, timestamps: Iterable[int]) -> int: + """Advance a watermark to max(current, *timestamps); never regress. + + Clojure: ``(apply max 0 last-timestamp (map timestamp-key results))`` + (poller.clj:27). Because the loop's ``WHERE created > watermark`` is a + STRICT ``>``, monotonic-max advancement guarantees each row is delivered + exactly once and the watermark can only move forward. + """ + result = current + for ts in timestamps: + if ts is not None and ts > result: + result = ts + return result + + +def initial_watermark(poll_from_days_ago: float, now_millis: Optional[int] = None) -> int: + """Starting watermark = now - poll_from_days_ago days (poller.clj:15).""" + if now_millis is None: + now_millis = int(time.time() * 1000) + return int(now_millis - poll_from_days_ago * _MS_PER_DAY) + + +def should_process_zid( + zid: int, allowlist: List[int], blocklist: List[int] +) -> bool: + """Allow/block filter (poller.clj:30-32). + + Clojure ``cond``: if an allowlist is set, only listed zids pass; else if a + blocklist is set, listed zids are excluded; else everything passes. The + allowlist branch is evaluated first, so it wins over the blocklist. + """ + if allowlist: + return zid in allowlist + if blocklist: + return zid not in blocklist + return True + + +def _group_by_zid(rows: List[Dict[str, Any]]) -> Dict[int, List[Dict[str, Any]]]: + """Group polled rows by zid, preserving row order within each group.""" + grouped: Dict[int, List[Dict[str, Any]]] = {} + for row in rows: + grouped.setdefault(int(row["zid"]), []).append(row) + return grouped + + +def _parse_int_list(raw: Optional[str]) -> List[int]: + if not raw: + return [] + return [int(x.strip()) for x in raw.split(",") if x.strip()] + + +def _env_first(*names: str, default: Optional[str] = None) -> Optional[str]: + """Return the first env var that is set among names, else default. + + Lets us PREFER delphi's existing config.py names while accepting the design + doc's aliases (documented in the poller package docstring / config mapping). + """ + for name in names: + val = os.environ.get(name) + if val is not None and val != "": + return val + return default + + +# --------------------------------------------------------------------------- # +# Config +# --------------------------------------------------------------------------- # +@dataclass +class PollerConfig: + """Poller configuration. + + Env-var mapping (preferred name first, then design-doc alias): + database_url DATABASE_URL + math_env MATH_ENV (default 'dev') + vote_interval_ms POLL_VOTE_INTERVAL_MS | VOTE_POLLING_INTERVAL | POLL_INTERVAL_MS (1000) + mod_interval_ms POLL_MOD_INTERVAL_MS | MOD_POLLING_INTERVAL | POLL_INTERVAL_MS (1000) + poll_from_days_ago POLL_FROM_DAYS_AGO (default 10) + allowlist POLL_ALLOWLIST | MATH_ZID_ALLOWLIST (default []) + blocklist POLL_BLOCKLIST | MATH_ZID_BLOCKLIST (default []) + engine_mode POLISMATH_ENGINE_MODE (default None -> compute's own default) + worker_pool_size MATH_WORKER_POOL_SIZE (default 4) + dump_dir MATH_POLLER_DUMP_DIR (default 'scratch/errorconv') + retry_cap MATH_POLLER_RETRY_CAP (default 1) + conv_cache_cap MATH_CONV_CACHE_CAP (default 0 = unlimited) + """ + + database_url: Optional[str] = None + math_env: str = "dev" + vote_interval_ms: int = 1000 + mod_interval_ms: int = 1000 + poll_from_days_ago: float = 10 + allowlist: List[int] = field(default_factory=list) + blocklist: List[int] = field(default_factory=list) + engine_mode: Optional[str] = None + worker_pool_size: int = 4 + dump_dir: str = "scratch/errorconv" + retry_cap: int = 1 + # Max in-memory conversations before LRU-evicting the coldest. 0 = unlimited + # (the default preserves current behavior; the compose deploy memory limit is + # the hard backstop). Clojure's 4h reboot was the de-facto memory cap, which + # we dropped — set this to bound a long shadow soak; an evicted conv is + # reloaded from math_main + fully rebuilt on next touch (= Clojure restart). + conv_cache_cap: int = 0 + + @classmethod + def from_env(cls) -> "PollerConfig": + return cls( + database_url=os.environ.get("DATABASE_URL"), + math_env=os.environ.get("MATH_ENV", "dev"), + vote_interval_ms=int( + _env_first( + "POLL_VOTE_INTERVAL_MS", + "VOTE_POLLING_INTERVAL", + "POLL_INTERVAL_MS", + default="1000", + ) + ), + mod_interval_ms=int( + _env_first( + "POLL_MOD_INTERVAL_MS", + "MOD_POLLING_INTERVAL", + "POLL_INTERVAL_MS", + default="1000", + ) + ), + poll_from_days_ago=float(os.environ.get("POLL_FROM_DAYS_AGO", "10")), + allowlist=_parse_int_list( + _env_first("POLL_ALLOWLIST", "MATH_ZID_ALLOWLIST") + ), + blocklist=_parse_int_list( + _env_first("POLL_BLOCKLIST", "MATH_ZID_BLOCKLIST") + ), + engine_mode=os.environ.get(ENGINE_MODE_ENV_VAR), + worker_pool_size=int(os.environ.get("MATH_WORKER_POOL_SIZE", "4")), + dump_dir=os.environ.get("MATH_POLLER_DUMP_DIR", "scratch/errorconv"), + retry_cap=int(os.environ.get("MATH_POLLER_RETRY_CAP", "1")), + conv_cache_cap=int(os.environ.get("MATH_CONV_CACHE_CAP", "0")), + ) + + +# --------------------------------------------------------------------------- # +# Service +# --------------------------------------------------------------------------- # +class MathPollerService: + """Owns the poll loops, the in-memory conv cache, the worker pool + writer.""" + + def __init__(self, pg_client: Any, config: PollerConfig): + self._pg = pg_client + self.config = config + self._writer = MathWriter(pg_client) + # LRU order: most-recently-touched zid last, so popitem(last=False) evicts + # the coldest (see _remember). + self._convs: "OrderedDict[int, Conversation]" = OrderedDict() + self._retry_counts: Dict[int, int] = {} + self._parked: set = set() + self._pool: Optional[ConversationWorkerPool] = None + self._threads: List[threading.Thread] = [] + self._stop = threading.Event() + self._vote_wm: Optional[int] = None + self._mod_wm: Optional[int] = None + + # -- engine-mode passthrough ------------------------------------------- # + def apply_engine_mode(self) -> str: + """Propagate the configured engine mode into the process environment so + the in-process compute (conversation._compute_pca/_compute_clusters, + which read POLISMATH_ENGINE_MODE at call time) honors it. Returns the + resolved mode actually in effect.""" + if self.config.engine_mode: + if self.config.engine_mode not in ENGINE_MODE_CHOICES: + logger.warning( + "Unknown POLISMATH_ENGINE_MODE=%r; compute will fall back to " + "its default", + self.config.engine_mode, + ) + os.environ[ENGINE_MODE_ENV_VAR] = self.config.engine_mode + return resolve_engine_mode() + + # -- lifecycle ---------------------------------------------------------- # + def _ensure_runtime(self) -> None: + if self._pool is None: + self._pool = ConversationWorkerPool( + self._handle_zid, max_workers=self.config.worker_pool_size + ) + if self._vote_wm is None: + self._vote_wm = initial_watermark(self.config.poll_from_days_ago) + if self._mod_wm is None: + self._mod_wm = initial_watermark(self.config.poll_from_days_ago) + + def start(self) -> None: + self.apply_engine_mode() + self._ensure_runtime() + self._stop.clear() + self._threads = [ + threading.Thread(target=self._vote_loop, name="vote-poller", daemon=True), + threading.Thread(target=self._mod_loop, name="mod-poller", daemon=True), + ] + for t in self._threads: + t.start() + logger.info( + "MathPollerService started (math_env=%s engine_mode=%s pool=%d)", + self.config.math_env, + resolve_engine_mode(), + self.config.worker_pool_size, + ) + + def stop(self) -> None: + self._stop.set() + for t in self._threads: + t.join(timeout=5.0) + if self._pool is not None: + self._pool.join(timeout=30.0) + self._pool.shutdown(wait=True) + logger.info("MathPollerService stopped") + + def run_forever(self) -> None: + self.start() + try: + while not self._stop.is_set(): + self._stop.wait(1.0) + finally: + self.stop() + + # -- poll cycles -------------------------------------------------------- # + def poll_once(self) -> None: + """Run one vote + one moderation cycle, blocking until processed. + + Used by ``--once`` and the integration test. + """ + self.apply_engine_mode() + self._ensure_runtime() + self._poll_votes_once() + self._poll_moderation_once() + assert self._pool is not None + self._pool.join(timeout=120.0) + + def _vote_loop(self) -> None: + while not self._stop.is_set(): + try: + self._poll_votes_once() + except Exception: + logger.exception("Vote poll cycle failed") + self._stop.wait(self.config.vote_interval_ms / 1000.0) + + def _mod_loop(self) -> None: + while not self._stop.is_set(): + try: + self._poll_moderation_once() + except Exception: + logger.exception("Moderation poll cycle failed") + self._stop.wait(self.config.mod_interval_ms / 1000.0) + + def _unpark(self, zid: int) -> None: + """Self-heal a parked zid when a NEW batch arrives (Clojure retry-chan + equivalent). Park is transient across cycles: a transient write blip must + not leave a zid dead until process restart. Clears the retry counter so + the zid gets a fresh retry budget; the next batch reprocesses on the + last-good conv (or a rebuild if it was evicted).""" + if zid not in self._parked: + return + self._parked.discard(zid) + self._retry_counts.pop(zid, None) + if self._pool is not None: + self._pool.unpark(zid) + logger.info("Un-parked zid=%s: a new batch arrived (self-heal)", zid) + + def _poll_votes_once(self) -> None: + assert self._pool is not None + rows = self._pg.poll_votes_since(self._vote_wm) + logger.info("Polled %d votes since watermark %s", len(rows), self._vote_wm) + for zid, batch in _group_by_zid(rows).items(): + if should_process_zid(zid, self.config.allowlist, self.config.blocklist): + self._unpark(zid) # new batch self-heals a parked zid + self._pool.submit(zid, VOTES, batch) + self._vote_wm = advance_watermark( + self._vote_wm, (r["created"] for r in rows) + ) + + def _poll_moderation_once(self) -> None: + assert self._pool is not None + rows = self._pg.poll_moderation_since(self._mod_wm) + logger.info("Polled %d mod changes since watermark %s", len(rows), self._mod_wm) + for zid, batch in _group_by_zid(rows).items(): + if should_process_zid(zid, self.config.allowlist, self.config.blocklist): + self._unpark(zid) # new batch self-heals a parked zid + self._pool.submit(zid, MODERATION, batch) + self._mod_wm = advance_watermark( + self._mod_wm, (r["modified"] for r in rows) + ) + + # -- per-zid processing (runs on pool threads) -------------------------- # + def _handle_zid(self, zid: int, coalesced: CoalescedBatch) -> None: + if zid in self._parked: + return + try: + self._run_engine(zid, coalesced) + self._retry_counts.pop(zid, None) + except Exception as error: # noqa: BLE001 - top of the per-zid boundary + self._on_engine_error(zid, coalesced, error) + + def _remember(self, zid: int, conv: Conversation) -> None: + """Store a conversation as most-recently-used, LRU-evicting the coldest + when conv_cache_cap (>0) is exceeded. An evicted conv is reloaded from + math_main and fully rebuilt on its next touch (= Clojure-restart + semantics), so eviction is lossless — just a memory/latency trade.""" + self._convs[zid] = conv + self._convs.move_to_end(zid) + cap = self.config.conv_cache_cap + if cap and len(self._convs) > cap: + while len(self._convs) > cap: + evicted_zid, _ = self._convs.popitem(last=False) # coldest + logger.info( + "LRU-evicting cold conversation zid=%s (cache cap=%d); it will " + "reload from math_main + rebuild on next touch", + evicted_zid, cap, + ) + + def _run_engine(self, zid: int, coalesced: CoalescedBatch) -> None: + conv = self._convs.get(zid) + if conv is not None: + self._convs.move_to_end(zid) # LRU touch + + if conv is None: + # First message for this zid: load-or-init (full rebuild + compute). + conv = self._load_or_init(zid) + self._remember(zid, conv) + self._writer.write_conv_updates(zid, conv) + # The triggering batch is subsumed by the full-history rebuild. + return + + if coalesced.votes: + last_ts = advance_watermark( + conv.last_updated, + (v.get("created") for v in coalesced.votes), + ) + conv = conv.update_votes( + {"votes": coalesced.votes, "lastVoteTimestamp": last_ts}, + recompute=False, + ) + if coalesced.moderation: + # Re-derive the FULL current moderation state (idempotent; also + # captures un-moderation), then apply. + mods = self._pg.poll_moderation(zid, None) + conv = conv.update_moderation(mods, recompute=False) + + conv = conv.recompute() + self._remember(zid, conv) + self._writer.write_conv_updates(zid, conv) + + def _load_or_init(self, zid: int) -> Conversation: + """Mirror Clojure load-or-init (conv_man.clj:188-207). + + Restores warm state from math_main via ``Conversation.from_dict`` when a + row exists, then ALWAYS rebuilds the rating matrices from the full vote + history and applies the full moderation state (from_dict restores neither + the matrices nor base_clusters — see the poller package docstring's + "load-or-init finding"). Non-persisted warm smoother state cold-starts, + exactly like a Clojure worker restart. + + last_updated is seeded NONZERO-but-low (not wall-clock): ``Conversation``'s + ``last_updated = last_updated or now`` footgun (conversation.py:205) means a + cold ``Conversation(str(zid))`` starts at wall-clock now, and + ``advance_watermark(now, historical_created)`` can never regress it — so + the wall-clock leaks into math_main.last_vote_timestamp forever (Clojure + floors at 0 -> true max(created), conversation.clj:161-165). Seeding 1 + (dodging the falsy-0 fallback), or the persisted last_vote_timestamp when + restoring, lets the full-history update_votes below resolve last_updated to + the true max(created). + """ + conv: Optional[Conversation] = None + try: + row = self._pg.load_math_main(zid) + except Exception: + logger.exception("load_math_main failed for zid=%s; cold start", zid) + row = None + + if row and row.get("data"): + try: + conv = Conversation.from_dict(row["data"]) + # Prefer the persisted last_vote_timestamp column over the blob's + # last_updated (which a prior wall-clock write may have poisoned). + # A persisted 0 is legitimate (Clojure's floor) and must be + # preserved — only a NULL column falls back to the 0 floor. The + # constructor's `last_updated or now` footgun does not apply to + # this post-construction assignment. + persisted_ts = row.get("last_vote_timestamp") + conv.last_updated = persisted_ts if persisted_ts is not None else 0 + logger.info( + "load-or-init: restored warm state (pca/moderation) from " + "math_main for zid=%s", + zid, + ) + except Exception: + logger.exception( + "from_dict restore failed for zid=%s; cold start", zid + ) + conv = None + if conv is None: + # NB the nonzero seed dodges the constructor's falsy-0 -> wall-clock + # fallback; immediately floor to 0 afterwards (Clojure's floor, + # conversation.clj:161-165) so a zero-votes conversation emits + # lastVoteTimestamp=0, not the internal seed. Any real vote advances + # it via max() in update_votes. + conv = Conversation(str(zid), last_updated=1) + conv.last_updated = 0 + + votes = self._pg.poll_votes(zid, None) # full history, ordered, sign-flipped + if votes: + last_ts = advance_watermark( + conv.last_updated, (v.get("created") for v in votes) + ) + conv = conv.update_votes( + {"votes": votes, "lastVoteTimestamp": last_ts}, recompute=False + ) + + mods = self._pg.poll_moderation(zid, None) + conv = conv.update_moderation(mods, recompute=False) + + conv = conv.recompute() + return conv + + # -- error handling ----------------------------------------------------- # + def _on_engine_error( + self, zid: int, coalesced: CoalescedBatch, error: BaseException + ) -> None: + dump_error(zid, self._convs.get(zid), coalesced, error, self.config.dump_dir) + attempts = self._retry_counts.get(zid, 0) + 1 + self._retry_counts[zid] = attempts + if attempts <= self.config.retry_cap: + logger.error( + "Conversation update failed for zid=%s (attempt %d/%d); retrying: %s", + zid, + attempts, + self.config.retry_cap, + error, + ) + self._requeue(zid, coalesced) + else: + logger.error( + "PARKING zid=%s after %d failed attempts (circuit breaker). " + "Last error: %s", + zid, + attempts, + error, + ) + self._parked.add(zid) + if self._pool is not None: + self._pool.park(zid) + + def _requeue(self, zid: int, coalesced: CoalescedBatch) -> None: + if self._pool is None: + return + if coalesced.votes: + self._pool.submit(zid, VOTES, list(coalesced.votes)) + if coalesced.moderation: + self._pool.submit(zid, MODERATION, list(coalesced.moderation)) diff --git a/delphi/polismath/poller/worker_pool.py b/delphi/polismath/poller/worker_pool.py new file mode 100644 index 000000000..e9981c6c7 --- /dev/null +++ b/delphi/polismath/poller/worker_pool.py @@ -0,0 +1,149 @@ +"""Per-conversation serialized worker pool with batch coalescing. + +Reproduces the Clojure conv-actor semantics (conv_man.clj) without core.async: + + * Each zid is processed by AT MOST ONE thread at a time (strict per-zid + serialization) — the analog of one go-loop per conv (go-act!, :351-371). + * Before processing, ALL queued batches for that zid are drained and merged + (take-all!, :227-234) and split by message-type into a fixed + [votes, moderation] order (split-batches :247-257, go-act! :368-370). + * Different zids run concurrently up to ``max_workers`` (bounded pool) — the + analog of many lightweight go-loops, capped for a thread-based runtime. +""" + +import logging +import threading +from collections import deque +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass, field +from typing import Any, Callable, Deque, Dict, List, Set, Tuple + +logger = logging.getLogger(__name__) + +# A queued message is (message_type, batch) where message_type is +# "votes" | "moderation" and batch is a list of rows. +Message = Tuple[str, List[Any]] + +VOTES = "votes" +MODERATION = "moderation" + + +@dataclass +class CoalescedBatch: + """The merged work for one processing cycle of a single zid.""" + + votes: List[Any] = field(default_factory=list) + moderation: List[Any] = field(default_factory=list) + + def has_work(self) -> bool: + return bool(self.votes) or bool(self.moderation) + + +def coalesce_messages(messages: List[Message]) -> CoalescedBatch: + """Merge queued (type, batch) messages into one CoalescedBatch. + + Flattens every ``votes`` batch into one list (first-appearance order + preserved) and every ``moderation`` batch into another, mirroring Clojure + ``split-batches`` grouping by :message-type then flattening each group. + Processing order (votes before moderation) is imposed by the consumer, which + always applies ``.votes`` before ``.moderation``. + """ + votes: List[Any] = [] + moderation: List[Any] = [] + for message_type, batch in messages: + if message_type == VOTES: + votes.extend(batch) + elif message_type == MODERATION: + moderation.extend(batch) + else: # pragma: no cover - defensive; unknown types ignored like Clojure + logger.warning("Ignoring unknown message-type %r", message_type) + return CoalescedBatch(votes=votes, moderation=moderation) + + +class ConversationWorkerPool: + """Bounded pool that serializes work per zid and coalesces queued batches. + + Args: + process_fn: callable(zid, CoalescedBatch) invoked once per drained cycle. + max_workers: max concurrent zids processed at once. + """ + + def __init__( + self, + process_fn: Callable[[int, CoalescedBatch], None], + max_workers: int = 4, + ): + self._process_fn = process_fn + self._executor = ThreadPoolExecutor( + max_workers=max_workers, thread_name_prefix="conv-worker" + ) + self._queues: Dict[int, Deque[Message]] = {} + self._active: Set[int] = set() + self._parked: Set[int] = set() + self._lock = threading.Lock() + self._idle = threading.Condition(self._lock) + self._closed = False + + def park(self, zid: int) -> None: + """Stop processing a zid (circuit breaker). Queued/future work dropped.""" + with self._lock: + self._parked.add(zid) + self._queues.pop(zid, None) + + def unpark(self, zid: int) -> None: + """Re-enable processing for a previously parked zid (self-heal on new + work). The Clojure retry-chan self-heals on the next message; park is + transient across cycles, not a permanent death sentence.""" + with self._lock: + self._parked.discard(zid) + + def is_parked(self, zid: int) -> bool: + with self._lock: + return zid in self._parked + + def submit(self, zid: int, message_type: str, batch: List[Any]) -> None: + """Queue a batch for a zid; ensure exactly one worker drains it.""" + with self._lock: + if self._closed or zid in self._parked: + return + self._queues.setdefault(zid, deque()).append((message_type, batch)) + if zid not in self._active: + self._active.add(zid) + self._executor.submit(self._run, zid) + + def _run(self, zid: int) -> None: + while True: + with self._lock: + q = self._queues.get(zid) + if not q or zid in self._parked: + # Nothing left (or parked mid-flight): release the zid. + self._queues.pop(zid, None) + self._active.discard(zid) + self._idle.notify_all() + return + messages = list(q) + q.clear() + + coalesced = coalesce_messages(messages) + if coalesced.has_work(): + try: + self._process_fn(zid, coalesced) + except Exception: # pragma: no cover - process_fn owns its errors + logger.exception("Unhandled error processing zid=%s", zid) + # loop: re-check for messages that arrived while we were processing + + def join(self, timeout: float = 30.0) -> bool: + """Block until all queues are drained and no worker is active. + + Returns True if fully idle, False on timeout. For tests / graceful stop. + """ + with self._idle: + return self._idle.wait_for( + lambda: not self._active and not any(self._queues.values()), + timeout=timeout, + ) + + def shutdown(self, wait: bool = True) -> None: + with self._lock: + self._closed = True + self._executor.shutdown(wait=wait) diff --git a/delphi/pyproject.toml b/delphi/pyproject.toml index 682e5cefa..2db8c3986 100644 --- a/delphi/pyproject.toml +++ b/delphi/pyproject.toml @@ -145,6 +145,7 @@ filterwarnings = [ markers = [ "local_dataset: mark test as using local (non-committed) datasets from real_data/.local/", "clojure_comparison: mark test as comparing with Clojure reference implementation (can be excluded with '-m \"not clojure_comparison\"')", + "integration: mark test as an opt-in integration test needing a real Postgres (self-skips if docker/port unavailable)", ] [tool.coverage.run] diff --git a/delphi/scripts/math_poller.py b/delphi/scripts/math_poller.py new file mode 100644 index 000000000..ead4b0a7f --- /dev/null +++ b/delphi/scripts/math_poller.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python3 +"""Math Poller CLI — the Python replacement for the Clojure math container. + +Polls Postgres for votes/moderation, maintains per-conversation math state in +memory, and writes math_main / math_bidtopid / math_ptptstats under one +math_env. See polismath.poller (package docstring) and +delphi/docs/MATH_POLLER_DESIGN.md. + +Usage: + uv run python scripts/math_poller.py # run forever (SIGTERM stops) + uv run python scripts/math_poller.py --once # one poll cycle then exit +""" + +import argparse +import logging +import os +import signal +import sys + +from polismath.database.postgres import PostgresClient, PostgresConfig +from polismath.poller.service import MathPollerService, PollerConfig + + +def _configure_logging() -> None: + level = os.environ.get("LOG_LEVEL", "INFO").upper() + logging.basicConfig( + level=getattr(logging, level, logging.INFO), + format="%(asctime)s %(levelname)s [%(threadName)s] %(name)s: %(message)s", + ) + + +def _build_service(config: PollerConfig) -> MathPollerService: + if not config.database_url: + print("DATABASE_URL is required", file=sys.stderr) + raise SystemExit(2) + pg = PostgresClient(PostgresConfig(url=config.database_url, math_env=config.math_env)) + pg.initialize() + return MathPollerService(pg, config) + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description="Polis Python math poller") + parser.add_argument( + "--once", + action="store_true", + help="Run a single vote+moderation poll cycle, block until processed, exit.", + ) + args = parser.parse_args(argv) + + _configure_logging() + log = logging.getLogger("math_poller") + + config = PollerConfig.from_env() + service = _build_service(config) + + if args.once: + log.info("Running a single poll cycle (--once)") + service.poll_once() + service.stop() + return 0 + + # Graceful shutdown on SIGTERM/SIGINT (docker stop, Ctrl-C). + def _handle_signal(signum, _frame): + log.info("Received signal %s; stopping poller", signum) + service._stop.set() + + signal.signal(signal.SIGTERM, _handle_signal) + signal.signal(signal.SIGINT, _handle_signal) + + log.info( + "Starting math poller: math_env=%s vote_interval=%dms mod_interval=%dms " + "pool=%d", + config.math_env, + config.vote_interval_ms, + config.mod_interval_ms, + config.worker_pool_size, + ) + service.run_forever() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/delphi/tests/conftest.py b/delphi/tests/conftest.py index 3632e438e..68d026704 100644 --- a/delphi/tests/conftest.py +++ b/delphi/tests/conftest.py @@ -25,6 +25,23 @@ from tests.common_utils import load_votes, load_comments +@pytest.fixture(autouse=True) +def _guard_engine_mode_env(): + """Restore POLISMATH_ENGINE_MODE around every test. + + Production code (e.g. MathPollerService.apply_engine_mode) writes this var + straight into os.environ; without this guard a single test exercising that + path leaks clojure-legacy mode into every later test in the same worker, + flipping in-conv/warm-start semantics suite-wide (bit us in CI on #2637). + """ + prev = os.environ.get("POLISMATH_ENGINE_MODE") + yield + if prev is None: + os.environ.pop("POLISMATH_ENGINE_MODE", None) + else: + os.environ["POLISMATH_ENGINE_MODE"] = prev + + def require_dynamodb( endpoint: str | None = None, timeout: float = 3.0, diff --git a/delphi/tests/poller/__init__.py b/delphi/tests/poller/__init__.py new file mode 100644 index 000000000..03c0c52cb --- /dev/null +++ b/delphi/tests/poller/__init__.py @@ -0,0 +1 @@ +"""Unit tests for the Python math poller (polismath.poller).""" diff --git a/delphi/tests/poller/test_coalescing.py b/delphi/tests/poller/test_coalescing.py new file mode 100644 index 000000000..bd0fd4f3d --- /dev/null +++ b/delphi/tests/poller/test_coalescing.py @@ -0,0 +1,59 @@ +"""Per-zid batch coalescing. + +Mirrors Clojure conv_man.clj: + - take-all! (:227-234) drains every queued batch, + - split-batches (:247-257) groups by :message-type and flattens each group, + - go-act! (:368-370) then processes types in the fixed order [:votes :moderation]. +""" + +from polismath.poller.worker_pool import coalesce_messages, CoalescedBatch + + +class TestCoalesceMessages: + def test_single_vote_batch(self): + c = coalesce_messages([("votes", [{"pid": "1", "tid": "1", "vote": 1}])]) + assert c.votes == [{"pid": "1", "tid": "1", "vote": 1}] + assert c.moderation == [] + + def test_multiple_vote_batches_merge_in_arrival_order(self): + # Clojure split-batches flattens all :votes batches into one sequence. + c = coalesce_messages( + [ + ("votes", [{"pid": "1"}, {"pid": "2"}]), + ("votes", [{"pid": "3"}]), + ] + ) + assert c.votes == [{"pid": "1"}, {"pid": "2"}, {"pid": "3"}] + assert c.moderation == [] + + def test_moderation_batches_merge(self): + c = coalesce_messages( + [ + ("moderation", [{"tid": "1", "mod": -1}]), + ("moderation", [{"tid": "2", "mod": 1}]), + ] + ) + assert c.moderation == [{"tid": "1", "mod": -1}, {"tid": "2", "mod": 1}] + assert c.votes == [] + + def test_interleaved_batches_separate_by_type_preserving_vote_order(self): + # Interleaved arrival [votes, moderation, votes] -> votes merged across, + # moderation kept separate. Votes are still in first-appearance order. + c = coalesce_messages( + [ + ("votes", [{"pid": "1"}]), + ("moderation", [{"tid": "9"}]), + ("votes", [{"pid": "2"}]), + ] + ) + assert c.votes == [{"pid": "1"}, {"pid": "2"}] + assert c.moderation == [{"tid": "9"}] + + def test_empty_message_list(self): + c = coalesce_messages([]) + assert c == CoalescedBatch(votes=[], moderation=[]) + + def test_has_work_true_when_any_batch(self): + assert coalesce_messages([("votes", [{"pid": "1"}])]).has_work() is True + assert coalesce_messages([("moderation", [{"tid": "1"}])]).has_work() is True + assert coalesce_messages([]).has_work() is False diff --git a/delphi/tests/poller/test_error_path.py b/delphi/tests/poller/test_error_path.py new file mode 100644 index 000000000..aa113eb5e --- /dev/null +++ b/delphi/tests/poller/test_error_path.py @@ -0,0 +1,136 @@ +"""Error handling per design §3: on a failing conv update, dump conv+batch to an +errorconv JSON, retry once, then park the zid (circuit breaker). + +Mirrors Clojure handle-errors (conv_man.clj:291-323): conv-update-dump + requeue +to the retry-chan. Our retry_cap=1 caps replays before parking. +""" + +import json + +from polismath.poller.service import MathPollerService, PollerConfig +from polismath.poller.worker_pool import CoalescedBatch +from unittest.mock import MagicMock + + +def _service(tmp_path, retry_cap=1): + pg = MagicMock() + cfg = PollerConfig(dump_dir=str(tmp_path), retry_cap=retry_cap) + svc = MathPollerService(pg, cfg) + svc._pool = MagicMock() # capture requeue / park without real threads + return svc + + +def _dumps(tmp_path, zid): + return sorted(tmp_path.glob(f"errorconv-zid{zid}-*.json")) + + +class TestErrorPath: + def test_first_failure_dumps_and_retries(self, tmp_path, monkeypatch): + svc = _service(tmp_path) + monkeypatch.setattr( + svc, "_run_engine", + lambda zid, c: (_ for _ in ()).throw(RuntimeError("kaboom")), + ) + batch = CoalescedBatch(votes=[{"pid": "1", "tid": "1", "vote": 1}], moderation=[]) + + svc._handle_zid(5, batch) + + dumps = _dumps(tmp_path, 5) + assert len(dumps) == 1, "a dump file must be written on failure" + # dump contains the batch + error + traceback + payload = json.loads(dumps[0].read_text()) + assert payload["zid"] == 5 + assert "kaboom" in payload["error"] + assert payload["batch"]["votes"] == [{"pid": "1", "tid": "1", "vote": 1}] + # retry: batch requeued, zid NOT parked yet + svc._pool.submit.assert_called() + assert 5 not in svc._parked + svc._pool.park.assert_not_called() + + def test_second_failure_parks_zid(self, tmp_path, monkeypatch): + svc = _service(tmp_path) + monkeypatch.setattr( + svc, "_run_engine", + lambda zid, c: (_ for _ in ()).throw(RuntimeError("boom")), + ) + batch = CoalescedBatch(votes=[{"pid": "1"}], moderation=[]) + + svc._handle_zid(5, batch) # attempt 1 -> retry + svc._handle_zid(5, batch) # attempt 2 -> park (exceeds retry_cap=1) + + assert 5 in svc._parked + svc._pool.park.assert_called_once_with(5) + assert len(_dumps(tmp_path, 5)) == 2 # dumped on each failure + + def test_parked_zid_is_skipped(self, tmp_path, monkeypatch): + svc = _service(tmp_path) + monkeypatch.setattr( + svc, "_run_engine", + lambda zid, c: (_ for _ in ()).throw(RuntimeError("boom")), + ) + batch = CoalescedBatch(votes=[{"pid": "1"}], moderation=[]) + svc._handle_zid(5, batch) + svc._handle_zid(5, batch) # now parked (2 dumps) + svc._handle_zid(5, batch) # skipped: no engine call, no new dump + assert len(_dumps(tmp_path, 5)) == 2 + + def test_success_clears_retry_counter(self, tmp_path, monkeypatch): + svc = _service(tmp_path) + calls = {"n": 0} + + def flaky(zid, c): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("transient") + # succeeds on retry + + monkeypatch.setattr(svc, "_run_engine", flaky) + batch = CoalescedBatch(votes=[{"pid": "1"}], moderation=[]) + svc._handle_zid(7, batch) # fail -> retry + svc._handle_zid(7, batch) # success -> counter cleared + assert 7 not in svc._parked + assert svc._retry_counts.get(7) is None + + def test_new_batch_unparks_a_parked_zid(self, tmp_path, monkeypatch): + """T8: a parked zid self-heals when a NEW batch arrives on the next poll + cycle (Clojure retry-chan equivalent) — a transient blip must not leave + the zid dead until process restart.""" + svc = _service(tmp_path) + monkeypatch.setattr( + svc, "_run_engine", + lambda zid, c: (_ for _ in ()).throw(RuntimeError("boom")), + ) + batch = CoalescedBatch(votes=[{"pid": "1"}], moderation=[]) + svc._handle_zid(5, batch) # attempt 1 -> retry + svc._handle_zid(5, batch) # attempt 2 -> park + assert 5 in svc._parked + + # A new poll cycle delivers a fresh batch for zid 5. + svc._pg.poll_votes_since.return_value = [{"zid": 5, "created": 100}] + svc._vote_wm = 0 + svc._poll_votes_once() + + assert 5 not in svc._parked + assert svc._retry_counts.get(5) is None + svc._pool.unpark.assert_called_once_with(5) + svc._pool.submit.assert_called_with(5, "votes", [{"zid": 5, "created": 100}]) + + +def test_pool_unpark_reenables_dispatch(): + """T8: ConversationWorkerPool.unpark re-enables dispatch for a parked zid.""" + from polismath.poller.worker_pool import ConversationWorkerPool, VOTES + + seen = [] + pool = ConversationWorkerPool(lambda z, c: seen.append(z), max_workers=1) + try: + pool.park(5) + pool.submit(5, VOTES, [1]) # dropped while parked + assert pool.join(timeout=2) + assert seen == [] + pool.unpark(5) + assert not pool.is_parked(5) + pool.submit(5, VOTES, [1]) # now dispatched + assert pool.join(timeout=2) + assert seen == [5] + finally: + pool.shutdown() diff --git a/delphi/tests/poller/test_integration_postgres.py b/delphi/tests/poller/test_integration_postgres.py new file mode 100644 index 000000000..5ea26e2b0 --- /dev/null +++ b/delphi/tests/poller/test_integration_postgres.py @@ -0,0 +1,219 @@ +"""End-to-end integration test for the math poller against a real Postgres. + +OPT-IN and self-skipping (like tests/test_postgres_real_data.py): it provisions +a THROWAWAY postgres:17 container on port 5435 (NEVER the host's live 5432), +applies server/postgres/migrations/000000_initial.sql, seeds one conversation, +and drives poll -> compute -> write, asserting: + + * a math_main row appears under the poller's math_env (shadow isolation), + * math_bidtopid + math_ptptstats share the cycle's math_tick, + * caching_tick / math_tick behave per the Clojure-exact SQL, + * a fresh service instance resumes and advances the tick (restart-resumes). + +If docker is unavailable or port 5435 is busy, the whole module is skipped with +a clear reason. +""" + +import os +import shutil +import subprocess +import time +import uuid + +import pytest + +pytestmark = pytest.mark.integration + +MIGRATION = os.path.join( + os.path.dirname(__file__), + "..", "..", "..", "server", "postgres", "migrations", "000000_initial.sql", +) +PORT = 5435 +DB_URL = f"postgresql://postgres:test@localhost:{PORT}/postgres" + + +def _docker() -> str: + exe = shutil.which("docker") + if not exe: + pytest.skip("docker not available") + return exe + + +def _run(*args, **kwargs): + return subprocess.run(args, capture_output=True, text=True, **kwargs) + + +@pytest.fixture(scope="module") +def pg_url(): + docker = _docker() + migration = os.path.abspath(MIGRATION) + if not os.path.exists(migration): + pytest.skip(f"migration not found: {migration}") + + name = f"delphi-poller-it-{uuid.uuid4().hex[:8]}" + started = _run( + docker, "run", "--rm", "-d", "--name", name, + "-p", f"{PORT}:5432", "-e", "POSTGRES_PASSWORD=test", "postgres:17", + ) + if started.returncode != 0: + pytest.skip(f"could not start postgres container (port {PORT} busy?): " + f"{started.stderr.strip()}") + cid = started.stdout.strip() + try: + # Wait for readiness. + deadline = time.time() + 40 + ready = False + while time.time() < deadline: + if _run(docker, "exec", cid, "pg_isready", "-U", "postgres").returncode == 0: + ready = True + break + time.sleep(1) + if not ready: + pytest.skip("postgres container did not become ready in time") + + # Apply the full initial migration. + with open(migration, "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 failed to apply: {applied.stderr[-500:]}") + + yield DB_URL + finally: + _run(docker, "stop", cid) + + +def _seed_conversation(engine, zid=1, n_ptpts=8, n_cmts=5): + """Seed one conversation with FK enforcement disabled for the session.""" + import sqlalchemy as sa + + now = int(time.time() * 1000) + # Votes are recent (within the poll window); comments were moderated LONG ago + # (older than the moderation watermark) so the moderation loop dispatches + # nothing and each poll_once deterministically triggers exactly one + # (votes -> load_or_init) write cycle. Full moderation state is still read + # by load_or_init's poll_moderation(zid, None), so the comments are exercised. + vote_created = now - 60_000 + old_modified = now - 2 * 24 * 60 * 60 * 1000 # 2 days ago + with engine.begin() as conn: + conn.execute(sa.text("SET session_replication_role = replica")) + conn.execute(sa.text("INSERT INTO conversations (zid) VALUES (:zid)"), + {"zid": zid}) + for p in range(n_ptpts): + conn.execute( + sa.text("INSERT INTO participants (pid, uid, zid, created, mod) " + "VALUES (:pid, :uid, :zid, :created, 0)"), + {"pid": p, "uid": 1000 + p, "zid": zid, "created": old_modified}, + ) + for t in range(n_cmts): + conn.execute( + sa.text("INSERT INTO comments (tid, zid, pid, uid, txt, mod, is_meta, " + "created, modified) VALUES " + "(:tid, :zid, 0, 1000, :txt, 0, false, :created, :modified)"), + {"tid": t, "zid": zid, "txt": f"comment {t}", + "created": old_modified, "modified": old_modified}, + ) + created = vote_created + # Raw DB vote signs: AGREE=-1, DISAGREE=+1. Two opposing camps. + for p in range(n_ptpts): + raw = -1 if p % 2 == 0 else 1 + for t in range(n_cmts): + created += 1 + conn.execute( + sa.text("INSERT INTO votes (zid, pid, tid, vote, created) " + "VALUES (:zid, :pid, :tid, :vote, :created)"), + {"zid": zid, "pid": p, "tid": t, "vote": raw, "created": created}, + ) + conn.execute(sa.text("SET session_replication_role = DEFAULT")) + + +def _make_service(url, math_env): + from polismath.database.postgres import PostgresClient, PostgresConfig + from polismath.poller.service import MathPollerService, PollerConfig + + pg = PostgresClient(PostgresConfig(url=url, math_env=math_env, ssl_mode="disable")) + pg.initialize() + cfg = PollerConfig( + database_url=url, math_env=math_env, poll_from_days_ago=1, + worker_pool_size=2, engine_mode="clojure-legacy", + ) + return MathPollerService(pg, cfg), pg + + +def _fetch_one(engine, sql, params): + import sqlalchemy as sa + + with engine.connect() as conn: + row = conn.execute(sa.text(sql), params).mappings().first() + return dict(row) if row else None + + +class TestPollerIntegration: + def test_end_to_end_poll_compute_write_and_restart(self, pg_url): + """Two phases in one test (single container, xdist-safe): + (1) poll -> compute -> write -> row visible under math_env + shadow + isolation + shared math_tick; (2) fresh service resumes and advances + the tick (restart-resumes).""" + import sqlalchemy as sa + + engine = sa.create_engine(pg_url) + _seed_conversation(engine, zid=1) + math_env = "delphi_it" + + # ---- Phase 1: first poll cycle ------------------------------------- + service, _pg = _make_service(pg_url, math_env) + service.poll_once() + service.stop() + + main = _fetch_one( + engine, + "select zid, math_env, caching_tick, math_tick, data from math_main " + "where zid = :zid and math_env = :me", + {"zid": 1, "me": math_env}, + ) + assert main is not None, "poller must write a math_main row" + assert main["data"] is not None + # First write: caching_tick = COALESCE(max+1, 1) = 1; math_tick default 0. + assert main["caching_tick"] == 1 + assert main["math_tick"] == 0 + + # Shadow isolation: nothing written under a different math_env. + other = _fetch_one( + engine, + "select zid from math_main where zid = :zid and math_env = :me", + {"zid": 1, "me": "prod"}, + ) + assert other is None + + # bidtopid + ptptstats share the cycle's math_tick. + bid = _fetch_one( + engine, + "select math_tick, data from math_bidtopid where zid=:zid and math_env=:me", + {"zid": 1, "me": math_env}, + ) + pts = _fetch_one( + engine, + "select math_tick from math_ptptstats where zid=:zid and math_env=:me", + {"zid": 1, "me": math_env}, + ) + assert bid is not None and pts is not None + assert bid["math_tick"] == main["math_tick"] == pts["math_tick"] + assert isinstance(bid["data"]["bidToPid"], list) # list of pid-lists + + # ---- Phase 2: restart resumes and advances the tick ---------------- + before = main + service2, _pg2 = _make_service(pg_url, math_env) # fresh in-memory cache + service2.poll_once() + service2.stop() + + after = _fetch_one( + engine, + "select caching_tick, math_tick from math_main where zid=:zid and math_env=:me", + {"zid": 1, "me": math_env}, + ) + # Atomic tick advanced; caching_tick advanced (MAX+1) -> resumed cleanly. + assert after["math_tick"] == before["math_tick"] + 1 + assert after["caching_tick"] == before["caching_tick"] + 1 diff --git a/delphi/tests/poller/test_load_or_init.py b/delphi/tests/poller/test_load_or_init.py new file mode 100644 index 000000000..609221d11 --- /dev/null +++ b/delphi/tests/poller/test_load_or_init.py @@ -0,0 +1,200 @@ +"""load-or-init + the from_dict restoration finding. + +These tests LOCK the finding documented in polismath/poller/__init__.py: +``Conversation.from_dict`` restores warm state (pca, moderation, counts) but NOT +the rating matrices or base_clusters, so load-or-init must ALWAYS rebuild the +matrices from the full vote history (mirroring conv_man.clj:188-207). +""" + +import time +from unittest.mock import MagicMock + +from polismath.conversation.conversation import Conversation +from polismath.poller.service import MathPollerService, PollerConfig + + +def _empty_mods(): + return {"mod_out_tids": [], "mod_in_tids": [], "meta_tids": [], "mod_out_ptpts": []} + + +def _build_votes(n_ptpts=8, n_cmts=5, created0=1000): + """Two opposing camps so PCA + base clusters are non-trivial.""" + votes = [] + created = created0 + for p in range(n_ptpts): + camp = 1 if p % 2 == 0 else -1 + for t in range(n_cmts): + votes.append( + {"pid": str(p), "tid": str(t), "vote": camp, "created": created} + ) + created += 1 + return votes + + +class TestFromDictFinding: + def test_from_dict_restores_pca_and_moderation_but_not_matrices(self): + conv = Conversation("42") + conv = conv.update_moderation({"mod_out_tids": ["3"]}, recompute=False) + conv = conv.update_votes( + {"votes": _build_votes(), "lastVoteTimestamp": 9999}, recompute=True + ) + + # Preconditions: the live conv has populated matrices + pca + clusters. + assert conv.raw_rating_mat.size > 0 + assert conv.pca is not None + + blob = conv.to_dict() + restored = Conversation.from_dict(blob) + + # RESTORED (warm state): pca, moderation, counts. + assert restored.pca is not None + assert set(restored.mod_out_tids) == {"3"} + assert restored.participant_count == conv.participant_count + + # NOT RESTORED: the vote matrices and base_clusters — hence a full + # rebuild is mandatory in load-or-init. + assert restored.raw_rating_mat.size == 0 + assert restored.rating_mat.size == 0 + assert restored.base_clusters == [] + + +class TestLoadOrInit: + def test_cold_start_when_no_math_main_row(self): + pg = MagicMock() + pg.load_math_main.return_value = None + pg.poll_votes.return_value = _build_votes() + pg.poll_moderation.return_value = { + "mod_out_tids": [], "mod_in_tids": [], "meta_tids": [], "mod_out_ptpts": [] + } + svc = MathPollerService(pg, PollerConfig()) + + conv = svc._load_or_init(42) + + assert isinstance(conv, Conversation) + # Full-history rebuild always runs (offset-0 analog). + pg.poll_votes.assert_called_once_with(42, None) + pg.poll_moderation.assert_called_once_with(42, None) + assert conv.raw_rating_mat.size > 0 + + def test_warm_restore_then_full_rebuild(self): + # Produce a real math_main blob from a computed conversation. + seed = Conversation("42") + seed = seed.update_votes( + {"votes": _build_votes(), "lastVoteTimestamp": 9999}, recompute=True + ) + blob = seed.to_dict() + + pg = MagicMock() + pg.load_math_main.return_value = {"zid": 42, "data": blob} + pg.poll_votes.return_value = _build_votes() + pg.poll_moderation.return_value = { + "mod_out_tids": [], "mod_in_tids": [], "meta_tids": [], "mod_out_ptpts": [] + } + svc = MathPollerService(pg, PollerConfig()) + + conv = svc._load_or_init(42) + + assert isinstance(conv, Conversation) + # Even with a warm row, matrices are rebuilt from full vote history. + pg.poll_votes.assert_called_once_with(42, None) + assert conv.raw_rating_mat.size > 0 + assert conv.pca is not None + + def test_from_dict_failure_falls_back_to_cold(self, monkeypatch): + pg = MagicMock() + pg.load_math_main.return_value = {"zid": 42, "data": {"garbage": object()}} + pg.poll_votes.return_value = _build_votes() + pg.poll_moderation.return_value = { + "mod_out_tids": [], "mod_in_tids": [], "meta_tids": [], "mod_out_ptpts": [] + } + + # Force from_dict to raise to exercise the guarded fallback. + def boom(cls, data): + raise ValueError("bad blob") + + monkeypatch.setattr(Conversation, "from_dict", classmethod(boom)) + svc = MathPollerService(pg, PollerConfig()) + + conv = svc._load_or_init(42) + assert isinstance(conv, Conversation) + assert conv.raw_rating_mat.size > 0 + + +class TestLastVoteTimestampSeed: + """T7: a cold rebuild must resolve last_updated to true max(created), not the + wall-clock leaked by Conversation's `last_updated or now` footgun (which + advance_watermark can never regress). Clojure floors at 0 (conversation.clj:161-165).""" + + def test_cold_start_last_updated_is_max_created_not_wall_clock(self): + pg = MagicMock() + pg.load_math_main.return_value = None + votes = _build_votes(created0=1000) + pg.poll_votes.return_value = votes + pg.poll_moderation.return_value = _empty_mods() + svc = MathPollerService(pg, PollerConfig()) + + wall_clock_before = int(time.time() * 1000) + conv = svc._load_or_init(42) + + max_created = max(v["created"] for v in votes) + assert conv.last_updated == max_created + # The historical timestamps are ~1e3 ms; a wall-clock leak would be ~1e12. + assert conv.last_updated < wall_clock_before + + def test_warm_restore_last_updated_from_history_not_wall_clock(self): + seed = Conversation("42").update_votes( + {"votes": _build_votes(), "lastVoteTimestamp": 9999}, recompute=True + ) + blob = seed.to_dict() + votes = _build_votes(created0=1000) + max_created = max(v["created"] for v in votes) + + pg = MagicMock() + # The persisted row carries a correct (historical) last_vote_timestamp. + pg.load_math_main.return_value = { + "zid": 42, "data": blob, "last_vote_timestamp": max_created, + } + pg.poll_votes.return_value = votes + pg.poll_moderation.return_value = _empty_mods() + svc = MathPollerService(pg, PollerConfig()) + + wall_clock_before = int(time.time() * 1000) + conv = svc._load_or_init(42) + + assert conv.last_updated == max_created + assert conv.last_updated < wall_clock_before + + def test_zero_votes_cold_start_floors_last_updated_to_zero(self): + """A conversation with NO votes at all (e.g. moderation-only activity) + must emit lastVoteTimestamp=0 (the Clojure floor, conversation.clj:161-165), + not the internal nonzero constructor-dodge seed.""" + pg = MagicMock() + pg.load_math_main.return_value = None + pg.poll_votes.return_value = [] + pg.poll_moderation.return_value = _empty_mods() + svc = MathPollerService(pg, PollerConfig()) + + conv = svc._load_or_init(42) + + assert conv.last_updated == 0 + assert conv.to_dict()["lastVoteTimestamp"] == 0 + + def test_persisted_zero_last_vote_timestamp_is_preserved(self): + """A legitimately persisted last_vote_timestamp of 0 must be preserved, + not coerced to 1 by a falsy-`or` default.""" + seed = Conversation("42").update_votes( + {"votes": _build_votes(), "lastVoteTimestamp": 9999}, recompute=True + ) + blob = seed.to_dict() + + pg = MagicMock() + pg.load_math_main.return_value = { + "zid": 42, "data": blob, "last_vote_timestamp": 0, + } + pg.poll_votes.return_value = [] + pg.poll_moderation.return_value = _empty_mods() + svc = MathPollerService(pg, PollerConfig()) + + conv = svc._load_or_init(42) + + assert conv.last_updated == 0 diff --git a/delphi/tests/poller/test_math_writer.py b/delphi/tests/poller/test_math_writer.py new file mode 100644 index 000000000..2b66078a3 --- /dev/null +++ b/delphi/tests/poller/test_math_writer.py @@ -0,0 +1,150 @@ +"""Writer tests: bidToPid derivation + the four Postgres writes. + +Verifies fidelity to the Clojure writers: + - upload-math-main caching_tick = COALESCE((select max(caching_tick)+1 ...),1) + (postgres.clj:323-338) + - inc-math-tick atomic INSERT ... ON CONFLICT ... math_tick+1 RETURNING + (postgres.clj:292-295) + - prep-bidToPid shape {:zid :bidToPid :lastVoteTimestamp} where bidToPid is a + vector of member-vectors sorted by base cluster id (conv_man.clj:35-40, + conversation.clj:585-586) + - write-conv-updates! writes math_main / math_bidtopid / math_ptptstats with + ONE shared math_tick (conv_man.clj:158-169). +""" + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from polismath.poller.math_writer import derive_bidtopid, MathWriter + + +def _fake_conv(zid=42, base_clusters=None, last_updated=1234567): + """Minimal stand-in exposing the public attributes the writer consumes.""" + conv = SimpleNamespace() + conv.conversation_id = str(zid) + conv.last_updated = last_updated + conv.base_clusters = base_clusters if base_clusters is not None else [] + conv.participant_info = {} + conv.to_dict = lambda: {"base-clusters": {"id": [], "members": []}, + "lastVoteTimestamp": last_updated} + return conv + + +class TestDeriveBidToPid: + def test_shape_is_list_of_member_lists_sorted_by_id(self): + # base_clusters intentionally out of id order to prove sorting. + conv = _fake_conv( + zid=7, + base_clusters=[ + {"id": 2, "members": ["30", "31"]}, + {"id": 0, "members": ["10", "11", "12"]}, + {"id": 1, "members": ["20"]}, + ], + ) + result = derive_bidtopid(conv, 7) + # bidToPid[i] must be the members of the base cluster whose id sorts to + # position i -> positionally aligned with base-clusters.id (ascending). + assert result["bidToPid"] == [["10", "11", "12"], ["20"], ["30", "31"]] + + def test_wrapper_keys_match_prep_bidToPid(self): + conv = _fake_conv(zid=7, base_clusters=[{"id": 0, "members": ["1"]}], + last_updated=999) + result = derive_bidtopid(conv, 7) + assert result["zid"] == 7 + assert result["lastVoteTimestamp"] == 999 + assert set(result.keys()) == {"zid", "bidToPid", "lastVoteTimestamp"} + + def test_empty_base_clusters_gives_empty_bidToPid(self): + conv = _fake_conv(zid=7, base_clusters=[]) + assert derive_bidtopid(conv, 7)["bidToPid"] == [] + + +class TestMathWriterSharedTick: + def test_all_writes_share_one_math_tick(self): + client = MagicMock() + client.increment_math_tick.return_value = 77 + conv = _fake_conv(zid=42, base_clusters=[{"id": 0, "members": ["1"]}]) + + writer = MathWriter(client) + writer.write_conv_updates(42, conv) + + # tick incremented exactly once for the zid + client.increment_math_tick.assert_called_once_with(42) + + # all three data writes carry the SAME tick value returned above + assert client.write_math_main.call_args.kwargs.get("math_tick") == 77 \ + or 77 in client.write_math_main.call_args.args + bidtopid_tick = client.write_math_bidtopid.call_args + ptptstats_tick = client.write_participant_stats.call_args + assert 77 in bidtopid_tick.args or bidtopid_tick.kwargs.get("math_tick") == 77 + assert 77 in ptptstats_tick.args or ptptstats_tick.kwargs.get("math_tick") == 77 + + def test_bidtopid_data_written_has_correct_shape(self): + client = MagicMock() + client.increment_math_tick.return_value = 1 + conv = _fake_conv(zid=42, base_clusters=[{"id": 0, "members": ["1", "2"]}]) + MathWriter(client).write_conv_updates(42, conv) + + # Find the data dict passed to write_math_bidtopid. + call = client.write_math_bidtopid.call_args + data = call.kwargs.get("data") + if data is None: + # positional: (zid, data, math_tick) + data = call.args[1] + assert data["bidToPid"] == [["1", "2"]] + + +class TestWriterSQLFidelity: + """The Clojure-exact SQL lives in PostgresClient; verify text + params via a + recorder that captures every raw query without touching a database.""" + + def _client_with_recorder(self): + from polismath.database.postgres import PostgresClient, PostgresConfig + + cfg = PostgresConfig(url="postgresql://u:p@h:5432/db", math_env="delphi") + client = PostgresClient(cfg) + calls = [] + + def recorder(sql, params=None): + calls.append((sql, params or {})) + # increment_math_tick reads [0]["math_tick"] off the result + return [{"math_tick": 5, "zid": 1}] + + # Writers persist via the committing _write_returning path (not query()). + client._write_returning = recorder # type: ignore[assignment] + client._initialized = True + return client, calls + + def test_increment_math_tick_is_atomic_upsert(self): + client, calls = self._client_with_recorder() + tick = client.increment_math_tick(42) + sql, params = calls[-1] + norm = " ".join(sql.lower().split()) + assert "insert into math_ticks" in norm + assert "on conflict" in norm + assert "math_tick" in norm and "+ 1" in norm.replace("+1", "+ 1") + assert "returning math_tick" in norm + assert tick == 5 + + def test_write_math_main_has_caching_tick_max_plus_one_subquery(self): + client, calls = self._client_with_recorder() + client.write_math_main( + 42, {"k": "v"}, last_vote_timestamp=111, math_tick=5 + ) + sql, params = calls[-1] + norm = " ".join(sql.lower().split()) + assert "insert into math_main" in norm + assert "coalesce" in norm + assert "max(caching_tick) + 1" in norm.replace("max(caching_tick)+1", + "max(caching_tick) + 1") + assert "on conflict" in norm + + def test_write_math_bidtopid_upsert(self): + client, calls = self._client_with_recorder() + client.write_math_bidtopid(42, {"bidToPid": [["1"]]}, math_tick=5) + sql, params = calls[-1] + norm = " ".join(sql.lower().split()) + assert "insert into math_bidtopid" in norm + assert "on conflict" in norm diff --git a/delphi/tests/poller/test_serialization.py b/delphi/tests/poller/test_serialization.py new file mode 100644 index 000000000..a5b078d87 --- /dev/null +++ b/delphi/tests/poller/test_serialization.py @@ -0,0 +1,78 @@ +"""Thread-safety: strict per-zid serialization + bounded cross-zid concurrency. + +Verifies the ConversationWorkerPool guarantee that mirrors Clojure's one-go-loop- +per-conv model: two batches for the SAME zid are never processed concurrently, +while DIFFERENT zids may run in parallel up to max_workers. +""" + +import threading +import time + +from polismath.poller.worker_pool import ConversationWorkerPool + + +class TestPerZidSerialization: + def test_same_zid_batches_never_interleave(self): + active = 0 + max_concurrent = 0 + call_count = 0 + lock = threading.Lock() + + def process(zid, coalesced): + nonlocal active, max_concurrent, call_count + with lock: + active += 1 + call_count += 1 + max_concurrent = max(max_concurrent, active) + time.sleep(0.03) + with lock: + active -= 1 + + pool = ConversationWorkerPool(process, max_workers=4) + # Submit in waves with a small gap so some batches arrive WHILE the zid + # is being processed -> forces >1 sequential process cycle for zid 7. + for i in range(6): + pool.submit(7, "votes", [{"i": i}]) + time.sleep(0.015) + + assert pool.join(timeout=10) is True + pool.shutdown() + + assert call_count >= 2, "expected multiple sequential cycles for the zid" + assert max_concurrent == 1, "same zid must never run on two workers at once" + + def test_different_zids_run_concurrently(self): + # A 2-party barrier only clears if two zids are processed at the same + # time; if the pool serialized across zids it would time out (broken). + barrier = threading.Barrier(2) + broken = [] + + def process(zid, coalesced): + try: + barrier.wait(timeout=5) + except threading.BrokenBarrierError as e: # pragma: no cover + broken.append(e) + + pool = ConversationWorkerPool(process, max_workers=2) + pool.submit(1, "votes", [{}]) + pool.submit(2, "votes", [{}]) + + assert pool.join(timeout=10) is True + pool.shutdown() + assert not broken, "distinct zids should be able to run concurrently" + + +class TestParking: + def test_parked_zid_is_not_processed(self): + seen = [] + + def process(zid, coalesced): + seen.append(zid) + + pool = ConversationWorkerPool(process, max_workers=2) + pool.park(9) + pool.submit(9, "votes", [{}]) + assert pool.join(timeout=5) is True + pool.shutdown() + assert 9 not in seen + assert pool.is_parked(9) is True diff --git a/delphi/tests/poller/test_service.py b/delphi/tests/poller/test_service.py new file mode 100644 index 000000000..d8721a90f --- /dev/null +++ b/delphi/tests/poller/test_service.py @@ -0,0 +1,128 @@ +"""Service-level dispatch: allow/block filtering, watermark advance on dispatch, +and engine-mode passthrough into the process environment.""" + +import os +from unittest.mock import MagicMock + +from polismath.poller.service import MathPollerService, PollerConfig +from polismath.utils.engine_mode import resolve_engine_mode + + +def _vote_row(zid, created, pid="1", tid="1"): + return {"zid": zid, "pid": pid, "tid": tid, "vote": 1, "created": created} + + +class TestDispatchFiltering: + def test_allowlist_only_dispatches_listed_zids(self): + pg = MagicMock() + pg.poll_votes_since.return_value = [ + _vote_row(1, 100), + _vote_row(2, 110), + _vote_row(3, 120), + ] + svc = MathPollerService(pg, PollerConfig(allowlist=[1, 3])) + svc._ensure_runtime() + svc._vote_wm = 0 + submitted = [] + svc._pool.submit = lambda zid, mt, batch: submitted.append((zid, mt)) + + svc._poll_votes_once() + + assert [z for z, _ in submitted] == [1, 3] + + def test_blocklist_excludes_listed_zids(self): + pg = MagicMock() + pg.poll_votes_since.return_value = [_vote_row(1, 100), _vote_row(2, 110)] + svc = MathPollerService(pg, PollerConfig(blocklist=[2])) + svc._ensure_runtime() + svc._vote_wm = 0 + submitted = [] + svc._pool.submit = lambda zid, mt, batch: submitted.append((zid, mt)) + + svc._poll_votes_once() + + assert [z for z, _ in submitted] == [1] + + def test_watermark_advances_past_all_rows_even_filtered(self): + # Clojure advances the watermark using max() over ALL polled rows and + # only the DISPATCH is filtered (poller.clj:27 vs :29-34). + pg = MagicMock() + pg.poll_votes_since.return_value = [_vote_row(1, 100), _vote_row(2, 999)] + svc = MathPollerService(pg, PollerConfig(allowlist=[1])) + svc._ensure_runtime() + svc._vote_wm = 0 + svc._pool.submit = lambda *a, **k: None + + svc._poll_votes_once() + assert svc._vote_wm == 999 + + def test_moderation_dispatch_and_watermark(self): + pg = MagicMock() + pg.poll_moderation_since.return_value = [ + {"zid": 5, "tid": 1, "modified": 200, "mod": -1, "is_meta": False}, + {"zid": 6, "tid": 2, "modified": 250, "mod": 1, "is_meta": False}, + ] + svc = MathPollerService(pg, PollerConfig()) + svc._ensure_runtime() + svc._mod_wm = 0 + submitted = [] + svc._pool.submit = lambda zid, mt, batch: submitted.append((zid, mt)) + + svc._poll_moderation_once() + assert {z for z, _ in submitted} == {5, 6} + assert all(mt == "moderation" for _, mt in submitted) + assert svc._mod_wm == 250 + + +class TestEngineModePassthrough: + def test_configured_mode_is_pushed_into_env(self, monkeypatch): + # apply_engine_mode() writes os.environ directly, which monkeypatch's + # delenv undo does NOT cover when the var was absent — restore by hand + # or the mode leaks into every later test in this worker. + monkeypatch.delenv("POLISMATH_ENGINE_MODE", raising=False) + try: + svc = MathPollerService( + MagicMock(), PollerConfig(engine_mode="clojure-legacy") + ) + resolved = svc.apply_engine_mode() + assert os.environ["POLISMATH_ENGINE_MODE"] == "clojure-legacy" + assert resolved == "clojure-legacy" + # The in-process compute resolves the SAME value at call time. + assert resolve_engine_mode() == "clojure-legacy" + finally: + os.environ.pop("POLISMATH_ENGINE_MODE", None) + + def test_no_configured_mode_leaves_compute_default(self, monkeypatch): + monkeypatch.delenv("POLISMATH_ENGINE_MODE", raising=False) + svc = MathPollerService(MagicMock(), PollerConfig(engine_mode=None)) + resolved = svc.apply_engine_mode() + assert resolved == "improved" # engine_mode.ENGINE_MODE_DEFAULT + + +class TestConvCacheEviction: + """T8: the in-memory conv registry never evicted (Clojure's 4h reboot was the + de-facto cap, which we dropped). LRU-evict beyond a configurable cap; an + evicted conv reloads from math_main + rebuilds on next touch.""" + + def test_lru_evicts_coldest_beyond_cap(self): + svc = MathPollerService(MagicMock(), PollerConfig(conv_cache_cap=2)) + svc._remember(1, object()) + svc._remember(2, object()) + assert list(svc._convs) == [1, 2] + + svc._remember(3, object()) # over cap -> evict coldest (1) + assert list(svc._convs) == [2, 3] + + svc._convs.move_to_end(2) # a touch on 2 makes it MRU + svc._remember(4, object()) # evict coldest (now 3) + assert list(svc._convs) == [2, 4] + + def test_cap_zero_never_evicts(self): + svc = MathPollerService(MagicMock(), PollerConfig(conv_cache_cap=0)) + for i in range(30): + svc._remember(i, object()) + assert len(svc._convs) == 30 + + def test_cap_from_env(self, monkeypatch): + monkeypatch.setenv("MATH_CONV_CACHE_CAP", "5") + assert PollerConfig.from_env().conv_cache_cap == 5 diff --git a/delphi/tests/poller/test_watermark.py b/delphi/tests/poller/test_watermark.py new file mode 100644 index 000000000..453592c12 --- /dev/null +++ b/delphi/tests/poller/test_watermark.py @@ -0,0 +1,74 @@ +"""Watermark advancement + zid allow/block filtering (poll-loop invariants). + +Mirrors the Clojure poll loop (math/src/polismath/poller.clj:22-37): + last-timestamp = (apply max 0 last-timestamp (map timestamp-key results)) +and the allow/block cond (poller.clj:30-32). +""" + +from polismath.poller.service import ( + advance_watermark, + should_process_zid, + initial_watermark, +) + + +class TestAdvanceWatermark: + def test_advances_to_max_of_batch(self): + # Clojure: (apply max 0 last-timestamp (map :created results)) + assert advance_watermark(100, [150, 120, 199, 130]) == 199 + + def test_strictly_greater_never_regresses_below_current(self): + # All timestamps below current watermark -> watermark unchanged. + assert advance_watermark(500, [100, 200, 499]) == 500 + + def test_empty_batch_leaves_watermark_unchanged(self): + assert advance_watermark(1234, []) == 1234 + + def test_uses_current_when_current_is_the_max(self): + assert advance_watermark(999, [10, 20]) == 999 + + def test_single_timestamp_above_current_advances(self): + assert advance_watermark(0, [42]) == 42 + + def test_returns_max_across_current_and_batch(self): + # Watermark should be the max of current and every timestamp in batch. + assert advance_watermark(300, [250, 700, 260]) == 700 + + def test_never_regresses_across_repeated_polls(self): + wm = initial_watermark(10, now_millis=1_000_000) + wm2 = advance_watermark(wm, [wm + 5, wm + 3]) + assert wm2 == wm + 5 + # A later poll that returns only older rows must NOT lower the watermark. + wm3 = advance_watermark(wm2, [wm + 1, wm + 4]) + assert wm3 == wm2 + + +class TestInitialWatermark: + def test_starts_poll_from_days_ago_back(self): + # 10 days ago = now - 10*86400*1000 ms (poller.clj:15). + now = 10_000_000_000 + wm = initial_watermark(10, now_millis=now) + assert wm == now - 10 * 24 * 60 * 60 * 1000 + + def test_zero_days_ago_is_now(self): + now = 555 + assert initial_watermark(0, now_millis=now) == 555 + + +class TestShouldProcessZid: + def test_no_lists_processes_everything(self): + assert should_process_zid(42, [], []) is True + + def test_allowlist_only_allows_listed(self): + # Clojure: allowlist takes priority; only listed zids pass. + assert should_process_zid(42, [42, 7], []) is True + assert should_process_zid(99, [42, 7], []) is False + + def test_blocklist_excludes_listed(self): + assert should_process_zid(42, [], [99, 100]) is True + assert should_process_zid(99, [], [99, 100]) is False + + def test_allowlist_takes_priority_over_blocklist(self): + # Clojure cond: allowlist branch evaluated first. + assert should_process_zid(42, [42], [42]) is True + assert should_process_zid(7, [42], [7]) is False diff --git a/docker-compose.yml b/docker-compose.yml index 3670fa2a3..792f51240 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -145,6 +145,59 @@ services: cpus: ${DELPHI_CONTAINER_CPUS:-2} restart: unless-stopped + # Python math poller — the eventual replacement for the Clojure `math` + # container. Profile-gated so it only runs when explicitly requested + # (`--profile delphi-math`). Reuses the delphi build target and overrides the + # command to run the poller CLI. Defaults to SHADOW mode: writes under a + # DISTINCT math_env (`delphi`) so its rows are invisible to the prod server + # (UNIQUE(zid, math_env)) — zero production risk while parity is validated. + # See delphi/docs/MATH_POLLER_DESIGN.md §4. + delphi-math-poller: + image: 050917022930.dkr.ecr.us-east-1.amazonaws.com/polis/delphi:latest + build: + context: ./delphi + target: final + labels: + polis_tag: ${TAG:-dev} + command: ["python", "scripts/math_poller.py"] + environment: + - DATABASE_URL=${DATABASE_URL} + - DATABASE_SSL_MODE=${DATABASE_SSL_MODE:-disable} + - POSTGRES_CONNECT_TIMEOUT=${POSTGRES_CONNECT_TIMEOUT:-30} + - LOG_LEVEL=${DELPHI_LOG_LEVEL:-INFO} + # Shadow-mode math_env (distinct from the Clojure math service's MATH_ENV). + - MATH_ENV=${DELPHI_MATH_ENV:-delphi} + # Poll cadences (ms) and boot window (days). + - POLL_VOTE_INTERVAL_MS=${POLL_VOTE_INTERVAL_MS:-1000} + - POLL_MOD_INTERVAL_MS=${POLL_MOD_INTERVAL_MS:-1000} + - POLL_FROM_DAYS_AGO=${POLL_FROM_DAYS_AGO:-10} + # Optional zid allow/block lists (comma-separated). + - POLL_ALLOWLIST=${POLL_ALLOWLIST:-} + - POLL_BLOCKLIST=${POLL_BLOCKLIST:-} + # Engine family: 'clojure-legacy' (the effective default below, for + # warm-start parity during shadow validation) or 'improved'. + - POLISMATH_ENGINE_MODE=${POLISMATH_ENGINE_MODE:-clojure-legacy} + # Per-zid serialized workers; concurrency across zids. + - MATH_WORKER_POOL_SIZE=${MATH_WORKER_POOL_SIZE:-4} + # Error dump dir + retry cap (dump -> retry -> park circuit breaker). + - MATH_POLLER_DUMP_DIR=${MATH_POLLER_DUMP_DIR:-scratch/errorconv} + - MATH_POLLER_RETRY_CAP=${MATH_POLLER_RETRY_CAP:-1} + networks: + - "polis-net" + extra_hosts: + - "host.docker.internal:host-gateway" + # Hard memory backstop (mirrors the delphi service). The in-memory conv cache + # never evicted by default (Clojure's 4h reboot was the de-facto cap, which we + # dropped); this limit bounds a long shadow soak. Set MATH_CONV_CACHE_CAP>0 to + # LRU-evict cold conversations before hitting it. + deploy: + resources: + limits: + memory: ${DELPHI_POLLER_CONTAINER_MEMORY:-16g} + restart: unless-stopped + profiles: + - delphi-math + postgres: restart: always build: diff --git a/example.env b/example.env index 094203c82..8ff9ab88d 100644 --- a/example.env +++ b/example.env @@ -53,6 +53,33 @@ LOCAL_SERVICES_DOCKER=true # Leave empty for autodetection on AWS deployment. See delphi/DELPHI_AUTOSCALING_SETUP.md for configuring instance size in production. INSTANCE_SIZE=dev +###### PYTHON MATH POLLER (delphi-math-poller, --profile delphi-math) ###### +# The Python replacement for the Clojure `math` container. Runs in SHADOW mode by +# default: writes math_main/math_bidtopid/math_ptptstats under a DISTINCT math_env +# so its rows stay invisible to the prod server (UNIQUE(zid, math_env)) while +# parity is validated. See delphi/docs/MATH_POLLER_DESIGN.md. +# +# math_env the poller writes under. Keep distinct from the Clojure MATH_ENV while +# shadowing; set equal to the server's MATH_ENV to cut over. Default: delphi +# DELPHI_MATH_ENV=delphi +# Watermark boot window: start polling from N days ago. Default 10 +# POLL_FROM_DAYS_AGO=10 +# Poll cadences in ms. Defaults 1000/1000 +# POLL_VOTE_INTERVAL_MS=1000 +# POLL_MOD_INTERVAL_MS=1000 +# Engine family: improved (cold recompute) or clojure-legacy (warm-start parity). +# Code default is improved; the docker-compose delphi-math-poller service overrides +# to clojure-legacy. Set explicitly for non-compose runs (e.g. the CLI directly). +# POLISMATH_ENGINE_MODE=clojure-legacy +# Conversations processed concurrently (each zid stays serialized). Default 4 +# MATH_WORKER_POOL_SIZE=4 +# errorconv dump dir + retries before parking a failing zid. Defaults scratch/errorconv, 1 +# MATH_POLLER_DUMP_DIR=scratch/errorconv +# MATH_POLLER_RETRY_CAP=1 +# Optional zid filters (comma-separated). Aliases: MATH_ZID_ALLOWLIST/MATH_ZID_BLOCKLIST +# POLL_ALLOWLIST= +# POLL_BLOCKLIST= + ###### PORTS ###### API_SERVER_PORT=5000 HTTP_PORT=80