Reduce publication review probes across report sizes and prepared plans

.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

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):

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":');