From 6aa40888ecd9bad157170739822872f48bd0d06b Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 27 Aug 2026 05:04:21 +0900 Subject: [PATCH 1/6] perf: bound dashboard backfill priority plan --- backend/app/post_content_queue.py | 102 +++++++++--------- .../adr/0206-evidence-operations-dashboard.md | 10 ++ docs/product-technical-gap-baseline.md | 26 ++++- scripts/explain_post_content_backfill.py | 101 +++++++++++++++++ tests/test_explain_post_content_backfill.py | 20 ++++ tests/test_post_content_queue.py | 28 +++-- 6 files changed, 225 insertions(+), 62 deletions(-) create mode 100644 scripts/explain_post_content_backfill.py create mode 100644 tests/test_explain_post_content_backfill.py diff --git a/backend/app/post_content_queue.py b/backend/app/post_content_queue.py index 494884f93..903bd67a1 100644 --- a/backend/app/post_content_queue.py +++ b/backend/app/post_content_queue.py @@ -381,28 +381,21 @@ async def ensure_post_content_job( ) -async def enqueue_post_content_backfill( - pool: asyncpg.Pool, - client: redis.Redis | None, - *, - limit: int, - require_embedding: bool, - require_structure: bool, -) -> dict[str, int]: - """Durably enqueue one bounded page of eligible incomplete source posts. - - PostgreSQL is committed before Valkey is touched. A missing wake-up is - therefore recoverable by :func:`republish_queued_post_content_jobs` rather - than turning an operator request into lost work. Active and terminal jobs - are excluded so repeated requests neither duplicate work nor reset the - explicit retry boundary. - """ - if not 1 <= limit <= 200: - raise ValueError("limit must be between 1 and 200") - query = f""" +POST_CONTENT_BACKFILL_CANDIDATE_SQL = f""" select post.post_id, post.post_body from source_post post left join post_content_ingestion_job job on job.post_id = post.post_id + left join operations_case_analysis analysis + on analysis.post_id = post.post_id + and analysis.source_body_sha256 = job.source_body_sha256 + left join post_product_analysis product_analysis + on product_analysis.post_id = post.post_id + and product_analysis.source_body_sha256 = job.source_body_sha256 + left join ( + select distinct project.post_id + from post_project_mention project + where nullif(btrim(project.ontology_iri), '') is not null + ) ontology_project on ontology_project.post_id = post.post_id where {SOURCE_POST_ELIGIBILITY_SQL.format(alias='post')} and (job.post_id is null or job.status_code = $1) and ( @@ -443,53 +436,62 @@ async def enqueue_post_content_backfill( or structure.decision_source_code = 'unresolved' ) )) - or ($3::boolean and not exists ( - select 1 - from operations_case_analysis analysis - where analysis.post_id = post.post_id - and analysis.source_body_sha256 = job.source_body_sha256 - )) - or ($3::boolean and not exists ( - select 1 - from post_product_analysis analysis - where analysis.post_id = post.post_id - and analysis.source_body_sha256 = job.source_body_sha256 - )) + or ($3::boolean and analysis.post_id is null) + or ($3::boolean and product_analysis.post_id is null) ) - order by case - when $3::boolean - and exists ( - select 1 - from post_project_mention project - where project.post_id = post.post_id - and nullif(btrim(project.ontology_iri), '') is not null - ) - and job.source_body_sha256 is not null - and not exists ( - select 1 - from operations_case_analysis analysis - where analysis.post_id = post.post_id - and analysis.source_body_sha256 = job.source_body_sha256 - ) - then 0 else 1 - end, - coalesce(post.event_occurred_at, post.created_at), + and ($5::boolean = ( + $3::boolean + and ontology_project.post_id is not null + and job.source_body_sha256 is not null + and analysis.post_id is null + )) + order by coalesce(post.event_occurred_at, post.created_at), post.created_at, post.post_id limit $4 for update of post skip locked """ + + +async def enqueue_post_content_backfill( + pool: asyncpg.Pool, + client: redis.Redis | None, + *, + limit: int, + require_embedding: bool, + require_structure: bool, +) -> dict[str, int]: + """Durably enqueue one bounded page of eligible incomplete source posts. + + PostgreSQL is committed before Valkey is touched. A missing wake-up is + therefore recoverable by :func:`republish_queued_post_content_jobs` rather + than turning an operator request into lost work. Active and terminal jobs + are excluded so repeated requests neither duplicate work nor reset the + explicit retry boundary. + """ + if not 1 <= limit <= 200: + raise ValueError("limit must be between 1 and 200") requests: list[PostContentJobRequest] = [] async with pool.acquire() as conn: async with conn.transaction(): # Safe SQL: the eligibility predicate is an immutable schema fragment; values are bound. rows = await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli - query, + POST_CONTENT_BACKFILL_CANDIDATE_SQL, SUCCEEDED, require_embedding, require_structure, limit, + True, ) + if len(rows) < limit: + rows += await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli + POST_CONTENT_BACKFILL_CANDIDATE_SQL, + SUCCEEDED, + require_embedding, + require_structure, + limit - len(rows), + False, + ) for row in rows: post_id = str(row["post_id"]) body = str(row["post_body"] or "") diff --git a/docs/adr/0206-evidence-operations-dashboard.md b/docs/adr/0206-evidence-operations-dashboard.md index 8abe72879..3b8a509f4 100644 --- a/docs/adr/0206-evidence-operations-dashboard.md +++ b/docs/adr/0206-evidence-operations-dashboard.md @@ -221,6 +221,16 @@ treated as a negative case. browser output, and k6 evidence outside the repository. An empty synthetic case list remains a valid UI/API shape check; it is not evidence that grounded production cases exist. +- `scripts/explain_post_content_backfill.py` executes the exact bounded + candidate SQL with `EXPLAIN (ANALYZE, BUFFERS, WAL, FORMAT JSON)` inside a + rolled-back transaction. It reports only aggregate timing, buffer, temporary + block, node-kind, and relation-scan counts, so priority-sort, correlated + subquery, index, spill, and lock-path evidence is reproducible without + emitting source rows. +- Backfill admission reads the ontology-backed priority tier first and reads + the remaining eligible tier only when fewer than the requested bounded page + are locked. This preserves the documented total order while avoiding a + corpus-wide priority `CASE` sort and its per-row correlated probes. ## References diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 62007ddee..6c4e785d1 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -89,9 +89,29 @@ Security/operability: every aggregation applies `post_read` plus row-level corporate-entity visibility before counting; source-body digests invalidate stale inference; provider errors persist no positive/negative result; PII remains authorized at the UI boundary and is excluded from telemetry. The -tables use composite keys and bounded kind-first indexes; production hot-path -acceptance still requires `EXPLAIN (ANALYZE, BUFFERS)` on an anonymized runtime -snapshot. +tables use composite keys and bounded kind-first indexes. Production hot-path +acceptance uses `scripts/explain_post_content_backfill.py` on an anonymized +runtime snapshot; the exact candidate SQL runs in a rolled-back transaction +and emits aggregate plan/buffer metrics only. A deployment-specific +capacity/SLO remains separate from this query-shape evidence. + +On an isolated exact-schema synthetic snapshot based on #716 `c01de078` +(20,000 eligible posts and jobs, 9,927 ontology-backed project mentions, and +4,951 current operations analyses), a consecutive rolled-back comparison +returned the same 200-row priority page in 2,056.629 ms before and 1,327.868 ms +after the change. Root shared-hit blocks fell from 275,642 to 100,514; both +plans recorded zero shared reads and zero temporary reads/writes. The former +plan made 20,000 correlated project probes and 9,927 correlated +operations-analysis probes, while the semantics-equivalent two-tier query +scans each relation once. +The plan remains `Limit -> LockRows -> Sort`; `SKIP LOCKED` and the transaction +boundary therefore remain intact, and the remaining tier runs only when the +priority tier cannot fill the requested page. A separate remaining-tier +observation returned 200 rows in 12,649.654 ms with 241,317 root shared-hit +blocks and no reads or temporary spill; it is retained as the next +distribution-specific optimization target, not hidden by the priority-path +improvement. These observations establish query shape only, not a deployment +capacity or latency SLO. ### Historical UI audit evidence diff --git a/scripts/explain_post_content_backfill.py b/scripts/explain_post_content_backfill.py new file mode 100644 index 000000000..97305a8c5 --- /dev/null +++ b/scripts/explain_post_content_backfill.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +"""Measure the exact backfill candidate query without exposing source rows.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +from collections import Counter +from collections.abc import Iterator, Mapping +from typing import Any + +import asyncpg + +from backend.app.post_content_queue import POST_CONTENT_BACKFILL_CANDIDATE_SQL, SUCCEEDED + + +def _nodes(plan: Mapping[str, Any]) -> Iterator[Mapping[str, Any]]: + """Yield every PostgreSQL plan node without retaining result rows.""" + yield plan + for child in plan.get("Plans", ()): + yield from _nodes(child) + + +def summarize_plan(document: list[Mapping[str, Any]]) -> dict[str, Any]: + """Project EXPLAIN JSON into non-identifying aggregate plan evidence.""" + root = document[0] + nodes = tuple(_nodes(root["Plan"])) + node_counts = Counter(str(node["Node Type"]) for node in nodes) + relation_scans = Counter( + str(node["Relation Name"]) for node in nodes if "Relation Name" in node + ) + return { + "planning_time_ms": root.get("Planning Time"), + "execution_time_ms": root.get("Execution Time"), + "actual_rows": root["Plan"].get("Actual Rows"), + "shared_hit_blocks": int(root["Plan"].get("Shared Hit Blocks", 0)), + "shared_read_blocks": int(root["Plan"].get("Shared Read Blocks", 0)), + "temp_read_blocks": int(root["Plan"].get("Temp Read Blocks", 0)), + "temp_written_blocks": int(root["Plan"].get("Temp Written Blocks", 0)), + "node_counts": dict(sorted(node_counts.items())), + "relation_scans": dict(sorted(relation_scans.items())), + } + + +async def _measure( + dsn: str, + *, + limit: int, + embeddings: bool, + structure: bool, + priority: bool, +) -> dict[str, Any]: + """Run EXPLAIN inside a rolled-back transaction and return its summary.""" + conn = await asyncpg.connect(dsn) + transaction = conn.transaction() + await transaction.start() + try: + value = await conn.fetchval( + "EXPLAIN (ANALYZE, BUFFERS, WAL, FORMAT JSON) " + + POST_CONTENT_BACKFILL_CANDIDATE_SQL, + SUCCEEDED, + embeddings, + structure, + limit, + priority, + ) + document = json.loads(value) if isinstance(value, str) else value + return summarize_plan(document) + finally: + await transaction.rollback() + await conn.close() + + +def main() -> None: + """Parse bounded operator inputs and print aggregate JSON only.""" + parser = argparse.ArgumentParser() + parser.add_argument("--dsn", default=os.environ.get("DATABASE_URL")) + parser.add_argument("--limit", type=int, default=200, choices=range(1, 201)) + parser.add_argument("--embeddings", action="store_true") + parser.add_argument("--structure", action="store_true") + parser.add_argument("--tier", choices=("priority", "remaining"), default="priority") + args = parser.parse_args() + if not args.dsn: + parser.error("--dsn or DATABASE_URL is required") + result = asyncio.run( + _measure( + args.dsn, + limit=args.limit, + embeddings=args.embeddings, + structure=args.structure, + priority=args.tier == "priority", + ) + ) + result["candidate_tier"] = args.tier + print(json.dumps(result, indent=2, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_explain_post_content_backfill.py b/tests/test_explain_post_content_backfill.py new file mode 100644 index 000000000..2a926cf3c --- /dev/null +++ b/tests/test_explain_post_content_backfill.py @@ -0,0 +1,20 @@ +"""Tests for non-identifying backfill plan evidence.""" + +from scripts.explain_post_content_backfill import summarize_plan + + +def test_summarize_plan_reports_aggregate_buffers_and_relations_only() -> None: + """The evidence summary contains plan metrics but no source-row values.""" + result = summarize_plan([{"Planning Time": 1.25, "Execution Time": 2.5, "Plan": {"Node Type": "Limit", "Actual Rows": 12, "Shared Hit Blocks": 2, "Plans": [{"Node Type": "Index Scan", "Relation Name": "source_post", "Shared Hit Blocks": 3, "Shared Read Blocks": 1}]}}]) + + assert result == { + "planning_time_ms": 1.25, + "execution_time_ms": 2.5, + "actual_rows": 12, + "shared_hit_blocks": 2, + "shared_read_blocks": 0, + "temp_read_blocks": 0, + "temp_written_blocks": 0, + "node_counts": {"Index Scan": 1, "Limit": 1}, + "relation_scans": {"source_post": 1}, + } diff --git a/tests/test_post_content_queue.py b/tests/test_post_content_queue.py index 54a2b1b88..727eb679f 100644 --- a/tests/test_post_content_queue.py +++ b/tests/test_post_content_queue.py @@ -56,6 +56,8 @@ async def __aexit__(self, *_args: object) -> None: return None class Connection: + fetch_count = 0 + def transaction(self) -> Transaction: return Transaction() @@ -63,25 +65,31 @@ async def fetch(self, query: str, *args: object) -> list[dict[str, str]]: assert "source_draft_code" in query assert "source_deleted_flag" in query assert "job.post_id is null or job.status_code = $1" in query - assert "from operations_case_analysis analysis" in query + assert "left join operations_case_analysis analysis" in query assert "analysis.post_id = post.post_id" in query assert "analysis.source_body_sha256 = job.source_body_sha256" in query - assert "from post_product_analysis analysis" in query + assert "left join post_product_analysis product_analysis" in query assert "from post_project_mention project" in query assert "nullif(btrim(project.ontology_iri), '') is not null" in query assert "job.source_body_sha256 is not null" in query assert query.count("from post_project_mention project") == 1 - assert "when $3::boolean" in query - assert "then 0 else 1" in query + assert "$5::boolean = (" in query assert "coalesce(post.event_occurred_at, post.created_at)" in query assert "post.post_body ilike" not in query.lower() assert "post.post_title ilike" not in query.lower() assert "for update of post skip locked" in query.lower() - assert args == (SUCCEEDED, True, True, 2) - return [ - {"post_id": "00000000-0000-0000-0000-000000000001", "post_body": "one"}, - {"post_id": "00000000-0000-0000-0000-000000000002", "post_body": "two"}, - ] + self.fetch_count += 1 + assert args == ( + SUCCEEDED, + True, + True, + 2 if self.fetch_count == 1 else 1, + self.fetch_count == 1, + ) + return [{ + "post_id": f"00000000-0000-0000-0000-{self.fetch_count:012d}", + "post_body": "one" if self.fetch_count == 1 else "two", + }] class Acquire: async def __aenter__(self) -> Connection: @@ -146,6 +154,8 @@ def transaction(self) -> Transaction: return Transaction() async def fetch(self, _query: str, *_args: object) -> list[dict[str, str]]: + if _args[-1] is False: + return [] return [ {"post_id": "00000000-0000-0000-0000-000000000001", "post_body": "done"} ] From 2fb753a84252dfd5a450d057d14c1bc5254c8e76 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 27 Aug 2026 05:18:16 +0900 Subject: [PATCH 2/6] test: report actual explain scan loops --- docs/product-technical-gap-baseline.md | 5 ++++- scripts/explain_post_content_backfill.py | 7 +++++++ tests/test_explain_post_content_backfill.py | 3 ++- 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 6c4e785d1..c5d291628 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -103,7 +103,10 @@ after the change. Root shared-hit blocks fell from 275,642 to 100,514; both plans recorded zero shared reads and zero temporary reads/writes. The former plan made 20,000 correlated project probes and 9,927 correlated operations-analysis probes, while the semantics-equivalent two-tier query -scans each relation once. +removes the corpus-wide priority `CASE` and its extra correlated priority +subplans. The reproducible summary includes both relation plan-node counts and +actual scan-loop totals so a single nested-loop node cannot be mislabeled as a +single execution. The plan remains `Limit -> LockRows -> Sort`; `SKIP LOCKED` and the transaction boundary therefore remain intact, and the remaining tier runs only when the priority tier cannot fill the requested page. A separate remaining-tier diff --git a/scripts/explain_post_content_backfill.py b/scripts/explain_post_content_backfill.py index 97305a8c5..17ae548e4 100644 --- a/scripts/explain_post_content_backfill.py +++ b/scripts/explain_post_content_backfill.py @@ -31,6 +31,12 @@ def summarize_plan(document: list[Mapping[str, Any]]) -> dict[str, Any]: relation_scans = Counter( str(node["Relation Name"]) for node in nodes if "Relation Name" in node ) + relation_scan_loops = Counter() + for node in nodes: + if "Relation Name" in node: + relation_scan_loops[str(node["Relation Name"])] += int( + node.get("Actual Loops", 0) + ) return { "planning_time_ms": root.get("Planning Time"), "execution_time_ms": root.get("Execution Time"), @@ -41,6 +47,7 @@ def summarize_plan(document: list[Mapping[str, Any]]) -> dict[str, Any]: "temp_written_blocks": int(root["Plan"].get("Temp Written Blocks", 0)), "node_counts": dict(sorted(node_counts.items())), "relation_scans": dict(sorted(relation_scans.items())), + "relation_scan_loops": dict(sorted(relation_scan_loops.items())), } diff --git a/tests/test_explain_post_content_backfill.py b/tests/test_explain_post_content_backfill.py index 2a926cf3c..982a0c4ec 100644 --- a/tests/test_explain_post_content_backfill.py +++ b/tests/test_explain_post_content_backfill.py @@ -5,7 +5,7 @@ def test_summarize_plan_reports_aggregate_buffers_and_relations_only() -> None: """The evidence summary contains plan metrics but no source-row values.""" - result = summarize_plan([{"Planning Time": 1.25, "Execution Time": 2.5, "Plan": {"Node Type": "Limit", "Actual Rows": 12, "Shared Hit Blocks": 2, "Plans": [{"Node Type": "Index Scan", "Relation Name": "source_post", "Shared Hit Blocks": 3, "Shared Read Blocks": 1}]}}]) + result = summarize_plan([{"Planning Time": 1.25, "Execution Time": 2.5, "Plan": {"Node Type": "Limit", "Actual Rows": 12, "Shared Hit Blocks": 2, "Plans": [{"Node Type": "Index Scan", "Relation Name": "source_post", "Actual Loops": 4, "Shared Hit Blocks": 3, "Shared Read Blocks": 1}]}}]) assert result == { "planning_time_ms": 1.25, @@ -17,4 +17,5 @@ def test_summarize_plan_reports_aggregate_buffers_and_relations_only() -> None: "temp_written_blocks": 0, "node_counts": {"Index Scan": 1, "Limit": 1}, "relation_scans": {"source_post": 1}, + "relation_scan_loops": {"source_post": 4}, } From 3eea373d3a9025c3833be593a947d2c767f4e213 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 27 Aug 2026 05:19:21 +0900 Subject: [PATCH 3/6] fix: skip impossible priority scan --- backend/app/post_content_queue.py | 18 ++++++++++-------- tests/test_post_content_queue.py | 3 +-- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/backend/app/post_content_queue.py b/backend/app/post_content_queue.py index 903bd67a1..d17b7bdcf 100644 --- a/backend/app/post_content_queue.py +++ b/backend/app/post_content_queue.py @@ -475,14 +475,16 @@ async def enqueue_post_content_backfill( async with pool.acquire() as conn: async with conn.transaction(): # Safe SQL: the eligibility predicate is an immutable schema fragment; values are bound. - rows = await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli - POST_CONTENT_BACKFILL_CANDIDATE_SQL, - SUCCEEDED, - require_embedding, - require_structure, - limit, - True, - ) + rows = [] + if require_structure: + rows = await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli + POST_CONTENT_BACKFILL_CANDIDATE_SQL, + SUCCEEDED, + require_embedding, + require_structure, + limit, + True, + ) if len(rows) < limit: rows += await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli POST_CONTENT_BACKFILL_CANDIDATE_SQL, diff --git a/tests/test_post_content_queue.py b/tests/test_post_content_queue.py index 727eb679f..7a3b20e01 100644 --- a/tests/test_post_content_queue.py +++ b/tests/test_post_content_queue.py @@ -154,8 +154,7 @@ def transaction(self) -> Transaction: return Transaction() async def fetch(self, _query: str, *_args: object) -> list[dict[str, str]]: - if _args[-1] is False: - return [] + assert _args[-1] is False return [ {"post_id": "00000000-0000-0000-0000-000000000001", "post_body": "done"} ] From 1aed4e6389f46227f210b222a550311333c2002e Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 27 Aug 2026 05:55:22 +0900 Subject: [PATCH 4/6] fix: audit shared backfill SQL calls --- backend/app/post_content_queue.py | 3 ++- tests/test_static_sql_review_contracts.py | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/backend/app/post_content_queue.py b/backend/app/post_content_queue.py index d17b7bdcf..3ae40413d 100644 --- a/backend/app/post_content_queue.py +++ b/backend/app/post_content_queue.py @@ -474,9 +474,9 @@ async def enqueue_post_content_backfill( requests: list[PostContentJobRequest] = [] async with pool.acquire() as conn: async with conn.transaction(): - # Safe SQL: the eligibility predicate is an immutable schema fragment; values are bound. rows = [] if require_structure: + # Safe SQL: the eligibility predicate is an immutable schema fragment; values are bound. rows = await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli POST_CONTENT_BACKFILL_CANDIDATE_SQL, SUCCEEDED, @@ -486,6 +486,7 @@ async def enqueue_post_content_backfill( True, ) if len(rows) < limit: + # Safe SQL: the same immutable candidate statement is reused with bound tier values. rows += await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli POST_CONTENT_BACKFILL_CANDIDATE_SQL, SUCCEEDED, diff --git a/tests/test_static_sql_review_contracts.py b/tests/test_static_sql_review_contracts.py index f991e2324..1f17c23c3 100644 --- a/tests/test_static_sql_review_contracts.py +++ b/tests/test_static_sql_review_contracts.py @@ -29,7 +29,7 @@ ) ASYNC_STATEMENT_METHODS = {"execute", "fetch", "fetchrow", "fetchval"} SQL_REVIEW_RULE = "python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli" -EXPECTED_SQL_SUPPRESSION_COUNT = 38 +EXPECTED_SQL_SUPPRESSION_COUNT = 39 @pytest.mark.parametrize("relative_path", SQL_REVIEW_PATHS) From f35241423465396f9b9cbac7a92a7d78d2839010 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 27 Aug 2026 06:04:28 +0900 Subject: [PATCH 5/6] fix: deduplicate backfill tier transitions --- backend/app/post_content_queue.py | 9 ++++ tests/test_post_content_queue.py | 70 +++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+) diff --git a/backend/app/post_content_queue.py b/backend/app/post_content_queue.py index 3ae40413d..38e79d4b6 100644 --- a/backend/app/post_content_queue.py +++ b/backend/app/post_content_queue.py @@ -495,6 +495,15 @@ async def enqueue_post_content_backfill( limit - len(rows), False, ) + unique_rows = [] + seen_post_ids: set[str] = set() + for row in rows: + post_id = str(row["post_id"]) + if post_id in seen_post_ids: + continue + seen_post_ids.add(post_id) + unique_rows.append(row) + rows = unique_rows for row in rows: post_id = str(row["post_id"]) body = str(row["post_body"] or "") diff --git a/tests/test_post_content_queue.py b/tests/test_post_content_queue.py index 7a3b20e01..4a09ecb15 100644 --- a/tests/test_post_content_queue.py +++ b/tests/test_post_content_queue.py @@ -196,6 +196,76 @@ async def ensure( } +def test_backfill_deduplicates_a_candidate_that_changes_tier( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A row observed in both READ COMMITTED tier queries is queued only once.""" + + class Transaction: + async def __aenter__(self) -> None: + return None + + async def __aexit__(self, *_args: object) -> None: + return None + + candidate = { + "post_id": "00000000-0000-0000-0000-000000000001", + "post_body": "tier changed", + } + + class Connection: + def transaction(self) -> Transaction: + return Transaction() + + async def fetch(self, _query: str, *_args: object) -> list[dict[str, str]]: + return [candidate] + + class Acquire: + async def __aenter__(self) -> Connection: + return Connection() + + async def __aexit__(self, *_args: object) -> None: + return None + + class Pool: + def acquire(self) -> Acquire: + return Acquire() + + processed_post_ids: list[str] = [] + + async def incomplete(*_args: object, **_kwargs: object) -> bool: + return False + + async def ensure( + _conn: object, post_id: str, body: str, *, content_complete: bool + ) -> PostContentJobRequest: + processed_post_ids.append(post_id) + return PostContentJobRequest(post_id, source_body_sha256(body), QUEUED, True) + + async def publish(*_args: object, **_kwargs: object) -> str: + return "1-0" + + from backend.app import post_content_queue + + monkeypatch.setattr(post_content_queue, "post_content_is_complete", incomplete) + monkeypatch.setattr(post_content_queue, "ensure_post_content_job", ensure) + monkeypatch.setattr(post_content_queue, "publish_post_content_event", publish) + + result = asyncio.run( + enqueue_post_content_backfill( + Pool(), object(), limit=2, require_embedding=True, require_structure=True + ) + ) + + assert processed_post_ids == [candidate["post_id"]] + assert result == { + "selected_posts": 1, + "queued_posts": 1, + "published_events": 1, + "recovery_pending": 0, + } + + def test_backfill_requeues_complete_content_missing_operations_analysis( monkeypatch: pytest.MonkeyPatch, ) -> None: From 73848b2efca6db40948ee23d2f45f8b011b07848 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 27 Aug 2026 06:05:49 +0900 Subject: [PATCH 6/6] docs: record backfill tier snapshot boundary --- docs/adr/0206-evidence-operations-dashboard.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/adr/0206-evidence-operations-dashboard.md b/docs/adr/0206-evidence-operations-dashboard.md index 3b8a509f4..71ff03498 100644 --- a/docs/adr/0206-evidence-operations-dashboard.md +++ b/docs/adr/0206-evidence-operations-dashboard.md @@ -231,6 +231,8 @@ treated as a negative case. the remaining eligible tier only when fewer than the requested bounded page are locked. This preserves the documented total order while avoiding a corpus-wide priority `CASE` sort and its per-row correlated probes. + Candidate post identifiers are de-duplicated before mutation because the two + `READ COMMITTED` statements may observe a target moving between tiers. ## References