From fd717b80cab20ad7c3217c3c1f85b3838ab47793 Mon Sep 17 00:00:00 2001 From: Abhimanyu Saharan Date: Sat, 24 Oct 2026 15:09:32 +0530 Subject: [PATCH 1/7] Reduce repeated review probes in publication reports --- .../publication-report-benchmark.yml | 70 ++++ apps/admin-api/src/automation.py | 52 ++- benchmarks/publication-reports.md | 34 ++ scripts/profile_publication_reports.py | 346 ++++++++++++++++++ .../operations/test_automation_policy_unit.py | 37 +- .../test_publication_report_projection.py | 209 +++++++++++ 6 files changed, 735 insertions(+), 13 deletions(-) create mode 100644 .github/workflows/publication-report-benchmark.yml create mode 100644 benchmarks/publication-reports.md create mode 100644 scripts/profile_publication_reports.py diff --git a/.github/workflows/publication-report-benchmark.yml b/.github/workflows/publication-report-benchmark.yml new file mode 100644 index 000000000..7a05891a8 --- /dev/null +++ b/.github/workflows/publication-report-benchmark.yml @@ -0,0 +1,70 @@ +name: Publication report benchmark + +on: + pull_request: + paths: + - apps/admin-api/src/automation.py + - scripts/profile_publication_reports.py + - .github/workflows/publication-report-benchmark.yml + workflow_dispatch: + +permissions: + contents: read + +jobs: + benchmark: + name: PostgreSQL 17 / ${{ matrix.rows }} articles (ARM64) + runs-on: ubuntu-24.04-arm + timeout-minutes: 30 + strategy: + fail-fast: false + matrix: + rows: [100000, 300000] + services: + postgres: + image: postgres:17.11 + env: + POSTGRES_USER: profile + POSTGRES_PASSWORD: profile + POSTGRES_DB: publication_report_test + ports: + - 127.0.0.1:55496:5432 + options: >- + --cpus=2 --memory=3g --shm-size=512m + --health-cmd="pg_isready -U profile -d publication_report_test" + --health-interval=5s --health-timeout=5s --health-retries=10 + env: + DEVFEED_TEST_DATABASE_URL: postgresql+psycopg://profile:profile@127.0.0.1:55496/publication_report_test + DEVFEED_DATABASE_URL: postgresql+psycopg://profile:profile@127.0.0.1:55496/publication_report_test + DEVFEED_REDIS_URL: redis://127.0.0.1:6379/15 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + - uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 + with: + version: "0.13.5" + enable-cache: true + - name: Install locked workspace dependencies + run: sh scripts/ci/python-bootstrap.sh . --editable --all-packages + - name: Apply disposable database migrations + run: .venv/bin/alembic upgrade head + - name: Compare fixed publication cohorts under steady and concurrent ingestion + env: + ARTICLE_ROWS: ${{ matrix.rows }} + run: | + mkdir -p reports/publication + git rev-parse HEAD > reports/publication/source-sha.txt + .venv/bin/python scripts/profile_publication_reports.py --seed \ + --rows "$ARTICLE_ROWS" --repeats 7 --inline-comparison \ + --output "reports/publication/${ARTICLE_ROWS}-arm64.json" + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + if: always() + with: + name: publication-report-${{ matrix.rows }}-arm64-${{ github.run_attempt }} + path: reports/publication/ + if-no-files-found: error + retention-days: 7 diff --git a/apps/admin-api/src/automation.py b/apps/admin-api/src/automation.py index ef4dc25f7..1f1acb84a 100644 --- a/apps/admin-api/src/automation.py +++ b/apps/admin-api/src/automation.py @@ -606,13 +606,22 @@ def automation_blocker_rows(session, article_ids): def _publication_automation_query(start, now): - cutoff = ( - ArticleReview.article_id == Article.id, - ArticleReview.created_at <= Article.published_to_feed_at, + publications = ( + select(Article.id, Article.published_to_feed_at, Article.discovered_at) + .where(Article.published_to_feed_at >= start, Article.published_to_feed_at <= now) + .cte("publications") + .prefix_with("MATERIALIZED") ) - # Fixed predicates match the narrow partial indexes even for prepared generic - # plans. Neither evidence JSON nor complete review histories are transferred. + # EXISTS in a WHERE clause can become a semi-join: unlike probes inside the + # aggregate FILTER, PostgreSQL can scan narrow evidence indexes once for a + # broad report, or stop at the first match for a selective cohort. Semi/anti + # joins retain one row per publication even with repeated review facts. + # Fixed predicates match the partial indexes for prepared generic plans too. empty = literal_column("'{}'::jsonb", type_=JSONB) + cutoff = ( + ArticleReview.article_id == publications.c.id, + ArticleReview.created_at <= publications.c.published_to_feed_at, + ) automatic = ( select(ArticleReview.article_id) .where( @@ -625,13 +634,32 @@ def _publication_automation_query(start, now): manual = ( select(ArticleReview.article_id).where(*cutoff, ArticleReview.automation == empty).exists() ) - return select( - func.count(), - func.count().filter(automatic, ~manual), - func.percentile_cont(0.5) - .within_group(func.extract("epoch", Article.published_to_feed_at - Article.discovered_at)) - .filter(Article.published_to_feed_at >= Article.discovered_at), - ).where(Article.published_to_feed_at >= start, Article.published_to_feed_at <= now) + totals = ( + select( + func.count().label("published"), + func.percentile_cont(0.5) + .within_group( + func.extract( + "epoch", publications.c.published_to_feed_at - publications.c.discovered_at + ) + ) + .filter(publications.c.published_to_feed_at >= publications.c.discovered_at) + .label("median"), + ) + .select_from(publications) + .subquery("publication_totals") + ) + autonomous = ( + select(func.count().label("automatic")) + .select_from(publications) + .where(automatic, ~manual) + .subquery("autonomous_publications") + ) + return ( + select(totals.c.published, autonomous.c.automatic, totals.c.median) + .select_from(totals) + .join(autonomous, true()) + ) def publication_automation(session, start, now): diff --git a/benchmarks/publication-reports.md b/benchmarks/publication-reports.md new file mode 100644 index 000000000..1d8d951e6 --- /dev/null +++ b/benchmarks/publication-reports.md @@ -0,0 +1,34 @@ +# Publication report scaling + +Issue [#296](https://github.com/abhi1693/devfeed.tech/issues/296) concerns repeated review-index probes for each publication. The existing two partial review indexes already support the original predicates. The change separates publication totals and duration calculations from autonomous classification. The classification count uses existence predicates in its WHERE clause, allowing PostgreSQL to choose semi/anti joins, scan narrow review indexes once for broad windows or stop at the first match for selective windows. Repeated review facts cannot multiply either count. The publication cohort contains only IDs and timestamps. No historical evidence JSON or analysis job payload enters the report. + +Manual evidence still takes precedence, regardless of the review action. Automatic evidence still requires a publish action and nonempty automation. Both kinds of evidence use each article's publication time as their inclusive cutoff; evidence may predate the reporting period. Counts include each article once, including articles with no matching evidence. The median retains its existing nonnegative-duration filter. + +## Reproduce + +Use a dedicated PostgreSQL 17 instance with a database ending in `_test`. The profiler refuses remote database hosts, connection query overrides and inherited libpq host/service overrides. `--seed` truncates articles and their dependent rows. Apply the repository's migrations to the disposable instance first, then run: + +```sh +export DEVFEED_TEST_DATABASE_URL=postgresql+psycopg://profile:profile@127.0.0.1:55496/publication_report_test +export DEVFEED_DATABASE_URL="$DEVFEED_TEST_DATABASE_URL" +export DEVFEED_REDIS_URL=redis://127.0.0.1:6379/15 +uv run alembic upgrade head +uv run python scripts/profile_publication_reports.py --seed --rows 100000 --repeats 7 \ + --output reports/publication/100000.json +uv run python scripts/profile_publication_reports.py --seed --rows 300000 --repeats 7 \ + --output reports/publication/300000.json +``` + +The `Publication report benchmark` workflow runs these populations on native `ubuntu-24.04-arm`, with PostgreSQL 17.11 limited to two CPUs and 3 GiB memory. It runs for relevant pull requests and accepts manual dispatch. Download the `publication-report-*-arm64-*` artifacts within seven days. Each artifact includes the exact SQL, source SHA, PostgreSQL architecture/version and settings, all timing/CPU samples and complete `EXPLAIN (ANALYZE, BUFFERS, WAL, SETTINGS, FORMAT JSON)` plans. Source and report data remain separate: generated reports belong in the ignored `reports/` directory. + +## Workload and interpretation + +The 24 October 2026 read-only production audit estimated 92,503 articles, 255,529 reviews and 156,002 analysis jobs. The fixed synthetic fixtures contain 100,000 and 300,000 articles, 3.12 reviews per article and 1.65 historical analysis jobs per article. Reviews include manual-only, automatic-only, overlapping facts, reviews after publication, reviews older than the window, duplicate evidence and a tail of 27 reviews on one article in 200. Publications span 60 days, so the 30-day cohort contains approximately half of the population. Article summaries, review notes and job JSON retain payload outside the narrow report columns. All normal database triggers and indexes remain enabled; fixture inserts use batches of at most 5,000 article identities per statement (review fanout produces more review rows). Fixture setup ends with vacuum/analyze. + +Each population exercises inclusive 1-, 7- and 30-day periods, both with and without a concurrent ingestion thread. The writer commits new unpublished articles, reviews and queued analysis jobs, with a five millisecond pause between commits. Those articles are outside the fixed report cohort, allowing exact result equality checks while ordinary writes, indexes and triggers compete for resources. Report queries use 4 MiB `work_mem`, disable parallel workers for attributable backend CPU measurements, and retain the server's JIT setting. Each strategy receives one unmeasured warmup, followed by measured samples and an instrumented plan. The baseline reproduces the correlated-existence SQL from commit `a483edec50fd331fea22c665e521e8a849cdf090`; the candidate compiles the actual application helper. + +Compare root-plan shared-buffer hits/reads, temp blocks, WAL and review-node loop counts as well as elapsed time. Buffer hits are logical page accesses. PostgreSQL backend CPU comes from user/system ticks in that backend's `/proc/self/stat`, sampled through the disposable superuser connection; resolution is the host's clock tick. Application CPU is Python's calling-thread CPU around statement execution and result consumption, excluding the concurrent writer thread. Neither is an estimate of complete API request CPU. Writer throughput, commit latency and generated WAL are recorded separately; WAL includes ordinary checkpoint/full-page effects in the isolated service. Timing and writer samples are descriptive, without a flaky elapsed-time CI gate. Exact result equality is mandatory for every comparison. At 100,000 articles and above, each 30-day comparison additionally requires candidate buffer accesses below 75% of the baseline, no correlated SubPlan nodes, and total temporary read/write bytes below 512 bytes per publication. These structural budgets check repeated work and bounded intermediate data while allowing normal planner choices. `--inline-comparison` additionally profiles an inlined publication CTE when evaluating planner alternatives. + +## Rollout and rollback + +This change adds no indexes, triggers, stored facts or migration, so write paths and recovery semantics remain unchanged. Rolling back the application restores the original SQL without a database downgrade. After deployment, compare fresh statement-statistics deltas under matched report periods and ingestion workload; confirm report freshness, review precedence and duration metrics. Check PostgreSQL CPU, buffer work, temporary writes, ingestion latency and queue age independently. Local x86 runs validate the SQL and profiling mechanism; native ARM64 artifacts supply architecture-specific execution evidence. Future production diagnostics must remain bounded and read-only; this benchmark does not execute the changed query against production. Do not run these fixture or `ANALYZE` operations against production. diff --git a/scripts/profile_publication_reports.py b/scripts/profile_publication_reports.py new file mode 100644 index 000000000..f8d981c0a --- /dev/null +++ b/scripts/profile_publication_reports.py @@ -0,0 +1,346 @@ +"""Compare publication-report SQL on a disposable local PostgreSQL 17 database. + +DEVFEED_TEST_DATABASE_URL must name a local database ending in _test. --seed +truncates article fixtures (including dependent rows). Apply migrations first. +No production rows or credentials are used. Full JSON plans are retained. +""" + +import argparse +import json +import os +import platform +import statistics +import threading +import time +import uuid +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from devfeed_admin_api.automation import _publication_automation_query +from sqlalchemy import create_engine, text +from sqlalchemy.engine import make_url + +NOW = datetime(2026, 10, 24, 12, tzinfo=UTC) +BASELINE = """SELECT count(*), count(*) FILTER ( + WHERE EXISTS (SELECT 1 FROM article_reviews r WHERE r.article_id=a.id + AND r.created_at<=a.published_to_feed_at AND r.action='publish' AND r.automation<>'{}'::jsonb) + AND NOT EXISTS (SELECT 1 FROM article_reviews r WHERE r.article_id=a.id + AND r.created_at<=a.published_to_feed_at AND r.automation='{}'::jsonb)), + percentile_cont(0.5) WITHIN GROUP (ORDER BY extract(epoch FROM + a.published_to_feed_at-a.discovered_at)) FILTER ( + WHERE a.published_to_feed_at>=a.discovered_at) + FROM articles a WHERE a.published_to_feed_at>=:start AND a.published_to_feed_at<=:now""" +ARTICLE_INSERT = """INSERT INTO articles + (id,canonical_url,url_hash,title,slug,summary,content_type,feed_at,discovered_at, + publication_status,review_status,published_to_feed_at) + SELECT md5(:namespace||n)::uuid,'https://profile.invalid/'||:namespace||n, + md5(:namespace||n), 'Publication fixture '||n, :namespace||n, repeat('fixture ',128), + 'article',:now, :now-(n%60)*interval '1 day'-(n%86400)*interval '1 second' + -(CASE WHEN n%37=0 THEN -60 ELSE 3600+n%100000 END)*interval '1 second', + 'published','approved',:now-(n%60)*interval '1 day'-(n%86400)*interval '1 second' + FROM generate_series(CAST(:first AS integer),CAST(:last AS integer)) n""" + + +def seed_batches(connection, sql, rows, params): + for first in range(1, rows + 1, 5000): + connection.execute(text(sql), {**params, "first": first, "last": min(rows, first + 4999)}) + + +def validate_target(value): + url = make_url(value) + if ( + url.get_backend_name() != "postgresql" + or url.host not in {"127.0.0.1", "localhost"} + or not (url.database or "").endswith("_test") + or url.query + or any(os.environ.get(key) for key in ("PGHOSTADDR", "PGSERVICE", "PGSERVICEFILE")) + ): + raise ValueError( + "Use a disposable local PostgreSQL database ending in _test without overrides" + ) + return url + + +def seed(engine, rows): + with engine.begin() as connection: + connection.execute(text("TRUNCATE articles CASCADE")) + seed_batches(connection, ARTICLE_INSERT, rows, {"now": NOW, "namespace": "publication-"}) + seed_batches( + connection, + """INSERT INTO article_reviews + (id,article_id,action,revision,automation,created_at,note) + SELECT md5('publication-review-'||n||'-'||f)::uuid, + md5('publication-'||n)::uuid, + CASE WHEN f=1 THEN 'publish' ELSE 'approve' END, f, + CASE WHEN (f=2 AND n%10 IN (1,3,6)) OR (f=3 AND n%10=4) + THEN '{}'::jsonb ELSE jsonb_build_object('profile',true) END, + a.published_to_feed_at + CASE + WHEN (f=1 AND n%10 IN (0,1,5,6)) OR (f=3 AND n%10=4) + THEN interval '1 hour' + WHEN f=2 THEN interval '-90 days' ELSE interval '-1 second' END, + repeat('review evidence ',32) + FROM generate_series(CAST(:first AS integer),CAST(:last AS integer)) n + JOIN articles a ON a.id=md5('publication-'||n)::uuid + CROSS JOIN LATERAL generate_series(1,CASE WHEN n%200=0 THEN 27 ELSE 3 END) f""", + rows, + {"rows": rows}, + ) + seed_batches( + connection, + """INSERT INTO article_analysis_jobs + (id,article_id,status,attempts,available_at,created_at,finished_at, + input_snapshot,catalog_snapshot,result) + SELECT md5('publication-job-'||n)::uuid, + md5('publication-'||(1+n%:rows))::uuid,'succeeded',1, + :now-(n%180)*interval '1 day',:now-(n%180)*interval '1 day',:now, + jsonb_build_object('text',repeat('historical payload ',32)), + '{}','{}' FROM generate_series(CAST(:first AS integer),CAST(:last AS integer)) n""", + rows * 165 // 100, + {"now": NOW, "rows": rows}, + ) + with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as connection: + for table in ("articles", "article_reviews", "article_analysis_jobs"): + connection.execute(text(f"VACUUM (ANALYZE) {table}")) + + +def backend_cpu(connection): + # /proc/self belongs to this backend, including on a Docker service. Field 14 + # and 15 are user/system ticks; the two tiny sampling queries add fixed cost. + value = connection.scalar(text("SELECT pg_read_file('/proc/self/stat')")) + fields = value[value.rindex(")") + 2 :].split() + return (int(fields[11]) + int(fields[12])) / os.sysconf("SC_CLK_TCK") + + +class Ingestion: + def __init__(self, engine): + self.engine = engine + self.stop = threading.Event() + self.ready = threading.Event() + self.rows = 0 + self.errors = [] + self.samples_ms = [] + self.wal_bytes = 0 + self.thread = threading.Thread(target=self.run, daemon=True) + + def run(self): + try: + with self.engine.connect() as connection: + first_wal = connection.scalar(text("SELECT pg_current_wal_insert_lsn()")) + connection.commit() + self.ready.set() + while not self.stop.is_set(): + started = time.perf_counter() + identifier = uuid.uuid4() + with connection.begin(): + connection.execute( + text("""INSERT INTO articles + (id,canonical_url,url_hash,title,slug,summary,content_type, + feed_at,discovered_at) VALUES + (:id,:url,:hash,'Concurrent fixture',:slug,'','article',:now,:now)"""), + { + "id": identifier, + "url": f"https://profile.invalid/{identifier}", + "hash": identifier.hex, + "slug": f"concurrent-{identifier}", + "now": NOW + timedelta(days=1), + }, + ) + connection.execute( + text("""INSERT INTO article_reviews + (id,article_id,action,revision,automation,created_at) + VALUES (:id,:article,'publish',1, + jsonb_build_object('profile',true),:now)"""), + {"id": uuid.uuid4(), "article": identifier, "now": NOW}, + ) + connection.execute( + text("""INSERT INTO article_analysis_jobs + (id,article_id,status,attempts,available_at,created_at, + input_snapshot,catalog_snapshot,result) + VALUES (:id,:article,'queued',0,:now,:now,'{}','{}','{}')"""), + {"id": uuid.uuid4(), "article": identifier, "now": NOW}, + ) + self.rows += 1 + self.samples_ms.append((time.perf_counter() - started) * 1000) + self.stop.wait(0.005) + self.wal_bytes = int( + connection.scalar( + text("SELECT pg_wal_lsn_diff(pg_current_wal_insert_lsn(),:first)"), + {"first": first_wal}, + ) + ) + except Exception as error: + self.errors.append(str(error)) + self.ready.set() + + +def measure(engine, sql, params, repeats, concurrent): + ingestion = Ingestion(engine) if concurrent else None + if ingestion: + ingestion.thread.start() + if not ingestion.ready.wait(10): + raise RuntimeError("Concurrent ingestion did not start") + samples, cpu, app_cpu = [], [], [] + started = time.perf_counter() + try: + with engine.connect() as connection: + connection.execute(text("SET statement_timeout='60s'")) + connection.execute(text("SET work_mem='4MB'")) + connection.execute(text("SET max_parallel_workers_per_gather=0")) + connection.execute(text(sql), params).one() # One unmeasured warmup per strategy. + for _ in range(repeats): + before_cpu = backend_cpu(connection) + before_app = time.thread_time() + before = time.perf_counter() + result = list(connection.execute(text(sql), params).one()) + samples.append((time.perf_counter() - before) * 1000) + app_cpu.append((time.thread_time() - before_app) * 1000) + cpu.append((backend_cpu(connection) - before_cpu) * 1000) + plan = connection.scalar( + text("EXPLAIN (ANALYZE, BUFFERS, WAL, SETTINGS, FORMAT JSON) " + sql), params + )[0] + finally: + elapsed = time.perf_counter() - started + if ingestion: + ingestion.stop.set() + ingestion.thread.join(10) + if ingestion.thread.is_alive() or ingestion.errors: + raise RuntimeError(f"Concurrent ingestion failed: {ingestion.errors}") + output = { + "result": result, + "samples_ms": samples, + "median_ms": statistics.median(samples), + "backend_cpu_samples_ms": cpu, + "application_thread_cpu_samples_ms": app_cpu, + "plan": plan, + } + if ingestion: + output["concurrent_ingestion"] = { + "commits": ingestion.rows, + "commits_per_second": ingestion.rows / elapsed, + "median_commit_ms": statistics.median(ingestion.samples_ms), + "max_commit_ms": max(ingestion.samples_ms), + "wal_bytes": ingestion.wal_bytes, + "wal_bytes_per_commit": ingestion.wal_bytes / ingestion.rows, + "errors": ingestion.errors, + } + return output + + +def plan_nodes(node): + yield node + for child in node.get("Plans", []): + yield from plan_nodes(child) + + +def check_work(report): + baseline = report["strategies"]["baseline"]["plan"]["Plan"] + candidate = report["strategies"]["candidate"]["plan"]["Plan"] + + def touches(node): + return node.get("Shared Hit Blocks", 0) + node.get("Shared Read Blocks", 0) + + assert touches(candidate) < touches(baseline) * 0.75, "Broad report buffer work regressed" + assert not any( + node.get("Parent Relationship") == "SubPlan" for node in plan_nodes(candidate) + ), "Candidate retained a per-publication correlated SubPlan" + publications = report["strategies"]["candidate"]["result"][0] + temp_bytes = 8192 * ( + candidate.get("Temp Read Blocks", 0) + candidate.get("Temp Written Blocks", 0) + ) + assert temp_bytes <= publications * 512, ( + "Intermediate temporary work exceeded the cohort budget" + ) + return { + "candidate_buffer_fraction": touches(candidate) / touches(baseline), + "no_correlated_subplan": True, + "temporary_bytes": temp_bytes, + "temporary_budget_bytes": publications * 512, + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--seed", action="store_true") + parser.add_argument("--rows", type=int, default=100000) + parser.add_argument("--repeats", type=int, default=5) + parser.add_argument("--inline-comparison", action="store_true") + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + try: + url = validate_target(os.environ["DEVFEED_TEST_DATABASE_URL"]) + if not 100 <= args.rows <= 1000000 or not 1 <= args.repeats <= 20: + raise ValueError("Use 100..1000000 rows and 1..20 repetitions") + except (ValueError, KeyError) as error: + parser.error(str(error)) + engine = create_engine(url, pool_size=3) + try: + with engine.connect() as connection: + version = connection.scalar(text("SELECT version()")) + if connection.dialect.server_version_info[:1] != (17,): + raise ValueError("Benchmark requires PostgreSQL 17") + settings = dict( + connection.execute( + text( + "SELECT name,setting FROM pg_settings WHERE name IN " + "('shared_buffers','work_mem','max_parallel_workers_per_gather'," + "'track_io_timing','jit','fsync','synchronous_commit')" + ) + ).all() + ) + if args.seed: + seed(engine, args.rows) + output = { + "rows": args.rows, + "fixture_review_rows": args.rows * 3 + args.rows // 200 * 24, + "fixture_analysis_job_rows": args.rows * 165 // 100, + "client_architecture": platform.machine(), + "postgresql": version, + "settings": settings, + "repeats": args.repeats, + "reports": [], + } + args.output.parent.mkdir(parents=True, exist_ok=True) + for concurrent in (False, True): + for days in (1, 7, 30): + params = {"start": NOW - timedelta(days=days), "now": NOW} + statement = _publication_automation_query(params["start"], NOW) + candidate = str(statement.compile(engine, compile_kwargs={"literal_binds": True})) + strategies = {"baseline": BASELINE, "candidate": candidate} + if args.inline_comparison: + strategies["candidate_inline"] = candidate.replace( + "AS MATERIALIZED", "AS NOT MATERIALIZED" + ) + report = {"days": days, "concurrent": concurrent, "strategies": {}} + for name, sql in strategies.items(): + report["strategies"][name] = measure( + engine, sql, params, args.repeats, concurrent + ) + report["strategies"][name]["sql"] = sql + results = [value["result"] for value in report["strategies"].values()] + output["reports"].append(report) + # Preserve failing plans and samples too; CI uploads partial output. + args.output.write_text(json.dumps(output, indent=2) + "\n") + assert all(result == results[0] for result in results), "Report results differ" + report["result_equality"] = True + if args.rows >= 100000 and days == 30: + report["work_checks"] = check_work(report) + args.output.write_text(json.dumps(output, indent=2) + "\n") + print( + json.dumps( + { + "days": days, + "concurrent": concurrent, + "strategies": { + name: round(value["median_ms"], 2) + for name, value in report["strategies"].items() + }, + } + ), + flush=True, + ) + finally: + engine.dispose() + + +if __name__ == "__main__": + main() diff --git a/tests/admin/operations/test_automation_policy_unit.py b/tests/admin/operations/test_automation_policy_unit.py index 644f181e4..fe5eca8b3 100644 --- a/tests/admin/operations/test_automation_policy_unit.py +++ b/tests/admin/operations/test_automation_policy_unit.py @@ -257,7 +257,8 @@ def execute(statement): @pytest.mark.parametrize( - "published,automatic,median,percent", [(0, 0, None, None), (3, 2, 120.0, 66.7)] + "published,automatic,median,percent", + [(0, 0, None, None), (4, 0, None, 0.0), (3, 2, 120.0, 66.7), (6, 1, 0.0, 16.7)], ) def test_publication_metrics_distinguish_empty_history_from_zero_automatic_publication( published, automatic, median, percent @@ -273,6 +274,40 @@ def test_publication_metrics_distinguish_empty_history_from_zero_automatic_publi } +def test_publication_query_keeps_narrow_review_probes_out_of_aggregate_filters(): + from sqlalchemy.dialects import postgresql + from sqlalchemy.sql import visitors + from sqlalchemy.sql.elements import FunctionFilter + from sqlalchemy.sql.selectable import Exists, Select + + start = NOW - timedelta(days=7) + statement = automation._publication_automation_query(start, NOW) + compiled = statement.compile(dialect=postgresql.dialect()) + sql = str(compiled) + # WHERE EXISTS can become semi/anti joins; EXISTS inside aggregate FILTER + # forces repeated per-publication subplans and recreates buffer amplification. + aggregate_filters = [ + clause for clause in visitors.iterate(statement) if isinstance(clause, FunctionFilter) + ] + assert aggregate_filters + assert all( + not any(isinstance(clause, Exists) for clause in visitors.iterate(aggregate.criterion)) + for aggregate in aggregate_filters + ) + assert "article_reviews.action = 'publish'" in sql + assert "article_reviews.automation != '{}'::jsonb" in sql + assert "article_reviews.automation = '{}'::jsonb" in sql + assert start in compiled.params.values() and NOW in compiled.params.values() + review_projections = [ + column.key + for select in visitors.iterate(statement) + if isinstance(select, Select) + for column in select.selected_columns + if getattr(getattr(column, "table", None), "name", None) == "article_reviews" + ] + assert review_projections and set(review_projections) == {"article_id"} + + @pytest.mark.parametrize("projected", [False, True]) def test_automation_overview_composes_one_consistent_window_and_selects_correct_blocker_reader( monkeypatch, projected diff --git a/tests/admin/reporting/projections/test_publication_report_projection.py b/tests/admin/reporting/projections/test_publication_report_projection.py index 3b94b2e36..cbef1b380 100644 --- a/tests/admin/reporting/projections/test_publication_report_projection.py +++ b/tests/admin/reporting/projections/test_publication_report_projection.py @@ -8,10 +8,219 @@ from sqlalchemy import text from tests.admin.reporting.projections.test_pipeline_projection_safety import add_article +from tests.support.source_readiness import NOW pytestmark = pytest.mark.integration +def test_review_facts_count_distinct_articles_and_do_not_weight_publication_duration(database): + start = NOW - timedelta(days=1) + automatic = {"policy": "automatic"} + cases = [ + ([], 60), + ([("approve", {})], 120), + ([("publish", automatic)], 180), + ([("publish", automatic), ("approve", {})], 240), + ([("publish", automatic)] * 4, 300), + ([("approve", automatic), ("reject", automatic)], 360), + ([("publish", automatic), ("reject", {})], 420), + ([("publish", automatic), ("unpublish", {})], 480), + ] + with database.begin() as session: + for index, (reviews, delay) in enumerate(cases): + # Different publication cutoffs in one report prevent treating the + # report's end as every article's review-intervention cutoff. + published = start + timedelta(hours=index + 1) + article = add_article(session, published=True, published_at=published) + article.discovered_at = published - timedelta(seconds=delay) + session.add_all( + ArticleReview( + article_id=article.id, + action=action, + revision=revision, + automation=evidence, + created_at=published, + ) + for revision, (action, evidence) in enumerate(reviews, start=1) + ) + session.flush() + assert publication_automation(session, start, NOW) == { + "published_in_window": 8, + "published_without_intervention": 2, + "automatic_publication_percent": 25.0, + "median_ingestion_to_publication_seconds": 270.0, + } + + +def test_review_cutoffs_are_inclusive_and_specific_to_each_publication(database): + start = NOW - timedelta(days=1) + automatic = {"policy": "automatic"} + with database.begin() as session: + for index, reviews in enumerate( + [ + [("publish", automatic, 0)], + [("publish", automatic, 1)], + [("publish", automatic, -1), ("approve", {}, 0)], + [("publish", automatic, -1), ("approve", {}, 1)], + [("publish", automatic, -1), ("approve", {}, -1)], + ] + ): + published = start + timedelta(hours=index + 1) + article = add_article(session, published=True, published_at=published) + article.discovered_at = published - timedelta(seconds=60) + session.add_all( + ArticleReview( + article_id=article.id, + action=action, + revision=revision, + automation=evidence, + created_at=published + timedelta(microseconds=offset), + ) + for revision, (action, evidence, offset) in enumerate(reviews, start=1) + ) + session.flush() + assert publication_automation(session, start, NOW) == { + "published_in_window": 5, + "published_without_intervention": 2, + "automatic_publication_percent": 40.0, + "median_ingestion_to_publication_seconds": 60.0, + } + + +def test_report_keeps_old_review_facts_and_excludes_unrelated_publications(database): + start = NOW - timedelta(days=1) + old = start - timedelta(days=90) + automatic = {"policy": "automatic"} + with database.begin() as session: + for published, reviews in [ + (NOW, [("publish", automatic, old)]), + # An old manual intervention still vetoes a later automatic publish; + # repeated old manual facts must not duplicate the publication row. + ( + NOW, + [("approve", {}, old), ("reject", {}, old), ("publish", automatic, NOW)], + ), + (NOW, []), + (start - timedelta(microseconds=1), [("publish", automatic, old)]), + (NOW + timedelta(microseconds=1), [("approve", {}, old)]), + (None, [("publish", automatic, old)]), + ]: + article = add_article(session, published=True, published_at=published) + article.discovered_at = (published or NOW) - timedelta(seconds=120) + session.add_all( + ArticleReview( + article_id=article.id, + action=action, + revision=revision, + automation=evidence, + created_at=created_at, + ) + for revision, (action, evidence, created_at) in enumerate(reviews, start=1) + ) + session.flush() + assert publication_automation(session, start, NOW) == { + "published_in_window": 3, + "published_without_intervention": 1, + "automatic_publication_percent": 33.3, + "median_ingestion_to_publication_seconds": 120.0, + } + + +def test_publication_boundaries_keep_invalid_durations_out_of_median_without_losing_counts( + database, +): + start = NOW - timedelta(days=1) + with database.begin() as session: + for published, delay in [ + (None, 0), + (start - timedelta(microseconds=1), 900), + (start, 0), + (NOW, 120), + (NOW, 480), + (NOW, -60), + (NOW + timedelta(microseconds=1), 900), + ]: + article = add_article(session, published=True, published_at=published) + article.discovered_at = (published or NOW) - timedelta(seconds=delay) + session.flush() + assert publication_automation(session, start, NOW) == { + "published_in_window": 4, + "published_without_intervention": 0, + "automatic_publication_percent": 0.0, + "median_ingestion_to_publication_seconds": 120.0, + } + empty = { + "published_in_window": 0, + "published_without_intervention": 0, + "automatic_publication_percent": None, + "median_ingestion_to_publication_seconds": None, + } + assert publication_automation(session, NOW + timedelta(days=1), NOW) == empty + assert ( + publication_automation(session, NOW + timedelta(days=1), NOW + timedelta(days=2)) + == empty + ) + + +def test_publication_with_only_negative_durations_keeps_null_median(database): + with database.begin() as session: + article = add_article(session, published=True, published_at=NOW) + article.discovered_at = NOW + timedelta(seconds=1) + session.add( + ArticleReview( + article_id=article.id, + action="publish", + revision=1, + automation={"policy": "automatic"}, + created_at=NOW, + ) + ) + session.flush() + assert publication_automation(session, NOW, NOW) == { + "published_in_window": 1, + "published_without_intervention": 1, + "automatic_publication_percent": 100.0, + "median_ingestion_to_publication_seconds": None, + } + + +def test_report_reloads_review_facts_and_publication_cutoff_after_editorial_changes(database): + with database.begin() as session: + published = NOW - timedelta(hours=1) + article = add_article(session, published=True, published_at=published) + article.discovered_at = published - timedelta(minutes=10) + + def report(): + session.flush() + return publication_automation(session, NOW - timedelta(days=1), NOW) + + assert report()["published_without_intervention"] == 0 + session.add( + ArticleReview( + article_id=article.id, + action="publish", + revision=1, + automation={"policy": "automatic"}, + created_at=published, + ) + ) + assert report()["published_without_intervention"] == 1 + manual = ArticleReview( + article_id=article.id, + action="approve", + revision=2, + automation={}, + created_at=published + timedelta(microseconds=1), + ) + session.add(manual) + assert report()["published_without_intervention"] == 1 + article.published_to_feed_at = manual.created_at + changed = report() + assert changed["published_in_window"] == 1 + assert changed["published_without_intervention"] == 0 + assert changed["median_ingestion_to_publication_seconds"] == pytest.approx(600.000001) + + def test_large_review_evidence_uses_partial_indexes_without_changing_manual_cutoff(database): now = utcnow() evidence = {"policy": "automatic", "evidence": "Unicode extraction evidence 設計 " * 3000} From 04d3e84adc6e87ca248fa9fc7acc6c5e4514572b Mon Sep 17 00:00:00 2001 From: Abhimanyu Saharan Date: Sat, 24 Oct 2026 15:11:11 +0530 Subject: [PATCH 2/7] Bound scale fixtures while preserving production cardinalities --- scripts/profile_publication_reports.py | 3 ++- .../test_recurring_query_budgets.py | 22 +++++++++++++++---- 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/scripts/profile_publication_reports.py b/scripts/profile_publication_reports.py index f8d981c0a..63c40df05 100644 --- a/scripts/profile_publication_reports.py +++ b/scripts/profile_publication_reports.py @@ -36,7 +36,8 @@ SELECT md5(:namespace||n)::uuid,'https://profile.invalid/'||:namespace||n, md5(:namespace||n), 'Publication fixture '||n, :namespace||n, repeat('fixture ',128), 'article',:now, :now-(n%60)*interval '1 day'-(n%86400)*interval '1 second' - -(CASE WHEN n%37=0 THEN -60 ELSE 3600+n%100000 END)*interval '1 second', + -(CASE WHEN n%37=0 THEN -60 WHEN n%10 IN (1,3,6) THEN 91*86400 + ELSE 3600+n%100000 END)*interval '1 second', 'published','approved',:now-(n%60)*interval '1 day'-(n%86400)*interval '1 second' FROM generate_series(CAST(:first AS integer),CAST(:last AS integer)) n""" diff --git a/tests/infrastructure/test_recurring_query_budgets.py b/tests/infrastructure/test_recurring_query_budgets.py index 30ff2b72c..70d5d685f 100644 --- a/tests/infrastructure/test_recurring_query_budgets.py +++ b/tests/infrastructure/test_recurring_query_budgets.py @@ -294,19 +294,33 @@ def recurring_query_data(database): """INSERT INTO article_origins(id,article_id,source_id,entry_key,original_url, source_metadata) SELECT gen_random_uuid(),md5('a'||i)::uuid,md5('s'||(1+i%1261))::uuid,i::text, - 'https://article.test/'||i,'{}'::jsonb FROM generate_series(1,60340) i""", + 'https://article.test/'||i,'{}'::jsonb FROM generate_series( + CAST(:batch_start AS integer),CAST(:batch_end AS integer)) i""", """INSERT INTO article_tags(article_id,tag_id) SELECT md5('a'||i)::uuid,md5('t'||(1+(i*10+j)%34074))::uuid - FROM generate_series(1,60340) i CROSS JOIN generate_series(1,10) j""", + FROM generate_series( + CAST(:batch_start AS integer),CAST(:batch_end AS integer)) i + CROSS JOIN generate_series(1,10) j""", """INSERT INTO article_reviews(id,article_id,action,revision,automation,created_at) SELECT gen_random_uuid(),a.id,CASE WHEN j=1 THEN 'publish' ELSE 'approve' END,j, CASE WHEN i%31=0 AND j=2 THEN '{}'::jsonb ELSE jsonb_build_object('policy','synthetic','context',repeat(md5(i::text),9)) END,coalesce(a.published_to_feed_at,a.discovered_at)-interval '1 minute' - FROM generate_series(1,60340) i JOIN articles a ON a.id=md5('a'||i)::uuid + FROM generate_series( + CAST(:batch_start AS integer),CAST(:batch_end AS integer)) i + JOIN articles a ON a.id=md5('a'||i)::uuid CROSS JOIN generate_series(1,3) j WHERE j<=2 OR i<=33000""", ): - connection.execute(text(sql), {"now": NOW}) + # Keep the full relationship/review cardinalities and every FK/outbox + # trigger, while bounding setup statements just like article inserts. + for batch_start in range(1, 60341, 1000): + connection.execute( + text(sql), + { + "batch_start": batch_start, + "batch_end": min(batch_start + 999, 60340), + }, + ) for table in ( "articles", "sources", From 3c345f28daf971719fb3f83d5b3caf4a24af024d Mon Sep 17 00:00:00 2001 From: Abhimanyu Saharan Date: Sat, 24 Oct 2026 15:15:52 +0530 Subject: [PATCH 3/7] Keep benchmark review outcomes independent of report periods --- benchmarks/publication-reports.md | 2 +- scripts/profile_publication_reports.py | 21 +++++++++++++-------- 2 files changed, 14 insertions(+), 9 deletions(-) diff --git a/benchmarks/publication-reports.md b/benchmarks/publication-reports.md index 1d8d951e6..0a9377e5a 100644 --- a/benchmarks/publication-reports.md +++ b/benchmarks/publication-reports.md @@ -23,7 +23,7 @@ The `Publication report benchmark` workflow runs these populations on native `ub ## Workload and interpretation -The 24 October 2026 read-only production audit estimated 92,503 articles, 255,529 reviews and 156,002 analysis jobs. The fixed synthetic fixtures contain 100,000 and 300,000 articles, 3.12 reviews per article and 1.65 historical analysis jobs per article. Reviews include manual-only, automatic-only, overlapping facts, reviews after publication, reviews older than the window, duplicate evidence and a tail of 27 reviews on one article in 200. Publications span 60 days, so the 30-day cohort contains approximately half of the population. Article summaries, review notes and job JSON retain payload outside the narrow report columns. All normal database triggers and indexes remain enabled; fixture inserts use batches of at most 5,000 article identities per statement (review fanout produces more review rows). Fixture setup ends with vacuum/analyze. +The 24 October 2026 read-only production audit estimated 92,503 articles, 255,529 reviews and 156,002 analysis jobs. The fixed synthetic fixtures contain 100,000 and 300,000 articles, 3.12 reviews per article and 1.65 historical analysis jobs per article. Job timestamps follow the owning article's discovery time and include retained input JSON; job rows do not enter either report query. Reviews include manual-only, automatic-only, overlapping facts, reviews after publication, reviews older than the window, duplicate evidence and a tail of 27 reviews on one article in 201. Those extra reviews are matching publish facts, so the tail exercises repeated qualifying evidence. Publication ages use `(n / 10) % 60` while review categories use `n % 10`, keeping the evidence mix independent of the reporting period. The tail interval is coprime with the ten review categories. Publications span 60 days, so the 30-day cohort contains approximately half of the population. Article summaries, review notes and job JSON retain payload outside the narrow report columns. All normal database triggers and indexes remain enabled; fixture inserts use batches of at most 5,000 article identities per statement (review fanout produces more review rows). Fixture setup ends with vacuum/analyze. Each population exercises inclusive 1-, 7- and 30-day periods, both with and without a concurrent ingestion thread. The writer commits new unpublished articles, reviews and queued analysis jobs, with a five millisecond pause between commits. Those articles are outside the fixed report cohort, allowing exact result equality checks while ordinary writes, indexes and triggers compete for resources. Report queries use 4 MiB `work_mem`, disable parallel workers for attributable backend CPU measurements, and retain the server's JIT setting. Each strategy receives one unmeasured warmup, followed by measured samples and an instrumented plan. The baseline reproduces the correlated-existence SQL from commit `a483edec50fd331fea22c665e521e8a849cdf090`; the candidate compiles the actual application helper. diff --git a/scripts/profile_publication_reports.py b/scripts/profile_publication_reports.py index 63c40df05..0afbbcca2 100644 --- a/scripts/profile_publication_reports.py +++ b/scripts/profile_publication_reports.py @@ -35,10 +35,10 @@ publication_status,review_status,published_to_feed_at) SELECT md5(:namespace||n)::uuid,'https://profile.invalid/'||:namespace||n, md5(:namespace||n), 'Publication fixture '||n, :namespace||n, repeat('fixture ',128), - 'article',:now, :now-(n%60)*interval '1 day'-(n%86400)*interval '1 second' + 'article',:now, :now-((n/10)%60)*interval '1 day'-(n%86400)*interval '1 second' -(CASE WHEN n%37=0 THEN -60 WHEN n%10 IN (1,3,6) THEN 91*86400 ELSE 3600+n%100000 END)*interval '1 second', - 'published','approved',:now-(n%60)*interval '1 day'-(n%86400)*interval '1 second' + 'published','approved',:now-((n/10)%60)*interval '1 day'-(n%86400)*interval '1 second' FROM generate_series(CAST(:first AS integer),CAST(:last AS integer)) n""" @@ -72,17 +72,18 @@ def seed(engine, rows): (id,article_id,action,revision,automation,created_at,note) SELECT md5('publication-review-'||n||'-'||f)::uuid, md5('publication-'||n)::uuid, - CASE WHEN f=1 THEN 'publish' ELSE 'approve' END, f, + CASE WHEN f=1 OR f>3 THEN 'publish' ELSE 'approve' END, f, CASE WHEN (f=2 AND n%10 IN (1,3,6)) OR (f=3 AND n%10=4) THEN '{}'::jsonb ELSE jsonb_build_object('profile',true) END, a.published_to_feed_at + CASE WHEN (f=1 AND n%10 IN (0,1,5,6)) OR (f=3 AND n%10=4) THEN interval '1 hour' - WHEN f=2 THEN interval '-90 days' ELSE interval '-1 second' END, + WHEN f=2 AND n%10 IN (1,3,6) + THEN interval '-90 days' ELSE interval '-1 second' END, repeat('review evidence ',32) FROM generate_series(CAST(:first AS integer),CAST(:last AS integer)) n JOIN articles a ON a.id=md5('publication-'||n)::uuid - CROSS JOIN LATERAL generate_series(1,CASE WHEN n%200=0 THEN 27 ELSE 3 END) f""", + CROSS JOIN LATERAL generate_series(1,CASE WHEN n%201=0 THEN 27 ELSE 3 END) f""", rows, {"rows": rows}, ) @@ -93,9 +94,12 @@ def seed(engine, rows): input_snapshot,catalog_snapshot,result) SELECT md5('publication-job-'||n)::uuid, md5('publication-'||(1+n%:rows))::uuid,'succeeded',1, - :now-(n%180)*interval '1 day',:now-(n%180)*interval '1 day',:now, + a.discovered_at+(n%60)*interval '1 second', + a.discovered_at+(n%60)*interval '1 second', + a.discovered_at+(60+n%60)*interval '1 second', jsonb_build_object('text',repeat('historical payload ',32)), - '{}','{}' FROM generate_series(CAST(:first AS integer),CAST(:last AS integer)) n""", + '{}','{}' FROM generate_series(CAST(:first AS integer),CAST(:last AS integer)) n + JOIN articles a ON a.id=md5('publication-'||(1+n%:rows))::uuid""", rows * 165 // 100, {"now": NOW, "rows": rows}, ) @@ -292,7 +296,7 @@ def main(): seed(engine, args.rows) output = { "rows": args.rows, - "fixture_review_rows": args.rows * 3 + args.rows // 200 * 24, + "fixture_review_rows": args.rows * 3 + args.rows // 201 * 24, "fixture_analysis_job_rows": args.rows * 165 // 100, "client_architecture": platform.machine(), "postgresql": version, @@ -322,6 +326,7 @@ def main(): # Preserve failing plans and samples too; CI uploads partial output. args.output.write_text(json.dumps(output, indent=2) + "\n") assert all(result == results[0] for result in results), "Report results differ" + assert 0 < results[0][1] < results[0][0], "Fixture lost mixed publication outcomes" report["result_equality"] = True if args.rows >= 100000 and days == 30: report["work_checks"] = check_work(report) From c3211deec1ebaa204bc5395868905f0d4fa69b9a Mon Sep 17 00:00:00 2001 From: Abhimanyu Saharan Date: Sat, 24 Oct 2026 15:26:30 +0530 Subject: [PATCH 4/7] Preserve publication cohort plans after prepared statement reuse --- .../publication-report-benchmark.yml | 3 +- apps/admin-api/src/automation.py | 11 ++- benchmarks/publication-reports.md | 10 +- scripts/profile_publication_reports.py | 92 ++++++++++++++++++- .../operations/test_automation_policy_unit.py | 31 +++++++ .../test_publication_report_projection.py | 70 +++++++++++++- 6 files changed, 206 insertions(+), 11 deletions(-) diff --git a/.github/workflows/publication-report-benchmark.yml b/.github/workflows/publication-report-benchmark.yml index 7a05891a8..7e4366258 100644 --- a/.github/workflows/publication-report-benchmark.yml +++ b/.github/workflows/publication-report-benchmark.yml @@ -40,6 +40,7 @@ jobs: steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} persist-credentials: false - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 with: @@ -59,7 +60,7 @@ jobs: mkdir -p reports/publication git rev-parse HEAD > reports/publication/source-sha.txt .venv/bin/python scripts/profile_publication_reports.py --seed \ - --rows "$ARTICLE_ROWS" --repeats 7 --inline-comparison \ + --rows "$ARTICLE_ROWS" --repeats 7 --inline-comparison --prepared \ --output "reports/publication/${ARTICLE_ROWS}-arm64.json" - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 if: always() diff --git a/apps/admin-api/src/automation.py b/apps/admin-api/src/automation.py index 1f1acb84a..8d36a04f9 100644 --- a/apps/admin-api/src/automation.py +++ b/apps/admin-api/src/automation.py @@ -40,6 +40,7 @@ from pydantic import BaseModel, Field from sqlalchemy import ( BigInteger, + DateTime, String, and_, case, @@ -606,9 +607,17 @@ def automation_blocker_rows(session, article_ids): def _publication_automation_query(start, now): + # Keep cohort selectivity visible to prepared generic plans, which otherwise + # estimate an unknown date range as small and return to per-article probes. + # SQLAlchemy caches the statement and renders these values with its typed + # literal processor only at execution; other parameters remain bound. + start_bound = literal(start, type_=DateTime(timezone=True), literal_execute=True) + now_bound = literal(now, type_=DateTime(timezone=True), literal_execute=True) publications = ( select(Article.id, Article.published_to_feed_at, Article.discovered_at) - .where(Article.published_to_feed_at >= start, Article.published_to_feed_at <= now) + .where( + Article.published_to_feed_at >= start_bound, Article.published_to_feed_at <= now_bound + ) .cte("publications") .prefix_with("MATERIALIZED") ) diff --git a/benchmarks/publication-reports.md b/benchmarks/publication-reports.md index 0a9377e5a..d2247fbcc 100644 --- a/benchmarks/publication-reports.md +++ b/benchmarks/publication-reports.md @@ -13,9 +13,9 @@ export DEVFEED_TEST_DATABASE_URL=postgresql+psycopg://profile:profile@127.0.0.1: export DEVFEED_DATABASE_URL="$DEVFEED_TEST_DATABASE_URL" export DEVFEED_REDIS_URL=redis://127.0.0.1:6379/15 uv run alembic upgrade head -uv run python scripts/profile_publication_reports.py --seed --rows 100000 --repeats 7 \ +uv run python scripts/profile_publication_reports.py --seed --rows 100000 --repeats 7 --prepared \ --output reports/publication/100000.json -uv run python scripts/profile_publication_reports.py --seed --rows 300000 --repeats 7 \ +uv run python scripts/profile_publication_reports.py --seed --rows 300000 --repeats 7 --prepared \ --output reports/publication/300000.json ``` @@ -25,10 +25,14 @@ The `Publication report benchmark` workflow runs these populations on native `ub The 24 October 2026 read-only production audit estimated 92,503 articles, 255,529 reviews and 156,002 analysis jobs. The fixed synthetic fixtures contain 100,000 and 300,000 articles, 3.12 reviews per article and 1.65 historical analysis jobs per article. Job timestamps follow the owning article's discovery time and include retained input JSON; job rows do not enter either report query. Reviews include manual-only, automatic-only, overlapping facts, reviews after publication, reviews older than the window, duplicate evidence and a tail of 27 reviews on one article in 201. Those extra reviews are matching publish facts, so the tail exercises repeated qualifying evidence. Publication ages use `(n / 10) % 60` while review categories use `n % 10`, keeping the evidence mix independent of the reporting period. The tail interval is coprime with the ten review categories. Publications span 60 days, so the 30-day cohort contains approximately half of the population. Article summaries, review notes and job JSON retain payload outside the narrow report columns. All normal database triggers and indexes remain enabled; fixture inserts use batches of at most 5,000 article identities per statement (review fanout produces more review rows). Fixture setup ends with vacuum/analyze. -Each population exercises inclusive 1-, 7- and 30-day periods, both with and without a concurrent ingestion thread. The writer commits new unpublished articles, reviews and queued analysis jobs, with a five millisecond pause between commits. Those articles are outside the fixed report cohort, allowing exact result equality checks while ordinary writes, indexes and triggers compete for resources. Report queries use 4 MiB `work_mem`, disable parallel workers for attributable backend CPU measurements, and retain the server's JIT setting. Each strategy receives one unmeasured warmup, followed by measured samples and an instrumented plan. The baseline reproduces the correlated-existence SQL from commit `a483edec50fd331fea22c665e521e8a849cdf090`; the candidate compiles the actual application helper. +Each population exercises inclusive 1-, 7- and 30-day periods, both with and without a concurrent ingestion thread. The writer commits new unpublished articles, reviews and queued analysis jobs, with a five millisecond pause between commits. Those articles are outside the fixed report cohort, allowing exact result equality checks while ordinary writes, indexes and triggers compete for resources. Report queries use 4 MiB `work_mem`, disable parallel workers for attributable backend CPU measurements, and retain the server's JIT setting. Each ordinary strategy receives one unmeasured warmup, followed by measured samples and an instrumented plan. The concurrent writer must complete its first transaction before report measurement starts. The baseline reproduces the correlated-existence SQL from commit `a483edec50fd331fea22c665e521e8a849cdf090`; the candidate compiles the actual application helper. Compare root-plan shared-buffer hits/reads, temp blocks, WAL and review-node loop counts as well as elapsed time. Buffer hits are logical page accesses. PostgreSQL backend CPU comes from user/system ticks in that backend's `/proc/self/stat`, sampled through the disposable superuser connection; resolution is the host's clock tick. Application CPU is Python's calling-thread CPU around statement execution and result consumption, excluding the concurrent writer thread. Neither is an estimate of complete API request CPU. Writer throughput, commit latency and generated WAL are recorded separately; WAL includes ordinary checkpoint/full-page effects in the isolated service. Timing and writer samples are descriptive, without a flaky elapsed-time CI gate. Exact result equality is mandatory for every comparison. At 100,000 articles and above, each 30-day comparison additionally requires candidate buffer accesses below 75% of the baseline, no correlated SubPlan nodes, and total temporary read/write bytes below 512 bytes per publication. These structural budgets check repeated work and bounded intermediate data while allowing normal planner choices. `--inline-comparison` additionally profiles an inlined publication CTE when evaluating planner alternatives. +## Prepared-plan stability + +Unknown date bounds can make PostgreSQL estimate a broad reporting window as a selective one when reusing a generic prepared plan. A disposable 100,000-article experiment reproduced that regression: the set-based query returned to per-article review probes under a generic plan. The application renders only its datetime window bounds through SQLAlchemy's typed `literal_execute` mechanism, keeping those constants visible to the planner while ordinary values such as the percentile remain bound. The date serializer handles quoting; no raw request text becomes SQL. This scope preserves the connection's and server's normal plan-cache settings. The prepared-plan benchmark compiles with `render_postcompile=True` to inspect the same statement shape sent by the driver. + ## Rollout and rollback This change adds no indexes, triggers, stored facts or migration, so write paths and recovery semantics remain unchanged. Rolling back the application restores the original SQL without a database downgrade. After deployment, compare fresh statement-statistics deltas under matched report periods and ingestion workload; confirm report freshness, review precedence and duration metrics. Check PostgreSQL CPU, buffer work, temporary writes, ingestion latency and queue age independently. Local x86 runs validate the SQL and profiling mechanism; native ARM64 artifacts supply architecture-specific execution evidence. Future production diagnostics must remain bounded and read-only; this benchmark does not execute the changed query against production. Do not run these fixture or `ANALYZE` operations against production. diff --git a/scripts/profile_publication_reports.py b/scripts/profile_publication_reports.py index 0afbbcca2..c9920f8ef 100644 --- a/scripts/profile_publication_reports.py +++ b/scripts/profile_publication_reports.py @@ -17,7 +17,8 @@ from pathlib import Path from devfeed_admin_api.automation import _publication_automation_query -from sqlalchemy import create_engine, text +from sqlalchemy import create_engine, literal, text +from sqlalchemy.dialects.postgresql.psycopg import dialect from sqlalchemy.engine import make_url NOW = datetime(2026, 10, 24, 12, tzinfo=UTC) @@ -132,7 +133,6 @@ def run(self): with self.engine.connect() as connection: first_wal = connection.scalar(text("SELECT pg_current_wal_insert_lsn()")) connection.commit() - self.ready.set() while not self.stop.is_set(): started = time.perf_counter() identifier = uuid.uuid4() @@ -166,6 +166,7 @@ def run(self): ) self.rows += 1 self.samples_ms.append((time.perf_counter() - started) * 1000) + self.ready.set() # Start reports only after a successful ordinary write. self.stop.wait(0.005) self.wal_bytes = int( connection.scalar( @@ -178,20 +179,28 @@ def run(self): self.ready.set() -def measure(engine, sql, params, repeats, concurrent): +def measure(engine, sql, params, repeats, concurrent, prepared=None): ingestion = Ingestion(engine) if concurrent else None + started = time.perf_counter() if ingestion: ingestion.thread.start() if not ingestion.ready.wait(10): raise RuntimeError("Concurrent ingestion did not start") samples, cpu, app_cpu = [], [], [] - started = time.perf_counter() try: with engine.connect() as connection: connection.execute(text("SET statement_timeout='60s'")) connection.execute(text("SET work_mem='4MB'")) connection.execute(text("SET max_parallel_workers_per_gather=0")) - connection.execute(text(sql), params).one() # One unmeasured warmup per strategy. + if prepared: + name, statement, mode = prepared + connection.execute(text(f"SET plan_cache_mode='{mode}'")) + connection.execute(text(f"PREPARE {name} AS {statement}")) + # Auto-prepared plans may switch after five executions. Warm beyond + # that decision point before collecting its measured samples. + warmups = 7 if prepared and prepared[2] == "auto" else 1 + for _ in range(warmups): + connection.execute(text(sql), params).one() for _ in range(repeats): before_cpu = backend_cpu(connection) before_app = time.thread_time() @@ -203,6 +212,19 @@ def measure(engine, sql, params, repeats, concurrent): plan = connection.scalar( text("EXPLAIN (ANALYZE, BUFFERS, WAL, SETTINGS, FORMAT JSON) " + sql), params )[0] + if prepared: + plan_counts = dict( + connection.execute( + text( + "SELECT generic_plans,custom_plans FROM pg_prepared_statements " + "WHERE name=:name" + ), + {"name": prepared[0]}, + ) + .mappings() + .one() + ) + connection.execute(text(f"DEALLOCATE {prepared[0]}")) finally: elapsed = time.perf_counter() - started if ingestion: @@ -218,6 +240,8 @@ def measure(engine, sql, params, repeats, concurrent): "application_thread_cpu_samples_ms": app_cpu, "plan": plan, } + if prepared: + output["prepared_plan_counts"] = plan_counts if ingestion: output["concurrent_ingestion"] = { "commits": ingestion.rows, @@ -263,12 +287,63 @@ def touches(node): } +def prepared_profiles(engine, repeats, rows=0, preserve=None): + reports = [] + for mode in ("force_custom_plan", "force_generic_plan", "auto"): + for days in (1, 7, 30): + start = NOW - timedelta(days=days) + compiled = _publication_automation_query(start, NOW).compile( + dialect=dialect(paramstyle="numeric_dollar"), + compile_kwargs={"render_postcompile": True}, + ) + strategies = { + "baseline": ( + BASELINE.replace("percentile_cont(0.5)", "percentile_cont($1)") + .replace(":start", "$2") + .replace(":now", "$3"), + [0.5, start, NOW], + ), + "candidate": ( + str(compiled), + [compiled.params[key] for key in compiled.positiontup], + ), + } + report = {"days": days, "plan_cache_mode": mode, "strategies": {}} + for strategy, (statement, values) in strategies.items(): + name = f"profile_{strategy}_{mode}_{days}" + arguments = ",".join( + str(literal(value).compile(engine, compile_kwargs={"literal_binds": True})) + for value in values + ) + report["strategies"][strategy] = measure( + engine, + f"EXECUTE {name}({arguments})", + {}, + repeats, + False, + prepared=(name, statement, mode), + ) + report["strategies"][strategy]["sql"] = statement + results = [value["result"] for value in report["strategies"].values()] + reports.append(report) + if preserve: + preserve(reports) + assert results[0] == results[1], "Prepared report results differ" + report["result_equality"] = True + if rows >= 100000 and days == 30: + report["work_checks"] = check_work(report) + if preserve: + preserve(reports) + return reports + + def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--seed", action="store_true") parser.add_argument("--rows", type=int, default=100000) parser.add_argument("--repeats", type=int, default=5) parser.add_argument("--inline-comparison", action="store_true") + parser.add_argument("--prepared", action="store_true") parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() try: @@ -344,6 +419,13 @@ def main(): ), flush=True, ) + if args.prepared: + + def preserve(reports): + output["prepared_reports"] = reports + args.output.write_text(json.dumps(output, indent=2) + "\n") + + prepared_profiles(engine, args.repeats, args.rows, preserve) finally: engine.dispose() diff --git a/tests/admin/operations/test_automation_policy_unit.py b/tests/admin/operations/test_automation_policy_unit.py index fe5eca8b3..9c15a7c80 100644 --- a/tests/admin/operations/test_automation_policy_unit.py +++ b/tests/admin/operations/test_automation_policy_unit.py @@ -308,6 +308,37 @@ def test_publication_query_keeps_narrow_review_probes_out_of_aggregate_filters() assert review_projections and set(review_projections) == {"article_id"} +def test_publication_window_remains_planner_visible_in_cached_prepared_queries(): + from sqlalchemy import DateTime + from sqlalchemy.dialects import postgresql + from sqlalchemy.sql import visitors + from sqlalchemy.sql.elements import BindParameter + + start = NOW - timedelta(days=7) + statement = automation._publication_automation_query(start, NOW) + other = automation._publication_automation_query(start - timedelta(days=1), NOW) + assert statement._generate_cache_key().key == other._generate_cache_key().key + bounds = [ + clause + for clause in visitors.iterate(statement) + if isinstance(clause, BindParameter) and clause.literal_execute + ] + assert {bound.value for bound in bounds} == {start, NOW} + assert all(isinstance(bound.type, DateTime) and bound.type.timezone for bound in bounds) + rendered = statement.compile( + dialect=postgresql.dialect(), compile_kwargs={"render_postcompile": True} + ) + assert start not in rendered.params.values() and NOW not in rendered.params.values() + assert set(rendered.params.values()) == {0.5} + assert "POSTCOMPILE" not in str(rendered) + # SQLAlchemy may reuse the compiled expression; PostgreSQL still receives + # each window's timestamps rather than unknown generic-plan parameters. + rendered_other = other.compile( + dialect=postgresql.dialect(), compile_kwargs={"render_postcompile": True} + ) + assert str(rendered) != str(rendered_other) + + @pytest.mark.parametrize("projected", [False, True]) def test_automation_overview_composes_one_consistent_window_and_selects_correct_blocker_reader( monkeypatch, projected diff --git a/tests/admin/reporting/projections/test_publication_report_projection.py b/tests/admin/reporting/projections/test_publication_report_projection.py index cbef1b380..753982442 100644 --- a/tests/admin/reporting/projections/test_publication_report_projection.py +++ b/tests/admin/reporting/projections/test_publication_report_projection.py @@ -5,6 +5,7 @@ import pytest from devfeed_admin_api.automation import _publication_automation_query, publication_automation from devfeed_core.models import ArticleReview, utcnow +from psycopg import sql from sqlalchemy import text from tests.admin.reporting.projections.test_pipeline_projection_safety import add_article @@ -221,6 +222,73 @@ def report(): assert changed["median_ingestion_to_publication_seconds"] == pytest.approx(600.000001) +def test_prepared_generic_report_keeps_exact_results_and_literal_window_bounds(database): + start = NOW - timedelta(days=7) + with database.begin() as session: + for published, reviews in [ + (start, [("publish", {"policy": "automatic"}, start)]), + (NOW, [("publish", {"policy": "automatic"}, NOW), ("approve", {}, start)]), + (NOW, []), + (start - timedelta(microseconds=1), [("publish", {"policy": "automatic"}, start)]), + ]: + article = add_article(session, published=True, published_at=published) + article.discovered_at = published - timedelta(seconds=120) + session.add_all( + ArticleReview( + article_id=article.id, + action=action, + revision=revision, + automation=evidence, + created_at=created_at, + ) + for revision, (action, evidence, created_at) in enumerate(reviews, start=1) + ) + session.flush() + session.execute(text("SET LOCAL plan_cache_mode = force_generic_plan")) + session.execute(text("SET LOCAL enable_seqscan = off")) + compiled = _publication_automation_query(start, NOW).compile( + session.get_bind(), compile_kwargs={"render_postcompile": True} + ) + driver = session.connection().connection.driver_connection + before = dict( + driver.execute( + "SELECT name, generic_plans FROM pg_prepared_statements", prepare=False + ).fetchall() + ) + for _ in range(3): + assert driver.execute(str(compiled), compiled.params, prepare=True).fetchone() == ( + 3, + 1, + 120.0, + ) + prepared = next( + ( + row + for row in driver.execute( + "SELECT name, generic_plans FROM pg_prepared_statements " + "WHERE statement LIKE 'WITH publications AS MATERIALIZED%'", + prepare=False, + ) + if row[1] - before.get(row[0], 0) >= 3 + ), + None, + ) + assert prepared is not None + plan = "\n".join( + row[0] + for row in driver.execute( + sql.SQL("EXPLAIN EXECUTE {} (0.5)").format(sql.Identifier(prepared[0])), + prepare=False, + ) + ) + # Even under a forced generic plan, selectivity estimates can see the actual + # cohort; the percentile fraction remains the sole prepared parameter. + assert "ix_article_reviews_manual_publication" in plan + assert "ix_article_reviews_automatic_publication" in plan + assert start.strftime("%Y-%m-%d %H:%M:%S") in plan + assert NOW.strftime("%Y-%m-%d %H:%M:%S") in plan + + def test_large_review_evidence_uses_partial_indexes_without_changing_manual_cutoff(database): now = utcnow() evidence = {"policy": "automatic", "evidence": "Unicode extraction evidence 設計 " * 3000} @@ -258,7 +326,7 @@ def test_large_review_evidence_uses_partial_indexes_without_changing_manual_cuto # exact application predicates can use narrow indexes at large scale. session.execute(text("SET LOCAL enable_seqscan = off")) compiled = _publication_automation_query(now - timedelta(days=1), now).compile( - session.get_bind() + session.get_bind(), compile_kwargs={"render_postcompile": True} ) plan = "\n".join( row[0] From cfed74b4d3e5f4949df15fdbef73a17372c9cc88 Mon Sep 17 00:00:00 2001 From: Abhimanyu Saharan Date: Sat, 24 Oct 2026 15:45:28 +0530 Subject: [PATCH 5/7] Record native publication report benchmarks and scaling tradeoffs --- benchmarks/publication-reports.md | 46 +++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/benchmarks/publication-reports.md b/benchmarks/publication-reports.md index d2247fbcc..bddf356b4 100644 --- a/benchmarks/publication-reports.md +++ b/benchmarks/publication-reports.md @@ -33,6 +33,52 @@ Compare root-plan shared-buffer hits/reads, temp blocks, WAL and review-node loo Unknown date bounds can make PostgreSQL estimate a broad reporting window as a selective one when reusing a generic prepared plan. A disposable 100,000-article experiment reproduced that regression: the set-based query returned to per-article review probes under a generic plan. The application renders only its datetime window bounds through SQLAlchemy's typed `literal_execute` mechanism, keeping those constants visible to the planner while ordinary values such as the percentile remain bound. The date serializer handles quoting; no raw request text becomes SQL. This scope preserves the connection's and server's normal plan-cache settings. The prepared-plan benchmark compiles with `render_postcompile=True` to inspect the same statement shape sent by the driver. +## Native ARM64 results + +Measured source [`c3211deec1ebaa204bc5395868905f0d4fa69b9a`](https://github.com/abhi1693/devfeed.tech/commit/c3211deec1ebaa204bc5395868905f0d4fa69b9a), [workflow run 40083944048](https://github.com/abhi1693/devfeed.tech/actions/runs/40083944048), completed 24 October 2026. Both jobs passed on PostgreSQL 17.11 `aarch64-unknown-linux-gnu`. The 100,000-article fixture has 311,928 reviews and 165,000 historical analysis jobs; the 300,000-article fixture has 935,808 reviews and 495,000 jobs. All 1-, 7- and 30-day comparisons preserved results, including forced custom, forced generic and warmed automatic prepared plans. All broad-period buffer, correlated-subplan and temporary-work guards passed. + +Each cell shows baseline → candidate. Times are medians of seven measured executions. Buffer accesses include shared hits **and** reads. Backend CPU is serial PostgreSQL user/system CPU with a 10 ms clock-tick resolution; it is not whole-service CPU. Temporary writes are MiB reported by the instrumented plan. + +| Articles | Days | Concurrent ingestion | Median ms | Buffer accesses | Backend CPU ms | Temp written MiB | +| -------: | ---: | :------------------- | --------------: | ----------------: | -------------: | ---------------: | +| 100,000 | 1 | No | 6.81 → 6.90 | 7,533 → 7,533 | 10 → 10 | 0.0 → 0.0 | +| 100,000 | 7 | No | 53.23 → 29.94 | 51,709 → 17,298 | 50 → 30 | 0.0 → 0.0 | +| 100,000 | 30 | No | 217.97 → 113.40 | 230,628 → 20,920 | 210 → 110 | 0.0 → 0.0 | +| 100,000 | 1 | Yes | 7.11 → 6.98 | 7,537 → 7,537 | 10 → 10 | 0.0 → 0.0 | +| 100,000 | 7 | Yes | 53.07 → 30.67 | 51,743 → 17,931 | 50 → 30 | 0.0 → 0.0 | +| 100,000 | 30 | Yes | 222.17 → 116.21 | 231,287 → 24,795 | 220 → 120 | 0.0 → 0.0 | +| 300,000 | 1 | No | 17.07 → 17.51 | 25,568 → 25,568 | 20 → 20 | 0.0 → 0.0 | +| 300,000 | 7 | No | 136.79 → 109.84 | 175,842 → 253,742 | 140 → 110 | 0.0 → 0.0 | +| 300,000 | 30 | No | 688.75 → 428.30 | 780,918 → 306,141 | 690 → 430 | 0.0 → 21.3 | +| 300,000 | 1 | Yes | 17.96 → 17.88 | 25,572 → 25,572 | 20 → 20 | 0.0 → 0.0 | +| 300,000 | 7 | Yes | 134.12 → 112.82 | 175,936 → 256,454 | 130 → 110 | 0.0 → 0.0 | +| 300,000 | 30 | Yes | 673.89 → 429.88 | 782,877 → 319,983 | 670 → 430 | 0.0 → 21.4 | + +The 300,000-article seven-day candidate uses more logical buffer accesses despite lower ordinary-run latency and CPU. In warmed automatic prepared mode it measured 117.74 ms against 113.40 ms for the baseline, with 272,379 against 176,736 accesses. One-day report times are effectively unchanged at this sample size. This change does not promise an improvement for every period and planner mode. + +The 300,000-article 30-day candidate adds about 21.3–21.4 MiB of temporary writes and the same amount of temporary reads; the baseline writes none. The combined 42.7–42.8 MiB is below the 73.2 MiB budget for its 150,000-publication cohort. Materialization performs less article work than the inlined alternative, which measured 528.73 ms steady and 535.86 ms with ingestion, compared with 428.30 and 429.88 ms for materialization. At 100,000 articles neither strategy spills. + +The exact driver SQL retains the improvement after prepared-plan adoption. These are the 30-day results; the artifacts also retain all shorter-period plans and samples. + +| Articles | Prepared plan mode | Median ms | Buffer accesses | +| -------: | :----------------- | --------------: | ----------------: | +| 100,000 | Forced custom | 215.66 → 118.76 | 231,807 → 26,101 | +| 100,000 | Forced generic | 168.74 → 116.20 | 222,202 → 26,101 | +| 100,000 | Warmed automatic | 168.25 → 116.55 | 222,202 → 26,101 | +| 300,000 | Forced custom | 663.70 → 431.11 | 784,734 → 324,778 | +| 300,000 | Forced generic | 547.84 → 433.45 | 755,649 → 324,778 | +| 300,000 | Warmed automatic | 567.32 → 436.38 | 755,649 → 324,778 | + +Each warmed automatic statement recorded five custom executions followed by ten generic executions, including the instrumented plan. The forced modes recorded nine executions of their requested plan type. The candidate therefore preserves known window selectivity even when PostgreSQL chooses a generic plan. + +At 100,000 articles, the concurrent 30-day writer measured 112.7 → 115.1 commits/s, 3.80 → 3.65 ms median commit time, and 24.3 → 22.0 KiB WAL per commit, with no errors. + +At 300,000 articles, the concurrent 30-day writer measured 119.5 → 121.6 commits/s, 3.26 → 3.10 ms median commit time, and 39.9 → 34.7 KiB WAL per commit, with no errors. + +The Python calling-thread CPU medians span 0.32–0.74 ms for the baseline and 0.24–0.60 ms for the candidate. All report plans record zero WAL bytes. Writer WAL differences include ordinary full-page/checkpoint effects and are descriptive. The short 300,000-article one-day candidate writer interval contains only 18 commits and one 35.4 ms maximum commit, producing 86.0 commits/s; this sample does not establish a sustained ingestion regression or a production percentile. + +These fixtures match representative job cardinality with smaller retained job payloads: a local 165,000-job fixture occupied about 182 MB, while the production audit estimated 2.45 GB for 156,002 jobs. Those are total relation footprints, including indexes. Neither publication query reads job payloads. The benchmark establishes publication/review query behavior under concurrent ordinary inserts; it does not reproduce every production body, historical job blob, storage condition, parallel plan or complete application workload. It makes no production p95 claim. Production requires the matched-workload follow-up described below. + ## Rollout and rollback This change adds no indexes, triggers, stored facts or migration, so write paths and recovery semantics remain unchanged. Rolling back the application restores the original SQL without a database downgrade. After deployment, compare fresh statement-statistics deltas under matched report periods and ingestion workload; confirm report freshness, review precedence and duration metrics. Check PostgreSQL CPU, buffer work, temporary writes, ingestion latency and queue age independently. Local x86 runs validate the SQL and profiling mechanism; native ARM64 artifacts supply architecture-specific execution evidence. Future production diagnostics must remain bounded and read-only; this benchmark does not execute the changed query against production. Do not run these fixture or `ANALYZE` operations against production. From a260b4dfbd11112d2d570c0f3f46b3f678cbd613 Mon Sep 17 00:00:00 2001 From: Abhimanyu Saharan Date: Sat, 24 Oct 2026 18:59:39 +0530 Subject: [PATCH 6/7] Harden publication profiling and cover benchmark behavior --- .../publication-report-benchmark.yml | 51 ++- benchmarks/publication-reports.md | 6 +- scripts/profile_publication_reports.py | 53 ++- tests/tooling/ci/test_sonar_coverage.py | 1 + .../test_publication_profile_script.py | 430 ++++++++++++++++++ 5 files changed, 515 insertions(+), 26 deletions(-) create mode 100644 tests/tooling/profiling/test_publication_profile_script.py diff --git a/.github/workflows/publication-report-benchmark.yml b/.github/workflows/publication-report-benchmark.yml index 7e4366258..92ee5f4df 100644 --- a/.github/workflows/publication-report-benchmark.yml +++ b/.github/workflows/publication-report-benchmark.yml @@ -20,22 +20,7 @@ jobs: fail-fast: false matrix: rows: [100000, 300000] - services: - postgres: - image: postgres:17.11 - env: - POSTGRES_USER: profile - POSTGRES_PASSWORD: profile - POSTGRES_DB: publication_report_test - ports: - - 127.0.0.1:55496:5432 - options: >- - --cpus=2 --memory=3g --shm-size=512m - --health-cmd="pg_isready -U profile -d publication_report_test" - --health-interval=5s --health-timeout=5s --health-retries=10 env: - DEVFEED_TEST_DATABASE_URL: postgresql+psycopg://profile:profile@127.0.0.1:55496/publication_report_test - DEVFEED_DATABASE_URL: postgresql+psycopg://profile:profile@127.0.0.1:55496/publication_report_test DEVFEED_REDIS_URL: redis://127.0.0.1:6379/15 steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 @@ -51,6 +36,39 @@ jobs: enable-cache: true - name: Install locked workspace dependencies run: sh scripts/ci/python-bootstrap.sh . --editable --all-packages + - name: Start isolated PostgreSQL with a fresh password + run: | + benchmark_password="$(openssl rand -hex 24)" + echo "::add-mask::$benchmark_password" + export POSTGRES_PASSWORD="$benchmark_password" + .venv/bin/python - <<'PY' + import os + from sqlalchemy import URL + + url = URL.create( + "postgresql+psycopg", username="profile", password=os.environ["POSTGRES_PASSWORD"], + host="127.0.0.1", port=55496, database="publication_report_test", + ).render_as_string(hide_password=False) + with open(os.environ["GITHUB_ENV"], "a") as output: + for key in ("DEVFEED_DATABASE_URL", "DEVFEED_TEST_DATABASE_URL"): + output.write(f"{key}={url}\n") + PY + docker run --detach --name publication-report-postgres \ + --cpus=2 --memory=3g --shm-size=512m \ + --env POSTGRES_USER=profile --env POSTGRES_DB=publication_report_test \ + --env POSTGRES_PASSWORD --publish 127.0.0.1:55496:5432 postgres:17.11 + for attempt in {1..30}; do + if docker exec publication-report-postgres \ + pg_isready --host 127.0.0.1 -U profile -d publication_report_test; then + exit 0 + fi + if [ "$attempt" -eq 30 ]; then + break + fi + sleep 1 + done + docker logs publication-report-postgres + exit 1 - name: Apply disposable database migrations run: .venv/bin/alembic upgrade head - name: Compare fixed publication cohorts under steady and concurrent ingestion @@ -69,3 +87,6 @@ jobs: path: reports/publication/ if-no-files-found: error retention-days: 7 + - name: Remove disposable PostgreSQL + if: always() + run: docker rm -fv publication-report-postgres >/dev/null 2>&1 || true diff --git a/benchmarks/publication-reports.md b/benchmarks/publication-reports.md index bddf356b4..c9c123776 100644 --- a/benchmarks/publication-reports.md +++ b/benchmarks/publication-reports.md @@ -9,7 +9,7 @@ Manual evidence still takes precedence, regardless of the review action. Automat Use a dedicated PostgreSQL 17 instance with a database ending in `_test`. The profiler refuses remote database hosts, connection query overrides and inherited libpq host/service overrides. `--seed` truncates articles and their dependent rows. Apply the repository's migrations to the disposable instance first, then run: ```sh -export DEVFEED_TEST_DATABASE_URL=postgresql+psycopg://profile:profile@127.0.0.1:55496/publication_report_test +export DEVFEED_TEST_DATABASE_URL="postgresql+psycopg://profile:${DEVFEED_PROFILE_PASSWORD:?Set the disposable database password}@127.0.0.1:55496/publication_report_test" export DEVFEED_DATABASE_URL="$DEVFEED_TEST_DATABASE_URL" export DEVFEED_REDIS_URL=redis://127.0.0.1:6379/15 uv run alembic upgrade head @@ -19,7 +19,9 @@ uv run python scripts/profile_publication_reports.py --seed --rows 300000 --repe --output reports/publication/300000.json ``` -The `Publication report benchmark` workflow runs these populations on native `ubuntu-24.04-arm`, with PostgreSQL 17.11 limited to two CPUs and 3 GiB memory. It runs for relevant pull requests and accepts manual dispatch. Download the `publication-report-*-arm64-*` artifacts within seven days. Each artifact includes the exact SQL, source SHA, PostgreSQL architecture/version and settings, all timing/CPU samples and complete `EXPLAIN (ANALYZE, BUFFERS, WAL, SETTINGS, FORMAT JSON)` plans. Source and report data remain separate: generated reports belong in the ignored `reports/` directory. +Set `DEVFEED_PROFILE_PASSWORD` to the disposable instance's password before running these commands. Output must be a `.json` file beneath this checkout's `reports/publication` directory; escaping paths and redirected report directories are rejected before database access. Each saved report replaces its predecessor atomically, preserving earlier output if serialization fails. + +The `Publication report benchmark` workflow runs these populations on native `ubuntu-24.04-arm`, with PostgreSQL 17.11 limited to two CPUs and 3 GiB memory. It creates a fresh masked password for each job, binds the disposable database to loopback, and removes the container afterward. It runs for relevant pull requests and accepts manual dispatch. Download the `publication-report-*-arm64-*` artifacts within seven days. Each artifact includes the exact SQL, source SHA, PostgreSQL architecture/version and settings, all timing/CPU samples and complete `EXPLAIN (ANALYZE, BUFFERS, WAL, SETTINGS, FORMAT JSON)` plans. Source and report data remain separate: generated reports belong in the ignored `reports/` directory. ## Workload and interpretation diff --git a/scripts/profile_publication_reports.py b/scripts/profile_publication_reports.py index c9920f8ef..bfafc9b31 100644 --- a/scripts/profile_publication_reports.py +++ b/scripts/profile_publication_reports.py @@ -2,7 +2,8 @@ DEVFEED_TEST_DATABASE_URL must name a local database ending in _test. --seed truncates article fixtures (including dependent rows). Apply migrations first. -No production rows or credentials are used. Full JSON plans are retained. +No production rows or credentials are used. Full JSON plans are retained under +the repository's reports/publication directory; --output must name a JSON file. """ import argparse @@ -10,6 +11,7 @@ import os import platform import statistics +import tempfile import threading import time import uuid @@ -63,6 +65,39 @@ def validate_target(value): return url +def validate_output(value): + repository = Path(__file__).resolve().parents[1] + root = repository / "reports" / "publication" + if root.resolve() != root: + raise ValueError("The reports/publication directory must not be redirected by symlinks") + output = (repository / value).resolve() + if not output.is_relative_to(root) or output.suffix != ".json": + raise ValueError("Output must be a JSON file beneath the repository's reports/publication") + if output.exists() and not output.is_file(): + raise ValueError("Output must be a regular JSON file") + if any(parent.exists() and not parent.is_dir() for parent in output.parents): + raise ValueError("Output parent paths must be directories") + return output + + +def write_report(path, report): + output = validate_output(path) + output.parent.mkdir(parents=True, exist_ok=True) + # Recheck when saving each sample: the benchmark may run for several minutes. + output = validate_output(output) + payload = json.dumps(report, indent=2) + "\n" + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=output.parent, prefix=".publication-", delete=False + ) as temporary: + try: + temporary.write(payload) + temporary.close() + # Replace the directory entry, avoiding writes through existing hard links. + Path(temporary.name).replace(output) + finally: + Path(temporary.name).unlink(missing_ok=True) + + def seed(engine, rows): with engine.begin() as connection: connection.execute(text("TRUNCATE articles CASCADE")) @@ -182,12 +217,12 @@ def run(self): def measure(engine, sql, params, repeats, concurrent, prepared=None): ingestion = Ingestion(engine) if concurrent else None started = time.perf_counter() - if ingestion: - ingestion.thread.start() - if not ingestion.ready.wait(10): - raise RuntimeError("Concurrent ingestion did not start") samples, cpu, app_cpu = [], [], [] try: + if ingestion: + ingestion.thread.start() + if not ingestion.ready.wait(10): + raise RuntimeError("Concurrent ingestion did not start") with engine.connect() as connection: connection.execute(text("SET statement_timeout='60s'")) connection.execute(text("SET work_mem='4MB'")) @@ -347,6 +382,7 @@ def main(): parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() try: + args.output = validate_output(args.output) url = validate_target(os.environ["DEVFEED_TEST_DATABASE_URL"]) if not 100 <= args.rows <= 1000000 or not 1 <= args.repeats <= 20: raise ValueError("Use 100..1000000 rows and 1..20 repetitions") @@ -379,7 +415,6 @@ def main(): "repeats": args.repeats, "reports": [], } - args.output.parent.mkdir(parents=True, exist_ok=True) for concurrent in (False, True): for days in (1, 7, 30): params = {"start": NOW - timedelta(days=days), "now": NOW} @@ -399,13 +434,13 @@ def main(): results = [value["result"] for value in report["strategies"].values()] output["reports"].append(report) # Preserve failing plans and samples too; CI uploads partial output. - args.output.write_text(json.dumps(output, indent=2) + "\n") + write_report(args.output, output) assert all(result == results[0] for result in results), "Report results differ" assert 0 < results[0][1] < results[0][0], "Fixture lost mixed publication outcomes" report["result_equality"] = True if args.rows >= 100000 and days == 30: report["work_checks"] = check_work(report) - args.output.write_text(json.dumps(output, indent=2) + "\n") + write_report(args.output, output) print( json.dumps( { @@ -423,7 +458,7 @@ def main(): def preserve(reports): output["prepared_reports"] = reports - args.output.write_text(json.dumps(output, indent=2) + "\n") + write_report(args.output, output) prepared_profiles(engine, args.repeats, args.rows, preserve) finally: diff --git a/tests/tooling/ci/test_sonar_coverage.py b/tests/tooling/ci/test_sonar_coverage.py index 1b942b2fc..e737a640c 100644 --- a/tests/tooling/ci/test_sonar_coverage.py +++ b/tests/tooling/ci/test_sonar_coverage.py @@ -167,6 +167,7 @@ def test_generated_duplication_policy_uses_only_shared_generator_output_conventi "scripts/ci/unit_shards.py", "scripts/ci/coverage_report.py", "scripts/ci/api_contracts.py", + "scripts/profile_publication_reports.py", ], ) def test_handwritten_product_migration_and_generator_code_keeps_all_quality_checks(source): diff --git a/tests/tooling/profiling/test_publication_profile_script.py b/tests/tooling/profiling/test_publication_profile_script.py new file mode 100644 index 000000000..0531cb7aa --- /dev/null +++ b/tests/tooling/profiling/test_publication_profile_script.py @@ -0,0 +1,430 @@ +"""Publication benchmarks fail safely and retain trustworthy measurement evidence.""" + +import json +import sys +from datetime import timedelta +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +from sqlalchemy import text +from sqlalchemy.dialects.postgresql.psycopg import dialect + +from scripts import profile_publication_reports as profile + + +@pytest.mark.parametrize( + "url", + [ + "sqlite:///publication_test", + "postgresql://ci@production.invalid/publication_test", + "postgresql://ci@localhost/publication", + "postgresql://ci@localhost", + "postgresql://ci@localhost/publication_test?host=production.invalid", + "postgresql://ci@localhost/publication_test?dbname=production", + "postgresql://ci@localhost/publication_test?service=production", + ], +) +def test_target_validation_refuses_non_disposable_or_redirected_databases(url): + with pytest.raises(ValueError, match="disposable local PostgreSQL"): + profile.validate_target(url) + + +@pytest.mark.parametrize("name", ["PGHOSTADDR", "PGSERVICE", "PGSERVICEFILE"]) +def test_target_validation_refuses_inherited_libpq_routing(monkeypatch, name): + monkeypatch.setenv(name, "production.invalid") + with pytest.raises(ValueError, match="without overrides"): + profile.validate_target("postgresql://ci@localhost/publication_test") + + +@pytest.mark.parametrize("host", ["localhost", "127.0.0.1"]) +def test_target_validation_accepts_explicit_local_test_database(host): + url = profile.validate_target(f"postgresql+psycopg://ci@{host}:54321/publication_test") + assert url.host == host + assert url.port == 54321 + assert url.database == "publication_test" + + +def test_failed_ingestion_signals_readiness_without_claiming_commits(measurement_engine): + engine, _ = measurement_engine + engine.connect.side_effect = RuntimeError("database disconnected") + ingestion = profile.Ingestion(engine) + ingestion.run() + assert ingestion.ready.is_set() + assert ingestion.errors == ["database disconnected"] + assert ingestion.rows == 0 + assert ingestion.samples_ms == [] + + +def work_report(*, candidate=None): + return { + "strategies": { + "baseline": {"plan": {"Plan": {"Shared Hit Blocks": 100, "Shared Read Blocks": 20}}}, + "candidate": { + "plan": {"Plan": candidate or {"Shared Hit Blocks": 10, "Shared Read Blocks": 20}}, + "result": [32, 10, 60], + }, + } + } + + +def test_work_gate_counts_reads_and_hits_and_reports_temporary_budget(): + report = work_report( + candidate={"Shared Hit Blocks": 10, "Shared Read Blocks": 20, "Temp Written Blocks": 2} + ) + assert profile.check_work(report) == { + "candidate_buffer_fraction": 0.25, + "no_correlated_subplan": True, + "temporary_bytes": 16384, + "temporary_budget_bytes": 16384, + } + + +@pytest.mark.parametrize( + ("candidate", "message"), + [ + ({"Shared Hit Blocks": 90}, "buffer work regressed"), + ({"Shared Read Blocks": 91}, "buffer work regressed"), + ( + {"Plans": [{"Plans": [{"Parent Relationship": "SubPlan"}]}]}, + "per-publication correlated SubPlan", + ), + ({"Temp Read Blocks": 1, "Temp Written Blocks": 2}, "temporary work"), + ], +) +def test_work_gate_rejects_repeated_probes_and_excessive_buffer_or_temporary_work( + candidate, message +): + with pytest.raises(AssertionError, match=message): + profile.check_work(work_report(candidate=candidate)) + + +def test_backend_cpu_parses_process_names_with_spaces_and_parentheses(monkeypatch): + fields = ["S", *("0" for _ in range(10)), "22", "11"] + connection = MagicMock() + connection.scalar.return_value = "41 (postgres (report worker)) " + " ".join(fields) + monkeypatch.setattr(profile.os, "sysconf", lambda name: 100) + assert profile.backend_cpu(connection) == 0.33 + + +def test_seed_batches_never_loses_the_partial_final_batch(): + connection = MagicMock() + profile.seed_batches(connection, "SELECT :first,:last,:tag", 5001, {"tag": "fixture"}) + assert [call.args[1] for call in connection.execute.call_args_list] == [ + {"first": 1, "last": 5000, "tag": "fixture"}, + {"first": 5001, "last": 5001, "tag": "fixture"}, + ] + + +@pytest.fixture +def measurement_engine(): + engine = MagicMock() + connection = engine.connect.return_value.__enter__.return_value + connection.execute.return_value.one.return_value = (32, 10, 60) + connection.execute.return_value.mappings.return_value.one.return_value = { + "generic_plans": 3, + "custom_plans": 5, + } + connection.scalar.return_value = [{"Plan": {"Node Type": "Aggregate"}}] + return engine, connection + + +@pytest.mark.parametrize("mode", [None, "force_custom_plan", "force_generic_plan", "auto"]) +def test_measurement_excludes_warmups_and_retains_prepared_plan_counters( + monkeypatch, measurement_engine, mode +): + engine, connection = measurement_engine + cpu = iter([0.1, 0.11, 0.2, 0.22]) + monkeypatch.setattr(profile, "backend_cpu", lambda _: next(cpu)) + prepared = ("publication_plan", "SELECT 32,10,60", mode) if mode else None + result = profile.measure(engine, "SELECT :count,10,60", {"count": 32}, 2, False, prepared) + assert result["result"] == [32, 10, 60] + assert len(result["samples_ms"]) == len(result["application_thread_cpu_samples_ms"]) == 2 + assert result["backend_cpu_samples_ms"] == pytest.approx([10, 20]) + assert result["plan"] == {"Plan": {"Node Type": "Aggregate"}} + commands = [str(call.args[0]) for call in connection.execute.call_args_list] + assert commands.count("SELECT :count,10,60") == (9 if mode == "auto" else 3) + if mode: + assert result["prepared_plan_counts"] == {"generic_plans": 3, "custom_plans": 5} + assert "PREPARE publication_plan AS SELECT 32,10,60" in commands + assert commands[-1] == "DEALLOCATE publication_plan" + else: + assert "prepared_plan_counts" not in result + engine.connect.return_value.__exit__.assert_called_once() + + +@pytest.mark.parametrize("failure", ["timeout", "writer_error", "thread_alive"]) +def test_measurement_stops_concurrent_writer_on_startup_or_measurement_failure( + monkeypatch, measurement_engine, failure +): + engine, connection = measurement_engine + ingestion = MagicMock() + ingestion.ready.wait.return_value = failure != "timeout" + ingestion.thread.is_alive.return_value = failure == "thread_alive" + ingestion.errors = ["write failed"] if failure == "writer_error" else [] + monkeypatch.setattr(profile, "Ingestion", lambda _: ingestion) + monkeypatch.setattr(profile, "backend_cpu", lambda _: 0) + if failure != "timeout": + connection.execute.side_effect = RuntimeError("report failed") + with pytest.raises(RuntimeError): + profile.measure(engine, "SELECT 1", {}, 1, True) + ingestion.stop.set.assert_called_once() + ingestion.thread.join.assert_called_once_with(10) + + +@pytest.mark.parametrize("unequal", [False, True]) +def test_prepared_profiles_keep_all_plan_modes_and_preserve_mismatched_evidence( + monkeypatch, unequal +): + calls, snapshots = [], [] + + def measure(engine, sql, params, repeats, concurrent, prepared): + calls.append((sql, prepared)) + result = [32, 10, 60 if not unequal or prepared[0].startswith("profile_baseline") else 61] + return {"result": result, "plan": {"Plan": {"Shared Hit Blocks": 1}}} + + engine = SimpleNamespace(dialect=dialect()) + monkeypatch.setattr(profile, "measure", measure) + monkeypatch.setattr(profile, "check_work", lambda _: {"no_correlated_subplan": True}) + + def preserve(reports): + snapshots.append(json.loads(json.dumps(reports))) + + if unequal: + with pytest.raises(AssertionError, match="Prepared report results differ"): + profile.prepared_profiles(engine, 2, 100000, preserve) + assert "result_equality" not in snapshots[-1][0] + else: + reports = profile.prepared_profiles(engine, 2, 100000, preserve) + assert {(row["days"], row["plan_cache_mode"]) for row in reports} == { + (days, mode) + for days in (1, 7, 30) + for mode in ("force_custom_plan", "force_generic_plan", "auto") + } + assert all(row["result_equality"] for row in reports) + assert all("work_checks" in row for row in reports if row["days"] == 30) + assert len(calls) == 18 + assert snapshots[-1] == reports + + +@pytest.fixture +def output_repository(monkeypatch, tmp_path): + monkeypatch.setattr( + profile, "__file__", str(tmp_path / "scripts/profile_publication_reports.py") + ) + return tmp_path + + +@pytest.mark.parametrize("absolute", [False, True]) +def test_output_validation_accepts_only_json_artifacts_beneath_publication_reports( + output_repository, absolute +): + relative = Path("reports/publication/nested/result.json") + requested = output_repository / relative if absolute else relative + target = profile.validate_output(requested) + assert target == output_repository / relative + profile.write_report(target, {"result": "first"}) + profile.write_report(target, {"result": "replacement"}) + assert json.loads(target.read_text()) == {"result": "replacement"} + assert list(target.parent.iterdir()) == [target] + + +@pytest.mark.parametrize( + "relative", + [ + "outside.json", + "reports/publication/../../outside.json", + "reports/publication/result.txt", + "reports/publication", + ], +) +def test_output_validation_rejects_paths_outside_artifact_boundary(output_repository, relative): + with pytest.raises(ValueError): + profile.validate_output(Path(relative)) + assert not (output_repository / "outside.json").exists() + + +@pytest.mark.parametrize("kind", ["root", "ancestor", "target", "directory", "file_parent"]) +def test_output_validation_rejects_symlink_escapes_and_directory_or_file_conflicts( + output_repository, tmp_path, kind +): + root = output_repository / "reports/publication" + outside = tmp_path / "outside" + outside.mkdir() + (outside / "result.json").write_text("original") + target = root / "nested/result.json" + if kind == "root": + root.parent.mkdir() + root.symlink_to(outside, target_is_directory=True) + elif kind == "ancestor": + root.mkdir(parents=True) + (root / "nested").symlink_to(outside, target_is_directory=True) + elif kind == "target": + target.parent.mkdir(parents=True) + target.symlink_to(outside / "result.json") + elif kind == "directory": + target.mkdir(parents=True) + else: + root.mkdir(parents=True) + (root / "nested").write_text("parent is a file") + with pytest.raises(ValueError): + profile.validate_output(target) + assert (outside / "result.json").read_text() == "original" + + +def test_report_serialization_failure_preserves_existing_artifact(output_repository): + target = profile.validate_output(Path("reports/publication/result.json")) + profile.write_report(target, {"result": "previous good evidence"}) + with pytest.raises(TypeError): + profile.write_report(target, {"result": object()}) + assert json.loads(target.read_text()) == {"result": "previous good evidence"} + assert list(target.parent.iterdir()) == [target] + + +def test_output_atomic_replacement_does_not_overwrite_an_external_hardlink(output_repository): + target = profile.validate_output(Path("reports/publication/result.json")) + target.parent.mkdir(parents=True) + outside = output_repository / "external.json" + outside.write_text("original") + target.hardlink_to(outside) + profile.write_report(target, {"result": "new evidence"}) + assert outside.read_text() == "original" + assert json.loads(target.read_text()) == {"result": "new evidence"} + + +def run_cli(monkeypatch, root, *options): + monkeypatch.setenv("DEVFEED_TEST_DATABASE_URL", "postgresql://ci@localhost/publication_test") + output = root / "reports/publication/result.json" + monkeypatch.setattr(sys, "argv", ["publication-profile", "--output", str(output), *options]) + profile.main() + return json.loads(output.read_text()) + + +@pytest.mark.parametrize( + "options", [("--rows", "99"), ("--rows", "1000001"), ("--repeats", "0"), ("--repeats", "21")] +) +def test_cli_refuses_unbounded_profiles_before_opening_database( + monkeypatch, output_repository, options +): + monkeypatch.setattr(profile, "create_engine", lambda *args, **kwargs: pytest.fail("Opened DB")) + with pytest.raises(SystemExit) as error: + run_cli(monkeypatch, output_repository, *options) + assert error.value.code == 2 + + +@pytest.mark.parametrize("failure", [None, "version", "different_results", "uniform_fixture"]) +def test_cli_retains_evidence_and_disposes_engine_after_report_failures( + monkeypatch, output_repository, measurement_engine, failure +): + engine, connection = measurement_engine + engine.dialect = dialect() + engine.dialect.server_version_info = (18, 0) if failure == "version" else (17, 5) + connection.dialect = engine.dialect + connection.scalar.return_value = "PostgreSQL fixture" + connection.execute.return_value.all.return_value = [("work_mem", "4MB")] + monkeypatch.setattr(profile, "create_engine", lambda *_args, **_kwargs: engine) + seeded, calls, prepared = [], [], [] + monkeypatch.setattr(profile, "seed", lambda _engine, rows: seeded.append(rows)) + + def measure(_engine, sql, params, repeats, concurrent): + calls.append((sql, concurrent)) + result = [32, 10, 60] + if failure == "different_results" and len(calls) == 2: + result[2] = 61 + if failure == "uniform_fixture": + result[1] = 0 + return {"result": result, "median_ms": 1, "plan": {"Plan": {}}} + + def prepared_profiles(_engine, repeats, rows, preserve): + prepared.append((repeats, rows)) + preserve([{"prepared": True}]) + + monkeypatch.setattr(profile, "measure", measure) + monkeypatch.setattr(profile, "prepared_profiles", prepared_profiles) + monkeypatch.setattr(profile, "check_work", lambda _: {"no_correlated_subplan": True}) + if failure: + with pytest.raises((ValueError, AssertionError)): + run_cli(monkeypatch, output_repository) + output = output_repository / "reports/publication/result.json" + if failure != "version": + report = json.loads(output.read_text())["reports"][0] + assert "result_equality" not in report + assert set(report["strategies"]) == {"baseline", "candidate"} + else: + output = run_cli( + monkeypatch, output_repository, "--seed", "--inline-comparison", "--prepared" + ) + assert seeded == [100000] + assert prepared == [(5, 100000)] + assert output["prepared_reports"] == [{"prepared": True}] + assert output["fixture_analysis_job_rows"] == 165000 + assert {(report["days"], report["concurrent"]) for report in output["reports"]} == { + (days, concurrent) for days in (1, 7, 30) for concurrent in (False, True) + } + assert all(report["result_equality"] for report in output["reports"]) + assert all("candidate_inline" in report["strategies"] for report in output["reports"]) + assert all("work_checks" in report for report in output["reports"] if report["days"] == 30) + engine.dispose.assert_called_once() + + +@pytest.mark.integration +def test_seed_produces_mixed_equivalent_reports_and_clears_previous_fixture(database): + engine = database.kw["bind"] + for rows in (203, 100): + profile.seed(engine, rows) + with engine.connect() as connection: + assert connection.scalar(text("SELECT count(*) FROM articles")) == rows + assert ( + connection.scalar(text("SELECT count(*) FROM article_reviews")) + == rows * 3 + rows // 201 * 24 + ) + assert ( + connection.scalar(text("SELECT count(*) FROM article_analysis_jobs")) + == rows * 165 // 100 + ) + for days in (1, 7, 30): + start = profile.NOW - timedelta(days=days) + baseline = connection.execute( + text(profile.BASELINE), {"start": start, "now": profile.NOW} + ).one() + candidate = connection.execute( + profile._publication_automation_query(start, profile.NOW) + ).one() + assert baseline == candidate + assert 0 < candidate[1] < candidate[0] + + +@pytest.mark.integration +def test_real_measurements_include_prepared_counters_and_durable_concurrent_writes(database): + engine = database.kw["bind"] + profile.seed(engine, 100) + result = profile.measure(engine, "SELECT 32,10,60", {}, 2, True) + assert result["result"] == [32, 10, 60] + writer = result["concurrent_ingestion"] + assert writer["commits"] >= 1 + assert writer["wal_bytes"] > 0 + assert writer["errors"] == [] + assert writer["commits_per_second"] > 0 + with engine.connect() as connection: + assert connection.scalar(text("SELECT count(*) FROM articles")) == 100 + writer["commits"] + assert ( + connection.scalar(text("SELECT count(*) FROM article_reviews")) + == 300 + writer["commits"] + ) + assert ( + connection.scalar(text("SELECT count(*) FROM article_analysis_jobs")) + == 165 + writer["commits"] + ) + for mode in ("force_custom_plan", "force_generic_plan", "auto"): + result = profile.measure( + engine, + "EXECUTE publication_fixture", + {}, + 2, + False, + ("publication_fixture", "SELECT 32,10,60", mode), + ) + assert result["result"] == [32, 10, 60] + counters = result["prepared_plan_counts"] + assert counters["generic_plans"] + counters["custom_plans"] >= 4 From 9467692702035486191b351c49f3be16df9655e7 Mon Sep 17 00:00:00 2001 From: Abhimanyu Saharan Date: Sat, 24 Oct 2026 21:05:24 +0530 Subject: [PATCH 7/7] Probe declared upload limits before sending rejected bodies --- apps/web/tests/browser/reader-gateway.mjs | 43 ++++++++++++++++++----- 1 file changed, 34 insertions(+), 9 deletions(-) diff --git a/apps/web/tests/browser/reader-gateway.mjs b/apps/web/tests/browser/reader-gateway.mjs index 9d722955c..d134f4e40 100644 --- a/apps/web/tests/browser/reader-gateway.mjs +++ b/apps/web/tests/browser/reader-gateway.mjs @@ -316,10 +316,11 @@ async function checkExactByteLimits() { assert.equal((await response.json()).received_bytes, limit); assert.deepEqual(requests.at(-1).body, exact); const before = requests.length; - const rejected = await call(resource, method, Buffer.alloc(limit + 1, 97)); + const rejected = await rawRequest(resource, method, { declaredLength: limit + 1 }); assert.equal(rejected.status, 413, `${method} ${resource} rejects a declared overflow`); - assert.deepEqual(await rejected.json(), { detail: "Request too large" }); - assert.equal(rejected.headers.get("connection"), "close"); + assert.deepEqual(JSON.parse(rejected.body.toString()), { detail: "Request too large" }); + assert.equal(rejected.headers.connection, "close"); + assert.equal(rejected.uploadEnded, false, "Declared overflow is rejected before body upload"); assert.equal(requests.length, before); const chunked = await rawRequest(resource, method, { body: Buffer.alloc(limit + 1, 97), @@ -339,9 +340,12 @@ async function checkExactByteLimits() { assert.equal((await accepted.json()).received_bytes, avatarLimit); assert.deepEqual(requests.at(-1).body, Buffer.from(unicode)); const unicodeBefore = requests.length; - const declaredUnicode = await call(avatar, "POST", unicode + "x"); + const declaredUnicode = await rawRequest(avatar, "POST", { + declaredLength: Buffer.byteLength(unicode + "x"), + }); assert.equal(declaredUnicode.status, 413); - assert.deepEqual(await declaredUnicode.json(), { detail: "Request too large" }); + assert.deepEqual(JSON.parse(declaredUnicode.body.toString()), { detail: "Request too large" }); + assert.equal(declaredUnicode.uploadEnded, false); const chunkedUnicode = await rawRequest(avatar, "POST", { body: Buffer.from(unicode + "x") }); assert.equal(chunkedUnicode.status, 413); assert.deepEqual(JSON.parse(chunkedUnicode.body.toString()), { detail: "Request too large" }); @@ -358,16 +362,28 @@ async function checkExactByteLimits() { assert.deepEqual(requests.at(-1).body, exact); assert.equal(requests.at(-1).headers.cookie, readerCookies); const overflowBefore = requests.length; - const overflow = await call(resource, "POST", Buffer.concat([exact, Buffer.from(" ")])); + const overflowBody = Buffer.concat([exact, Buffer.from(" ")]); + const overflow = await rawRequest(resource, "POST", { + declaredLength: overflowBody.byteLength, + }); assert.equal(overflow.status, 413); - assert.deepEqual(await overflow.json(), { detail: "Request too large" }); + assert.deepEqual(JSON.parse(overflow.body.toString()), { detail: "Request too large" }); + assert.equal(overflow.uploadEnded, false); + assert.equal(requests.length, overflowBefore); + const actualOverflow = await rawRequest(resource, "POST", { body: overflowBody }); + assert.equal(actualOverflow.status, 413); + assert.deepEqual(JSON.parse(actualOverflow.body.toString()), { detail: "Request too large" }); assert.equal(requests.length, overflowBefore); console.log( "Reader body limits: exact ordinary/avatar/placement byte boundaries, ASCII/UTF-8 declared and chunked overflow, dedicated placement validation passed.", ); } -async function rawRequest(resource, method, { body, headers: extraHeaders = {}, deadline } = {}) { +async function rawRequest( + resource, + method, + { body, headers: extraHeaders = {}, deadline, declaredLength } = {}, +) { let upload; let watchdog; try { @@ -388,7 +404,12 @@ async function rawRequest(resource, method, { body, headers: extraHeaders = {}, { method, agent: false, - headers: headers({ "Transfer-Encoding": "chunked", ...extraHeaders }), + headers: headers({ + ...(declaredLength === undefined + ? { "Transfer-Encoding": "chunked" } + : { "Content-Length": String(declaredLength) }), + ...extraHeaders, + }), }, (response) => { const chunks = []; @@ -411,6 +432,10 @@ async function rawRequest(resource, method, { body, headers: extraHeaders = {}, if (body !== undefined) { upload.write(body); upload.end(); + } else if (declaredLength !== undefined) { + // Require the declared-length rejection before any body is uploaded. + // A large concurrent write can race the intentional connection close. + upload.flushHeaders(); } else { // This first chunk is deliberately followed by no end() call. upload.write('{"incomplete-private-fixture":');