diff --git a/.github/workflows/publication-report-benchmark.yml b/.github/workflows/publication-report-benchmark.yml new file mode 100644 index 000000000..92ee5f4df --- /dev/null +++ b/.github/workflows/publication-report-benchmark.yml @@ -0,0 +1,92 @@ +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] + env: + DEVFEED_REDIS_URL: redis://127.0.0.1:6379/15 + 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: + 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: 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 + 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 --prepared \ + --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 + - name: Remove disposable PostgreSQL + if: always() + run: docker rm -fv publication-report-postgres >/dev/null 2>&1 || true diff --git a/apps/admin-api/src/automation.py b/apps/admin-api/src/automation.py index b70571a19..ee15423a7 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, @@ -614,13 +615,30 @@ 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, + # 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_bound, Article.published_to_feed_at <= now_bound + ) + .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( @@ -633,13 +651,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/apps/web/tests/browser/reader-gateway.mjs b/apps/web/tests/browser/reader-gateway.mjs index 6b894427a..2f5e2c257 100644 --- a/apps/web/tests/browser/reader-gateway.mjs +++ b/apps/web/tests/browser/reader-gateway.mjs @@ -364,6 +364,13 @@ async function checkExactByteLimits() { assert.deepEqual(JSON.parse(overflow.body.toString()), { detail: "Request too large" }); assert.equal(overflow.headers.connection, "close"); assert.equal(requests.length, overflowBefore); + const actualOverflow = await rawRequest(resource, "POST", { + body: Buffer.concat([exact, Buffer.from(" ")]), + }); + assert.equal(actualOverflow.status, 413); + assert.deepEqual(JSON.parse(actualOverflow.body.toString()), { detail: "Request too large" }); + assert.equal(actualOverflow.headers.connection, "close"); + 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.", ); @@ -433,6 +440,9 @@ async function rawRequest(resource, method, { body, headers: extraHeaders = {}, if (body !== undefined) { upload.write(body); upload.end(); + } else if (uploadHeaders.has("content-length")) { + // Exercise declared-size rejection before any body is uploaded. + upload.flushHeaders(); } else { // This first chunk is deliberately followed by no end() call. upload.write('{"incomplete-private-fixture":'); diff --git a/benchmarks/publication-reports.md b/benchmarks/publication-reports.md new file mode 100644 index 000000000..c9c123776 --- /dev/null +++ b/benchmarks/publication-reports.md @@ -0,0 +1,86 @@ +# 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:${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 +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 --prepared \ + --output reports/publication/300000.json +``` + +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 + +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 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. + +## 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. diff --git a/scripts/profile_publication_reports.py b/scripts/profile_publication_reports.py new file mode 100644 index 000000000..bfafc9b31 --- /dev/null +++ b/scripts/profile_publication_reports.py @@ -0,0 +1,469 @@ +"""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 under +the repository's reports/publication directory; --output must name a JSON file. +""" + +import argparse +import json +import os +import platform +import statistics +import tempfile +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, literal, text +from sqlalchemy.dialects.postgresql.psycopg import dialect +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/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/10)%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 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")) + 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 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 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%201=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, + 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 + JOIN articles a ON a.id=md5('publication-'||(1+n%:rows))::uuid""", + 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() + 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.ready.set() # Start reports only after a successful ordinary write. + 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, prepared=None): + ingestion = Ingestion(engine) if concurrent else None + started = time.perf_counter() + 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'")) + connection.execute(text("SET max_parallel_workers_per_gather=0")) + 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() + 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] + 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: + 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 prepared: + output["prepared_plan_counts"] = plan_counts + 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 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: + 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") + 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 // 201 * 24, + "fixture_analysis_job_rows": args.rows * 165 // 100, + "client_architecture": platform.machine(), + "postgresql": version, + "settings": settings, + "repeats": args.repeats, + "reports": [], + } + 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. + 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) + write_report(args.output, output) + print( + json.dumps( + { + "days": days, + "concurrent": concurrent, + "strategies": { + name: round(value["median_ms"], 2) + for name, value in report["strategies"].items() + }, + } + ), + flush=True, + ) + if args.prepared: + + def preserve(reports): + output["prepared_reports"] = reports + write_report(args.output, output) + + prepared_profiles(engine, args.repeats, args.rows, preserve) + 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 7b699b981..bb45c3ced 100644 --- a/tests/admin/operations/test_automation_policy_unit.py +++ b/tests/admin/operations/test_automation_policy_unit.py @@ -272,7 +272,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 @@ -288,6 +289,71 @@ 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"} + + +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 3b94b2e36..753982442 100644 --- a/tests/admin/reporting/projections/test_publication_report_projection.py +++ b/tests/admin/reporting/projections/test_publication_report_projection.py @@ -5,13 +5,290 @@ 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 +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_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} @@ -49,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] 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", 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