diff --git a/.github/workflows/topic-report-profile.yml b/.github/workflows/topic-report-profile.yml new file mode 100644 index 000000000..d5d6d7259 --- /dev/null +++ b/.github/workflows/topic-report-profile.yml @@ -0,0 +1,407 @@ +name: Topic report history profile +on: + pull_request: + paths: + - .github/workflows/topic-report-profile.yml + - scripts/profile_topic_reports.py + - scripts/ci/topic-report-comment.test.mjs + - packages/core/src/topic_decision_metrics.py + - packages/core/src/inference_metrics.py + - migrations/versions/*_topic_decision_calls.py +permissions: + contents: read +concurrency: + group: topic-report-profile-${{ github.ref }} + cancel-in-progress: true +jobs: + profile: + name: PostgreSQL 17 / ${{ matrix.articles }} articles (ARM64) + runs-on: ubuntu-24.04-arm + timeout-minutes: 45 + strategy: + fail-fast: false + max-parallel: 2 + matrix: + articles: [100000, 300000] + env: + DEVFEED_REDIS_URL: redis://127.0.0.1:6379/15 + OTEL_SDK_DISABLED: "true" + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + path: candidate + ref: ${{ github.event.pull_request.head.sha }} + persist-credentials: false + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + path: baseline + ref: ${{ github.event.pull_request.base.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 + cache-dependency-glob: candidate/uv.lock + - name: Install profiling dependencies + run: sh candidate/scripts/ci/python-bootstrap.sh candidate --editable --all-packages + - name: Start the disposable database with random credentials + run: | + python3 - <<'PY' + import os + import secrets + import subprocess + import tempfile + import time + + password = secrets.token_hex(32) + database_url = ( + f"postgresql+psycopg://topic_profile:{password}" + "@127.0.0.1:5432/topic_profile_test" + ) + print(f"::add-mask::{password}", flush=True) + print(f"::add-mask::{database_url}", flush=True) + with open(os.environ["GITHUB_ENV"], "a", encoding="utf-8") as environment: + environment.write(f"DEVFEED_DATABASE_URL={database_url}\n") + environment.write(f"DEVFEED_TEST_DATABASE_URL={database_url}\n") + with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as secret: + secret.write("POSTGRES_USER=topic_profile\n") + secret.write(f"POSTGRES_PASSWORD={password}\n") + secret.write("POSTGRES_DB=topic_profile_test\n") + try: + subprocess.run( + [ + "docker", "run", "--detach", "--name", "topic-report-postgres", + "--publish", "127.0.0.1:5432:5432", "--env-file", secret.name, + "postgres:17.11@sha256:b8faeff7712663c2773a190181e1246e37e364b887611c04658c28b44e2408fe", + ], + check=True, + stdout=subprocess.DEVNULL, + ) + finally: + os.unlink(secret.name) + for _ in range(60): + ready = subprocess.run( + [ + "docker", "exec", "topic-report-postgres", "pg_isready", + "-h", "127.0.0.1", "-U", "topic_profile", "-d", "topic_profile_test", + ], + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + if ready.returncode == 0: + break + time.sleep(2) + else: + raise RuntimeError("Disposable PostgreSQL did not become ready within 120 seconds") + PY + - name: Migrate the disposable database + working-directory: candidate + run: uv run --locked --no-sync --no-build alembic upgrade head + - name: Verify native ARM64 and profile historical reports + working-directory: candidate + env: + PROFILE_ARTICLES: ${{ matrix.articles }} + run: | + test "$(uname -m)" = aarch64 + PROFILE_GATE_ARGS=() + if rg -q jsonb_array_elements ../baseline/packages/core/src/topic_decision_metrics.py; then + PROFILE_GATE_ARGS+=(--require-improvement) + fi + uv run --locked --no-sync --no-build python -m scripts.profile_topic_reports \ + --baseline-root ../baseline --seed --articles "$PROFILE_ARTICLES" \ + --require-postgres-major 17 \ + "${PROFILE_GATE_ARGS[@]}" \ + --output reports/topic-reports.json + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + if: always() + with: + name: topic-reports-${{ matrix.articles }}-arm64-${{ github.run_attempt }} + path: candidate/reports/ + if-no-files-found: error + retention-days: 30 + - name: Remove the disposable database + if: always() + run: docker rm --force topic-report-postgres + report-tests: + name: Test topic profile summary + runs-on: ubuntu-24.04 + timeout-minutes: 5 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha }} + persist-credentials: false + - uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 + with: + node-version: "26.12.1" + package-manager-cache: false + - run: node --test scripts/ci/topic-report-comment.test.mjs + report: + name: Publish topic profile summary + needs: [profile, report-tests] + if: always() + runs-on: ubuntu-24.04 + timeout-minutes: 5 + permissions: + contents: read + actions: read + pull-requests: write + concurrency: + group: topic-report-comment-${{ github.event.pull_request.number }} + cancel-in-progress: false + steps: + # No checkout: artifacts are parsed as data, never executed with write permissions. + - uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 + continue-on-error: true + with: + pattern: topic-reports-*-arm64-${{ github.run_attempt }} + path: report + merge-multiple: false + - uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0 + env: + PROFILE_RESULT: ${{ needs.profile.result }} + TEST_RESULT: ${{ needs.report-tests.result }} + with: + script: | + // BEGIN TOPIC REPORT COMMENT + const fs = require('node:fs'); + const marker = ''; + const pr = context.payload.pull_request; + const attempt = Number(process.env.GITHUB_RUN_ATTEMPT); + const url = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`; + const finite = value => typeof value === 'number' && Number.isFinite(value) && value >= 0; + const integer = value => Number.isSafeInteger(value) && value >= 0; + const ensure = (condition, message) => { if (!condition) throw new Error(message); }; + const fields = (value, names) => value && names.every(name => finite(value[name])); + const operations = ['append', 'settle', 'correction', 'checkpoint']; + const validDurable = condition => ensure(condition, 'Invalid committed-write evidence'); + const validateDurableStrategy = (row, repeats) => { + validDurable(fields(row, ['median_ms', 'p95_ms', 'elapsed_seconds', 'application_thread_cpu_seconds']) && + row.p95_ms >= row.median_ms && row.committed === true && + integer(row.iterations) && row.iterations > 0 && row.iterations % repeats === 0 && + integer(row.committed_transactions) && row.committed_transactions === row.iterations * operations.length && + ['wal_bytes', 'wal_records', 'wal_full_page_images'].every(key => integer(row[key]))); + for (const operation of operations) { + const sample = row.operations?.[operation]; + validDurable(fields(sample, ['median_ms', 'p95_ms']) && sample.p95_ms >= sample.median_ms && + sample.samples === row.iterations); + } + }; + const durableProfiles = data => { + if (!Object.hasOwn(data, 'durable_writes')) return null; + const profiles = data.durable_writes; + validDurable(Array.isArray(profiles) && profiles.length === 2 && + [5, 20].every(calls => profiles.filter(row => row.calls_per_workflow === calls).length === 1)); + for (const profile of profiles) { + validDurable(integer(profile.repeats) && profile.repeats > 0 && + Array.isArray(profile.paired_order) && profile.paired_order.length === profile.repeats && + profile.paired_order.every((order, index) => JSON.stringify(order) === + JSON.stringify(index % 2 === 0 ? ['legacy', 'facts'] : ['facts', 'legacy'])) && + profile.source_states_equal === true && profile.projection_verified === true); + for (const strategy of ['legacy', 'facts']) { + validateDurableStrategy(profile.strategies?.[strategy], profile.repeats); + } + validDurable(profile.strategies.legacy.iterations === profile.strategies.facts.iterations); + } + return profiles; + }; + const writeResult = profiles => { + if (!profiles) return 'UNVERIFIED'; + const higherLatency = profiles.some(profile => { + const before = profile.strategies.legacy, after = profile.strategies.facts; + return after.p95_ms > before.p95_ms || operations.some(operation => + after.operations[operation].median_ms > before.operations[operation].median_ms); + }); + return higherLatency ? 'HIGHER LATENCY OBSERVED' : 'NO HIGHER LATENCY IN THESE SAMPLES'; + }; + const read = articles => { + const nested = `report/topic-reports-${articles}-arm64-${attempt}/topic-reports.json`; + try { + ensure(Number.isSafeInteger(attempt) && attempt > 0, 'Invalid run attempt'); + // download-artifact can flatten a single match even with merge-multiple false. + const file = !fs.existsSync(nested) && fs.existsSync('report/topic-reports.json') ? + 'report/topic-reports.json' : nested; + ensure(fs.statSync(file).size <= 4 * 1024 * 1024, 'Oversized artifact'); + const raw = fs.readFileSync(file, 'utf8'); + const data = JSON.parse(raw); + ensure(data.complete === true, 'Incomplete artifact'); + if (file === 'report/topic-reports.json') { + ensure(data.fixture?.articles === articles, 'Missing artifact for this fixture'); + } + const serverVersion = data.server?.version?.match(/^PostgreSQL (17\.[0-9]+)\b/)?.[1]; + ensure(data.client_architecture === 'aarch64' && + /^17[0-9]{4}$/.test(String(data.server?.version_num)) && + serverVersion && data.server.version.includes(' on aarch64-') && + data.server?.default_work_mem === '4MB', 'Invalid native PostgreSQL 17 identity'); + const fixture = data.fixture; + ensure(fixture?.articles === articles && + ['runs', 'calls', 'inference_calls'].every(key => integer(fixture[key])) && + fixture.runs >= 5000 && fixture.calls >= fixture.runs && + fixture.inference_calls >= articles, 'Invalid fixture cardinalities'); + ensure(Array.isArray(data.reports) && data.reports.length === 4 && + [1, 7, 30, 90].every(days => data.reports.filter(row => + row.days === days && row.exact_output_parity === true).length === 1), + 'Missing exact report parity'); + for (const strategy of ['legacy', 'facts']) { + const row = data.concurrent?.[strategy]; + ensure(fields(row, ['median_report_pair_ms', 'application_process_cpu_seconds']) && + row.days === 90 && integer(row.report_workers) && row.report_workers > 0 && + integer(row.report_pairs) && row.report_pairs > 0 && + integer(row.writer_phase_overlap_pairs) && + row.writer_phase_overlap_pairs <= row.report_pairs, 'Invalid concurrent report metrics'); + ensure(fields(row.report_plan_totals, ['execution_ms', 'temp_written_blocks']), + 'Invalid topic query metrics'); + ensure(fields(row.ingestion, ['median_ms', 'p95_ms']) && + integer(row.ingestion.iterations) && row.ingestion.iterations > 0 && + row.ingestion.committed === false, 'Invalid ingestion metrics'); + const correction = data.correction_plans?.[strategy]; + ensure(fields(correction, ['Execution Time', 'statement_wal_bytes']) && + fields(correction.Plan, ['WAL FPI']), 'Invalid correction metrics'); + } + ensure(fields(data.backfill_equivalent_plan, ['Execution Time']), 'Invalid backfill metrics'); + const before = data.concurrent.legacy, after = data.concurrent.facts; + ensure(before.report_pairs === after.report_pairs && + before.report_workers === after.report_workers && + before.ingestion.iterations === after.ingestion.iterations, 'Unmatched workloads'); + const checks = { + report_runtime_lower: after.median_report_pair_ms < before.median_report_pair_ms, + topic_sql_runtime_lower: after.report_plan_totals.execution_ms < before.report_plan_totals.execution_ms, + topic_sql_zero_temp_writes: after.report_plan_totals.temp_written_blocks === 0, + all_reports_overlap_writer_phase: [before, after].every(row => + row.writer_phase_overlap_pairs === row.report_pairs), + }; + ensure(Object.entries(checks).every(([key, value]) => data.improvement?.[key] === value), + 'Inconsistent reported checks'); + return {articles, data, checks, serverVersion, durable: durableProfiles(data)}; + } catch (error) { + const known = ['Incomplete artifact', 'Oversized artifact', 'Missing artifact for this fixture', + 'Invalid native PostgreSQL 17 identity', 'Invalid fixture cardinalities', + 'Missing exact report parity', 'Invalid concurrent report metrics', + 'Invalid topic query metrics', 'Invalid ingestion metrics', + 'Invalid correction metrics', 'Invalid backfill metrics', + 'Unmatched workloads', 'Inconsistent reported checks', 'Invalid committed-write evidence']; + return {articles, error: known.includes(error.message) ? error.message : + 'Missing or invalid artifact for this run attempt'}; + } + }; + const reports = [100000, 300000].map(read); + const evidenceComplete = process.env.PROFILE_RESULT === 'success' && + process.env.TEST_RESULT === 'success' && reports.every(row => row.data); + const readResult = evidenceComplete ? + (reports.every(row => Object.values(row.checks).every(Boolean)) ? 'IMPROVED' : 'MIXED') : + 'UNVERIFIED'; + const durable = evidenceComplete && reports.every(row => row.durable) ? + reports.flatMap(row => row.durable) : null; + const number = (value, digits = 3) => value.toLocaleString('en-US', { + minimumFractionDigits: digits, maximumFractionDigits: digits}); + const delta = (before, after, unit) => { + const difference = after - before; + const sign = difference < 0 ? '-' : '+'; + const text = `${sign}${number(Math.abs(difference))} ${unit}`; + return difference > 0 ? `🔺 **${text}**` : text; + }; + const improvement = (before, after) => { + if (before === 0) return after === 0 ? '0.0%' : 'N/A (baseline zero)'; + const value = (before - after) / before * 100; + const sign = value > 0 ? '+' : value < 0 ? '-' : ''; + return `${sign}${number(Math.abs(value), 1)}%`; + }; + const lines = [marker, '### Topic report history', '', + `Read result: **${readResult}**. Write result: **${writeResult(durable)}**.`, '', + 'Native ARM64 PostgreSQL 17 · 90-day concurrent reports.', '', + 'Delta = after - before. Improvement = (before - after) / before × 100: positive means less time/work; negative means slower or higher cost.', '', + `Tested commit: \`${pr.head.sha}\` · [Workflow, artifacts and full plans](${url}) · attempt ${attempt}.`]; + if (process.env.PROFILE_RESULT !== 'success' || process.env.TEST_RESULT !== 'success') { + lines.push('', `Native matrix: **${process.env.PROFILE_RESULT}**; summary tests: **${process.env.TEST_RESULT}**. Any available measurements below do not establish a passing run.`); + } + for (const report of reports) { + lines.push('', `#### ${number(report.articles, 0)} articles`, ''); + if (report.error) { + lines.push(`**Unavailable: ${report.error}.** This is not a passing measurement.`); + continue; + } + const {data, checks} = report; + const before = data.concurrent.legacy, after = data.concurrent.facts; + lines.push(`PostgreSQL ${report.serverVersion} / ARM64 / work_mem 4MB. ` + + `${number(data.fixture.runs, 0)} workflows · ${number(data.fixture.calls, 0)} topic calls · ${number(data.fixture.inference_calls, 0)} inference rows.`, '', + '| Metric | Before | After | Delta (after - before) | Improvement |', + '| --- | ---: | ---: | ---: | ---: |'); + const row = (label, first, last, unit) => + lines.push(`| ${label} | ${number(first)} ${unit} | ${number(last)} ${unit} | ${delta(first, last, unit)} | ${improvement(first, last)} |`); + row('Report pair median', before.median_report_pair_ms, after.median_report_pair_ms, 'ms'); + row('Topic SQL replay total', before.report_plan_totals.execution_ms, after.report_plan_totals.execution_ms, 'ms'); + row('Topic SQL replay temp writes', before.report_plan_totals.temp_written_blocks / 128, after.report_plan_totals.temp_written_blocks / 128, 'MiB'); + row('Application CPU', before.application_process_cpu_seconds, after.application_process_cpu_seconds, 's'); + row('Rolled-back synthetic write median', before.ingestion.median_ms, after.ingestion.median_ms, 'ms'); + row('Rolled-back synthetic write p95', before.ingestion.p95_ms, after.ingestion.p95_ms, 'ms'); + const first = data.correction_plans.legacy, last = data.correction_plans.facts; + row('Rolled-back correction', first['Execution Time'], last['Execution Time'], 'ms'); + row('WAL snapshot (FPI-sensitive)', first.statement_wal_bytes / 1024, last.statement_wal_bytes / 1024, 'KiB'); + lines.push('', `Exact output parity: **1 / 7 / 30 / 90 days passed**. Observed checks: **${Object.values(checks).filter(Boolean).length}/4**. ` + + `${before.report_workers} report workers; ${before.report_pairs} report pairs total; ${before.ingestion.iterations} ingestion transactions per strategy. ` + + `Equivalent backfill: ${number(data.backfill_equivalent_plan['Execution Time'] / 1000)} s. ` + + `Correction full-page images: ${first.Plan['WAL FPI']} → ${last.Plan['WAL FPI']}.`, '', + 'The synthetic writes and correction above roll back and exclude commit latency; they cannot establish durable write throughput or overall production benefit.'); + if (!report.durable) { + lines.push('', '**Write result: UNVERIFIED.** Committed append, settlement, correction and checkpoint costs are absent from this artifact.'); + continue; + } + lines.push('', `**Write result: ${writeResult(report.durable)}.** Committed topic operations follow.`); + for (const profile of report.durable) { + const initial = profile.strategies.legacy, final = profile.strategies.facts; + lines.push('', `##### Committed topic writes: ${profile.calls_per_workflow} calls per workflow`, '', + `${profile.repeats} alternating repeats; ${number(initial.iterations, 0)} iterations and ` + + `${number(initial.committed_transactions, 0)} committed transactions per strategy. ` + + 'Source histories match and normalized facts were verified.', '', + '| Metric | Before | After | Delta (after - before) | Improvement |', + '| --- | ---: | ---: | ---: | ---: |'); + for (const operation of operations) { + row(`${operation[0].toUpperCase()}${operation.slice(1)} median (commit included)`, + initial.operations[operation].median_ms, final.operations[operation].median_ms, 'ms'); + } + row('All committed operations p95', initial.p95_ms, final.p95_ms, 'ms'); + row('Committed lifecycle phase', initial.elapsed_seconds, final.elapsed_seconds, 's'); + row('Application thread CPU', initial.application_thread_cpu_seconds, final.application_thread_cpu_seconds, 's'); + row('Committed writes WAL snapshot', initial.wal_bytes / 1024, final.wal_bytes / 1024, 'KiB'); + lines.push('', `WAL full-page images: ${number(initial.wal_full_page_images, 0)} → ` + + `${number(final.wal_full_page_images, 0)}. WAL is a server-wide snapshot sensitive to full-page images and background activity; it is not a stable amplification ratio.`); + } + } + lines.push('', 'Before = PR base; after = tested commit. 🔺 marks increased cost. ' + + 'Concurrent report diagnostics use paced writers and roll back; that phase includes SQL and sleeps, not continuous SQL overlap or durable-commit throughput. ' + + 'Committed lifecycle samples include COMMIT; phase elapsed time and application thread CPU compare fixed logical work, not pipeline capacity. ' + + 'Concurrent diagnostic CPU is application process CPU. Statement WAL uses a server-wide LSN delta including AFTER triggers/background activity; unequal full-page images prevent a stable WAL-amplification estimate. ' + + 'Topic SQL replay totals come from separate EXPLAIN executions after each timed report. Backfill excludes migration DDL and lock waits. ' + + 'Committed topic measurements include commit latency; they do not measure full ingestion-pipeline capacity or establish an overall production benefit.'); + const body = lines.join('\n'); + fs.mkdirSync('summary', {recursive: true}); + fs.writeFileSync('summary/topic-reports.md', body + '\n'); + await core.summary.addRaw(body).write(); + if (!evidenceComplete) core.setFailed('Topic benchmark evidence is incomplete or the native matrix failed'); + const sameRepository = pr.head.repo.full_name === `${context.repo.owner}/${context.repo.repo}`; + if (!sameRepository) return; + const current = await github.rest.pulls.get({...context.repo, pull_number: pr.number}); + if (current.data.head.sha !== pr.head.sha) return; + const comments = await github.paginate(github.rest.issues.listComments, + {...context.repo, issue_number: pr.number}); + const previous = comments.find(comment => comment.user?.type === 'Bot' && + comment.body?.startsWith(marker)); + const latest = await github.rest.pulls.get({...context.repo, pull_number: pr.number}); + if (latest.data.head.sha !== pr.head.sha) return; + if (previous) { + await github.rest.issues.updateComment({...context.repo, comment_id: previous.id, body}); + } else { + await github.rest.issues.createComment({...context.repo, issue_number: pr.number, body}); + } + // END TOPIC REPORT COMMENT + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + if: always() + with: + name: topic-report-summary-${{ github.run_attempt }} + path: summary/ + if-no-files-found: error + retention-days: 30 diff --git a/benchmarks/topic-report-history.md b/benchmarks/topic-report-history.md new file mode 100644 index 000000000..424fb6cc7 --- /dev/null +++ b/benchmarks/topic-report-history.md @@ -0,0 +1,138 @@ +# Topic report history profiling + +Issue [#299](https://github.com/abhi1693/devfeed.tech/issues/299) concerns repeatedly +expanding large decision state JSON and sorting repeated stages in administrative +reports. The disposable profiler compares the complete previous and current +decision metrics and inference chart responses on the same fixture. It alternates +strategy order and requires exact JSON equality for 1-, 7-, 30- and 90-day reports. + +`topic_decision_calls` stores one typed fact for each original `(proposal_id, +ordinal)`, including its lifetime occurrence number within its stage. Source run +JSON remains authoritative. A transactional trigger filters unchanged facts before +the upsert, avoiding conflict locks on unaffected calls, and deletes removed tail +entries; unrelated state updates skip projection work. Each +write handles that run's calls, with no global rollup or separate refresh scheduler. +Report time filters use the fact timestamp index, while completed proposal usage +joins all lifetime facts for proposals reviewed in the requested window. + +Migration `0036` follows the scheduler clock migration `0035`, installs the topic +projection, backfills existing runs and creates the indexes atomically. Its bounded +lock admission waits at most 30 seconds before a retryable failure. Once admitted, +backfill/index creation pauses proposal and run writes while allowing ordinary reads. +The current application requires schema `0036`; rollback application code together +with the schema. Downgrading to `0035` removes the derived topic facts while preserving +source JSON and the scheduler clock changes, allowing a later upgrade to rebuild the facts. +Downgrade briefly takes exclusive table locks, waiting for active readers and +writers before removing the trigger and projection; schedule it during a quiet +period. + +The production reference was collected read-only on 24 October 2026 at 09:29 UTC: +92,581 articles, 6,011 proposals, 5,000 decision runs and 14,373 calls. PostgreSQL +17.11 ran natively on ARM64 with `work_mem=4MB`. A bounded 100-run sample had mean +state JSON size 20,511 bytes, median 22,131 and p95 35,444 bytes. These cardinalities +describe that observation, rather than a continuously refreshed production count. + +`scripts/profile_topic_reports.py` uses migrated, empty, explicitly local +PostgreSQL 17+ databases ending in `_test`. It refuses libpq routing overrides and +never reads the checkout's `.env` or truncates fixture tables. The benchmark uses +ordinary article, inference and decision triggers during seeding. Each run has +approximately 20 KiB of incompressible unrelated evidence, plus one to five ordered +calls with repeated draft and verification stages spaced six hours apart and offset +by one hour from exact integer-day boundaries. Repeated stages span UTC dates and +the 30- and 90-day report boundaries. The first 6,000 runs cover the +recent 90 days; further runs add older unrelated history. Proposal review dates +vary independently, exercising whole-workflow accounting beyond the report window. + +The ARM64 workflow runs 100,000 and 300,000 article fixtures, with respectively +6,000 and 18,000 decision runs and 500,000 and 1,500,000 inference ledger rows. +It saves complete `EXPLAIN (ANALYZE, BUFFERS, WAL, FORMAT JSON)` plans, report wall +time, application process CPU, shared buffers and temporary blocks as artifacts. +Four concurrent workers run 90-day report pairs alongside 200 article insert and decision +correction transactions, rotating through one- to five-call workflows. Both +strategies execute the same operations, paced identically using the measured legacy +runtime to keep the writer phase active throughout reporting. The artifact records +`writer_phase_overlap` for every report. That phase includes SQL, rollback and pacing +sleeps; the event does not prove that every report overlaps an actual writer SQL +statement. Transaction latency excludes pacing. These writes +roll back so report inputs remain identical; ingestion latency measures transaction +work without durable commit/fsync cost. Concurrent process CPU is reported once for +the entire process; summing per-report CPU under concurrency would double count. + +The legacy concurrency phase briefly disables only the new decision projection +trigger on this dedicated fixture database, restoring it in `finally`. Committing +that short trigger-state change prevents an open `ALTER TABLE` transaction from +holding an exclusive lock during reporting. No workload writes commit. A separate +rolled-back five-call correction captures before/after WAL and trigger execution after a +warm-up sample avoids charging only the first strategy for checkpoint-dependent +full-page images. The root `EXPLAIN` WAL excludes AFTER-trigger work, so +`statement_wal_bytes` also records the server-wide LSN delta around the statement, +including trigger WAL. This dedicated-server measurement can include background +WAL and excludes the later transaction rollback. A full +equivalent backfill measures initial extraction and index insertion separately from +steady-state reporting. An equivalent backfill excludes migration DDL/lock wait. + +Separate committed-write comparisons run after the read diagnostics. Fresh cohorts +start with four or nineteen calls, then append a fifth or twentieth call, settle it, +correct an earlier call and save an unrelated checkpoint. These are four separate +transactions per workflow; the append also inserts an article. Latency ends after +COMMIT. Five calls cover the configured default; twenty cover the maximum initial +budget plus both operator grants. Each strategy uses the same logical source states +and 20 KiB of unrelated evidence. Repeated pairs alternate strategy order. Final +source-state hashes must match, and all current facts must match their source before +cleanup. Setup, trigger changes, verification and cleanup are outside timing. + +Committed samples report per-operation median and overall p95, transaction counts, +application thread CPU and WAL snapshots. They measure these lifecycle transactions, +not the full worker pipeline or maximum ingestion capacity. Each phase uses fresh +rows on the same dedicated database; cleanup does not restore identical heap/index +bloat. Server-wide WAL counters include background work and can contain unequal +full-page images. Retain those counts rather than interpreting one ratio as stable +write amplification. No universal acceptable writer-latency budget has been defined. + +To run locally, create and migrate a new disposable database and make a clean +checkout of the pre-change revision. Set `DEVFEED_DATABASE_URL` to that database +for migrations, and export the same URL as `DEVFEED_TEST_DATABASE_URL` for profiling: + +```sh +uv run --locked --no-sync alembic upgrade head +uv run --locked --no-sync python -m scripts.profile_topic_reports \ + --baseline-root /path/to/baseline --seed --articles 100000 \ + --output reports/topic-reports.json +``` + +Output must be a new `.json` artifact under the invocation's `reports/` directory. +Absolute paths inside that directory also work. Traversal components, symlink files +or directories, and existing destinations are rejected before database work. Choose +a new filename for each local run; artifact creation never overwrites an earlier report. + +The path-filtered `Topic report history profile` pull-request workflow uses a native +`ubuntu-24.04-arm` runner and an immutable PostgreSQL 17.11 image, explicitly requiring +major version 17 with `--require-postgres-major 17`. Profiling has read-only repository +permissions and no deployment or production credentials. Local AMD64 +smoke measurements validate the harness; they do not establish native ARM64 speed. +When the pull-request base still uses historical JSON expansion, the workflow +requires lower concurrent report and topic SQL runtime, zero topic SQL temporary +writes, and complete overlap between reports and the paced writer phase. Later changes compare +against their current base and retain exact output parity without requiring every +change to make the already normalized queries faster again. Full phase +elapsed time includes identical writer pacing and is not a report latency metric. +These are read checks, not an overall performance verdict. The summary reports +committed write costs separately and marks them unverified when absent; a passing +read check cannot establish acceptable long-term write overhead. + +Each run publishes separate 100,000- and 300,000-article before/after tables with +signed cost deltas and improvement percentages in its job summary and summary artifact. +Positive improvement means reduced cost; negative improvement means increased cost. +Same-repository pull requests +also receive one updated bot comment. The publishing job executes no checkout or +artifact code, validates the current attempt's JSON as data, and checks the current +PR head before writing. Forks retain the summary and artifact. Failed jobs and +missing or incomplete artifacts are reported explicitly instead of reusing a prior +attempt's passing measurement. + +After deployment, compare fresh `pg_stat_statements` deltas under matched report +windows and concurrency: calls, total execution time, temporary blocks per call and +buffer reads. Also check report freshness, worker latency and write amplification, +then verify a retried correction and deletion converge immediately without duplicate +facts. These deployment checks require a separately authorized release; the +benchmark workflow only accesses its disposable database. diff --git a/migrations/versions/0036_topic_decision_calls.py b/migrations/versions/0036_topic_decision_calls.py new file mode 100644 index 000000000..cc40041cb --- /dev/null +++ b/migrations/versions/0036_topic_decision_calls.py @@ -0,0 +1,134 @@ +"""Project topic call facts once, preserving lifetime stage and token accounting.""" + +from alembic import op + +revision = "0036" +down_revision = "0035" +branch_labels = None +depends_on = None + + +def _admit_locks(*, downgrading=False): + # These fixed modes cover every relation lock needed by the subsequent DDL. + # A failed attempt releases all partial locks before waiting, including when + # a writer flushes run -> proposal or a report reads facts -> run/proposal. + locks = ( + """LOCK TABLE topic_proposals IN ACCESS EXCLUSIVE MODE NOWAIT; + LOCK TABLE topic_decision_runs IN ACCESS EXCLUSIVE MODE NOWAIT; + LOCK TABLE topic_decision_calls IN ACCESS EXCLUSIVE MODE NOWAIT;""" + if downgrading + else """LOCK TABLE topic_proposals IN SHARE MODE NOWAIT; + LOCK TABLE topic_decision_runs IN EXCLUSIVE MODE NOWAIT;""" + ) + op.execute(f"""DO $body$ + DECLARE deadline timestamptz := clock_timestamp() + interval '30 seconds'; + BEGIN + LOOP + BEGIN + {locks} + EXIT; + EXCEPTION WHEN lock_not_available THEN + IF clock_timestamp() >= deadline THEN RAISE; END IF; + END; + PERFORM pg_sleep(0.05); + END LOOP; + END $body$""") + + +def upgrade(): + # Exclude run row lockers: backfill's foreign-key checks acquire row locks. + # Reads continue; source writes pause once both locks are held through commit. + _admit_locks() + op.execute("""CREATE TABLE topic_decision_calls ( + proposal_id uuid NOT NULL REFERENCES topic_decision_runs(proposal_id) ON DELETE CASCADE, + ordinal bigint NOT NULL, stage text, stage_call bigint NOT NULL, + started_at timestamptz, tokens numeric NOT NULL, + charged_tokens bigint, escalated boolean NOT NULL, + PRIMARY KEY (proposal_id, ordinal))""") + # One projection is shared by backfill and live writes. Scalar array entries + # still occupy their original ordinal and retain an unknown token contribution. + # Numeric addition avoids overflowing when individually valid bigint fields add up. + op.execute("""CREATE FUNCTION topic_decision_call_values(call_array jsonb) + RETURNS TABLE (ordinal bigint, stage text, stage_call bigint, started_at timestamptz, + tokens numeric, charged_tokens bigint, escalated boolean) + LANGUAGE sql STABLE SET TimeZone = 'UTC' AS $body$ + WITH calls AS ( + SELECT c, n AS ordinal FROM jsonb_array_elements( + CASE WHEN jsonb_typeof(call_array) = 'array' THEN call_array ELSE '[]'::jsonb END + ) WITH ORDINALITY AS items(c, n) + ), parsed AS ( + SELECT *, + CASE WHEN pg_input_is_valid(c#>>'{tokens,totalTokens}', 'bigint') + THEN (c#>>'{tokens,totalTokens}')::numeric ELSE 0 END AS total_tokens, + CASE WHEN pg_input_is_valid(c#>>'{tokens,inputTokens}', 'bigint') + THEN (c#>>'{tokens,inputTokens}')::numeric ELSE 0 END AS input_tokens, + CASE WHEN pg_input_is_valid(c#>>'{tokens,outputTokens}', 'bigint') + THEN (c#>>'{tokens,outputTokens}')::numeric ELSE 0 END AS output_tokens + FROM calls + ) + SELECT ordinal, c->>'stage', + row_number() OVER (PARTITION BY c->>'stage' ORDER BY ordinal), + -- Relative and infinite timestamps would move historical facts on a later + -- correction or rebuild. Only finite ISO dates/date-times are telemetry. + CASE WHEN c->>'started_at' ~ ( + '^[0-9]{4}-[0-9]{2}-[0-9]{2}([Tt ][0-9]{2}:[0-9]{2}' || + '(:[0-9]{2}([.][0-9]+)?)?([Zz]|[+-][0-9]{2}(:?[0-9]{2})?)?)?$') + AND pg_input_is_valid(c->>'started_at', 'timestamptz') + THEN (c->>'started_at')::timestamptz END, + greatest(total_tokens, input_tokens + output_tokens), + CASE WHEN pg_input_is_valid(c->>'charged_tokens', 'bigint') + THEN (c->>'charged_tokens')::bigint END, + CASE WHEN pg_input_is_valid(c->>'escalated', 'boolean') + THEN (c->>'escalated')::boolean ELSE false END + FROM parsed + $body$""") + op.execute("""INSERT INTO topic_decision_calls + SELECT r.proposal_id, calls.* FROM topic_decision_runs r + CROSS JOIN LATERAL topic_decision_call_values(r.state->'calls') calls""") + op.execute("""CREATE FUNCTION topic_decision_calls_sync() RETURNS trigger + LANGUAGE plpgsql AS $body$ + DECLARE call_count integer; + BEGIN + IF TG_OP = 'UPDATE' AND OLD.state->'calls' IS NOT DISTINCT FROM NEW.state->'calls' THEN + RETURN NULL; + END IF; + INSERT INTO topic_decision_calls AS target + SELECT NEW.proposal_id, calls.* FROM topic_decision_call_values(NEW.state->'calls') calls + LEFT JOIN topic_decision_calls existing + ON existing.proposal_id = NEW.proposal_id AND existing.ordinal = calls.ordinal + -- ON CONFLICT locks even unchanged rows before checking its WHERE clause. + -- The source run row serializes writers; skip those conflicts entirely. + WHERE existing.ordinal IS NULL OR + (existing.stage, existing.stage_call, existing.started_at, + existing.tokens, existing.charged_tokens, existing.escalated) + IS DISTINCT FROM (calls.stage, calls.stage_call, calls.started_at, + calls.tokens, calls.charged_tokens, calls.escalated) + ON CONFLICT (proposal_id, ordinal) DO UPDATE SET + (stage, stage_call, started_at, tokens, charged_tokens, escalated) = + (excluded.stage, excluded.stage_call, excluded.started_at, + excluded.tokens, excluded.charged_tokens, excluded.escalated) + WHERE (target.stage, target.stage_call, target.started_at, + target.tokens, target.charged_tokens, target.escalated) + IS DISTINCT FROM (excluded.stage, excluded.stage_call, excluded.started_at, + excluded.tokens, excluded.charged_tokens, excluded.escalated); + call_count := CASE WHEN jsonb_typeof(NEW.state->'calls') = 'array' + THEN jsonb_array_length(NEW.state->'calls') ELSE 0 END; + DELETE FROM topic_decision_calls + WHERE proposal_id = NEW.proposal_id AND ordinal > call_count; + RETURN NULL; + END $body$""") + op.execute("""CREATE TRIGGER topic_decision_calls_sync + AFTER INSERT OR UPDATE OF state ON topic_decision_runs + FOR EACH ROW EXECUTE FUNCTION topic_decision_calls_sync()""") + # The FK propagates deletion and TRUNCATE ... CASCADE without a second writer. + op.create_index("ix_topic_decision_calls_started_at", "topic_decision_calls", ["started_at"]) + op.create_index("ix_topic_proposals_reviewed_at", "topic_proposals", ["reviewed_at"]) + + +def downgrade(): + _admit_locks(downgrading=True) + op.drop_index("ix_topic_proposals_reviewed_at", table_name="topic_proposals") + op.execute("DROP TRIGGER topic_decision_calls_sync ON topic_decision_runs") + op.execute("DROP FUNCTION topic_decision_calls_sync()") + op.execute("DROP FUNCTION topic_decision_call_values(jsonb)") + op.drop_table("topic_decision_calls") diff --git a/packages/core/src/inference_metrics.py b/packages/core/src/inference_metrics.py index 68fd666c5..eb76d5923 100644 --- a/packages/core/src/inference_metrics.py +++ b/packages/core/src/inference_metrics.py @@ -161,8 +161,9 @@ def inference_charts(session, start: datetime, now: datetime) -> InferenceCharts else _ledger_activity(session, start, now) ) days = {} - day = start.date() - while day <= now.date(): + day = (start.replace(tzinfo=UTC) if start.tzinfo is None else start.astimezone(UTC)).date() + last_day = (now.replace(tzinfo=UTC) if now.tzinfo is None else now.astimezone(UTC)).date() + while day <= last_day: key = day.isoformat() days[key] = TopicDecisionDay(date=key) day += timedelta(days=1) @@ -179,17 +180,11 @@ def inference_charts(session, start: datetime, now: datetime) -> InferenceCharts days[row["date"]].approved, days[row["date"]].rejected = row["approved"], row["rejected"] calls = session.execute( text(""" - WITH calls AS ( - SELECT proposal_id, c, n, - row_number() OVER (PARTITION BY proposal_id, c->>'stage' ORDER BY n) AS stage_call - FROM topic_decision_runs, - jsonb_array_elements(state->'calls') WITH ORDINALITY AS items(c, n) - ) - SELECT to_char((c->>'started_at')::timestamptz AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS date, + SELECT to_char(started_at AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS date, count(*) AS calls, - count(*) FILTER (WHERE (c->>'escalated')::boolean) AS escalations, + count(*) FILTER (WHERE escalated) AS escalations, count(*) FILTER (WHERE stage_call > 1) AS repeated_stages - FROM calls WHERE (c->>'started_at')::timestamptz BETWEEN :start AND :now GROUP BY date + FROM topic_decision_calls WHERE started_at BETWEEN :start AND :now GROUP BY date """), {"start": start, "now": now}, ).mappings() diff --git a/packages/core/src/models.py b/packages/core/src/models.py index 7b7619ba8..447230227 100644 --- a/packages/core/src/models.py +++ b/packages/core/src/models.py @@ -286,7 +286,7 @@ class TopicProposal(Base): status: Mapped[str] = mapped_column(String(20), default="pending", index=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) created_by: Mapped[dict] = mapped_column(JSONB) - reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True) reviewed_by: Mapped[dict | None] = mapped_column(JSONB) review_note: Mapped[str | None] = mapped_column(String(1000)) applied: Mapped[dict | None] = mapped_column(JSONB) @@ -1542,6 +1542,22 @@ class TopicDecisionRun(Base): updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) +class TopicDecisionCall(Base): + """Indexed report facts, projected transactionally from each run's call ledger.""" + + __tablename__ = "topic_decision_calls" + proposal_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("topic_decision_runs.proposal_id", ondelete="CASCADE"), primary_key=True + ) + ordinal: Mapped[int] = mapped_column(BigInteger, primary_key=True) + stage: Mapped[str | None] = mapped_column(Text) + stage_call: Mapped[int] = mapped_column(BigInteger) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True) + tokens: Mapped[Decimal] = mapped_column(Numeric) + charged_tokens: Mapped[int | None] = mapped_column(BigInteger) + escalated: Mapped[bool] = mapped_column(Boolean) + + class DiscoverySeed(Base): __tablename__ = "discovery_seeds" id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4) diff --git a/packages/core/src/topic_decision_metrics.py b/packages/core/src/topic_decision_metrics.py index a2981e7fc..d49388c34 100644 --- a/packages/core/src/topic_decision_metrics.py +++ b/packages/core/src/topic_decision_metrics.py @@ -51,35 +51,26 @@ def decision_metrics(session, start: datetime, now: datetime) -> TopicDecisionMe result.decisions_per_hour = round((result.approved + result.rejected) / hours, 2) if result.decisions_per_hour: result.estimated_drain_hours = round(result.actionable / result.decisions_per_hour, 1) + # A completed topic's efficiency includes its entire workflow, even calls + # outside this reporting window. Only the activity counters use call time. usage = ( session.execute( text(""" - WITH calls AS ( - SELECT r.proposal_id, r.status, c, n, - (p.status IN ('approved', 'rejected') - AND p.reviewed_at BETWEEN :start AND :now) AS completed - FROM topic_decision_runs r JOIN topic_proposals p ON p.id = r.proposal_id, - jsonb_array_elements(r.state->'calls') WITH ORDINALITY AS items(c, n) - ), totals AS ( - SELECT *, greatest(coalesce((c->'tokens'->>'totalTokens')::bigint, 0), - coalesce((c->'tokens'->>'inputTokens')::bigint, 0) - + coalesce((c->'tokens'->>'outputTokens')::bigint, 0)) AS tokens, - (c->>'started_at')::timestamptz BETWEEN :start AND :now AS in_window, - row_number() OVER (PARTITION BY proposal_id, c->>'stage' ORDER BY n) AS stage_call - FROM calls - ) - SELECT count(*) FILTER (WHERE in_window) AS calls, - coalesce(sum(tokens) FILTER (WHERE in_window), 0) AS tokens, - coalesce(sum((c->>'charged_tokens')::bigint) - FILTER (WHERE in_window), 0) AS charged_tokens, - count(*) FILTER (WHERE in_window AND tokens = 0) AS unreported_calls, - count(*) FILTER (WHERE in_window AND (c->>'escalated')::boolean) AS escalations, - count(*) FILTER (WHERE in_window AND stage_call > 1) AS repeat_calls, - CASE WHEN bool_and(tokens > 0) FILTER (WHERE completed) - THEN sum(tokens) FILTER (WHERE completed)::float - / nullif(count(DISTINCT proposal_id) FILTER (WHERE completed), 0) + WITH window_usage AS ( + SELECT count(*) AS calls, coalesce(sum(tokens), 0) AS tokens, + coalesce(sum(charged_tokens), 0) AS charged_tokens, + count(*) FILTER (WHERE tokens = 0) AS unreported_calls, + count(*) FILTER (WHERE escalated) AS escalations, + count(*) FILTER (WHERE stage_call > 1) AS repeat_calls + FROM topic_decision_calls WHERE started_at BETWEEN :start AND :now + ), completed_usage AS ( + SELECT CASE WHEN bool_and(c.tokens > 0) + THEN sum(c.tokens)::float / nullif(count(DISTINCT c.proposal_id), 0) ELSE NULL END AS tokens_per_completed_topic - FROM totals + FROM topic_proposals p JOIN topic_decision_calls c ON c.proposal_id = p.id + WHERE p.status IN ('approved', 'rejected') AND p.reviewed_at BETWEEN :start AND :now + ) + SELECT * FROM window_usage CROSS JOIN completed_usage """), {"start": start, "now": now}, ) diff --git a/packages/core/src/version.py b/packages/core/src/version.py index 5b7bfbc16..d01da1a21 100644 --- a/packages/core/src/version.py +++ b/packages/core/src/version.py @@ -6,7 +6,8 @@ # Schema revisions are not application versions. Advance this when adding a # migration required by this application; never change an existing migration. -SCHEMA_REVISION = "0035" +SCHEMA_REVISION = "0036" -# Pipeline publication requires the transactional scheduler clock capture in 0035. +# Pipeline publication requires the scheduler clock capture in 0035, and topic +# decision reports require the indexed call facts introduced in 0036. BACKWARD_COMPATIBLE_SCHEMA_REVISIONS: frozenset[str] = frozenset() diff --git a/scripts/ci/fleet_contracts/README.md b/scripts/ci/fleet_contracts/README.md index 37b295f83..2c76e3713 100644 --- a/scripts/ci/fleet_contracts/README.md +++ b/scripts/ci/fleet_contracts/README.md @@ -22,7 +22,7 @@ production URLs are forwarded. Typesense uses the immutable official 30.2 index already pinned in `compose.yaml`, covering native AMD64 and ARM64 fixtures. The backend runs packaged Alembic to `0033`, seeds an article and account, executes -the exact migration hook, and verifies `0034` plus retained records. This includes +the exact migration hook, and verifies the required schema revision plus retained records. This includes the actual CLI, migration files and hook connection settings. The trimmed admin, user, partner and dedicated worker images deliberately use packaged `Base.metadata.create_all()` for their separate runtime fixtures. Their output diff --git a/scripts/ci/smoke_fleet_contracts.py b/scripts/ci/smoke_fleet_contracts.py index 254b4e99d..2f085e0ed 100644 --- a/scripts/ci/smoke_fleet_contracts.py +++ b/scripts/ci/smoke_fleet_contracts.py @@ -180,7 +180,7 @@ def schema(self): self.run(CONTRACTS["migrate"]["command"]) self.run(["python", "/fleet/runtime.py", "retained"]) print( - "PASS native packaged Alembic 0033 -> 0034 and Fleet migration hook; " + "PASS native packaged Alembic 0033 -> current schema and Fleet migration hook; " "retained article/account" ) else: diff --git a/scripts/ci/topic-report-comment.test.mjs b/scripts/ci/topic-report-comment.test.mjs new file mode 100644 index 000000000..8a9c86dfe --- /dev/null +++ b/scripts/ci/topic-report-comment.test.mjs @@ -0,0 +1,639 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { createRequire } from "node:module"; +import { posix as path } from "node:path"; +import { test } from "node:test"; + +const require = createRequire(import.meta.url); +const marker = ""; +const sha = "abcdef0123456789abcdef0123456789abcdef01"; +const attempts = [100000, 300000]; + +function artifact(articles) { + const legacy = { + median_report_pair_ms: 100, + application_process_cpu_seconds: 1, + report_plan_totals: { + execution_ms: 1000, + shared_hit_blocks: 100, + shared_read_blocks: 20, + temp_read_blocks: 12, + temp_written_blocks: 10, + }, + report_pairs: 12, + report_workers: 4, + days: 90, + writer_phase_overlap_pairs: 12, + ingestion: { median_ms: 4, p95_ms: 8, iterations: 200, committed: false }, + }; + const facts = structuredClone(legacy); + facts.median_report_pair_ms = 50; + facts.application_process_cpu_seconds = 0.75; + facts.report_plan_totals.execution_ms = 250; + facts.report_plan_totals.temp_read_blocks = 0; + facts.report_plan_totals.temp_written_blocks = 0; + facts.ingestion.median_ms = 5; + facts.ingestion.p95_ms = 10; + return { + complete: true, + client_architecture: "aarch64", + server: { + version: "PostgreSQL 17.11 on aarch64-unknown-linux-gnu", + version_num: "170011", + default_work_mem: "4MB", + }, + fixture: { + articles, + runs: articles * 0.06, + calls: articles * 0.18, + inference_calls: articles * 5, + }, + reports: [1, 7, 30, 90].map((days) => ({ + days, + exact_output_parity: true, + strategies: { + legacy: [{ elapsed_ms: 100 }, { elapsed_ms: 100 }, { elapsed_ms: 100 }], + facts: [{ elapsed_ms: 50 }, { elapsed_ms: 50 }, { elapsed_ms: 50 }], + }, + })), + concurrent: { legacy, facts }, + correction_plans: { + legacy: { statement_wal_bytes: 1000, "Execution Time": 0.5, Plan: { "WAL FPI": 0 } }, + facts: { statement_wal_bytes: 1500, "Execution Time": 0.75, Plan: { "WAL FPI": 0 } }, + }, + backfill_equivalent_plan: { "Execution Time": 2000 }, + improvement: { + report_runtime_lower: true, + topic_sql_runtime_lower: true, + topic_sql_zero_temp_writes: true, + all_reports_overlap_writer_phase: true, + }, + }; +} + +function committedProfiles({ slower = true } = {}) { + return [5, 20].map((calls) => { + const legacy = { + iterations: 600, + committed_transactions: 2400, + committed: true, + median_ms: 1, + p95_ms: 2, + elapsed_seconds: 3, + application_thread_cpu_seconds: 1, + wal_bytes: 1024, + wal_records: 20, + wal_full_page_images: 1, + operations: Object.fromEntries( + ["append", "settle", "correction", "checkpoint"].map((operation) => [ + operation, + { samples: 600, median_ms: 1, p95_ms: 2 }, + ]), + ), + }; + const facts = structuredClone(legacy); + facts.operations.append.median_ms = slower ? 1.5 : 0.5; + facts.elapsed_seconds = slower ? 4.5 : 2.5; + facts.application_thread_cpu_seconds = slower ? 1.5 : 0.75; + facts.wal_bytes = 2048; + facts.wal_full_page_images = 2; + return { + calls_per_workflow: calls, + repeats: 3, + paired_order: [ + ["legacy", "facts"], + ["facts", "legacy"], + ["legacy", "facts"], + ], + source_states_equal: true, + projection_verified: true, + strategies: { legacy, facts }, + }; + }); +} + +function changeArtifacts(fixture, change) { + for (const [file, raw] of fixture.files) { + const data = JSON.parse(raw); + change(data); + fixture.files.set(file, JSON.stringify(data)); + } +} + +function script() { + const workflow = readFileSync( + new URL("../../.github/workflows/topic-report-profile.yml", import.meta.url), + "utf8", + ); + const begin = "// BEGIN TOPIC REPORT COMMENT"; + const end = "// END TOPIC REPORT COMMENT"; + assert.equal( + workflow.split(begin).length, + 2, + "The actual workflow script needs one start marker", + ); + assert.equal(workflow.split(end).length, 2, "The actual workflow script needs one end marker"); + const code = workflow + .split(begin)[1] + .split(end)[0] + .split("\n") + .map((line) => line.replace(/^ {12}/, "")) + .join("\n"); + const AsyncFunction = Object.getPrototypeOf(async function () {}).constructor; + return new AsyncFunction("github", "context", "core", "require", "process", code); +} + +function harness({ attempt = "2", heads = [sha], fork = false, comments = [] } = {}) { + const files = new Map( + attempts.map((articles) => [ + `report/topic-reports-${articles}-arm64-${attempt}/topic-reports.json`, + JSON.stringify(artifact(articles)), + ]), + ); + const calls = { + reads: [], + writes: [], + gets: [], + lists: [], + creates: [], + updates: [], + failures: [], + }; + const summary = []; + const fs = { + existsSync: (file) => files.has(path.normalize(String(file))), + statSync(file) { + const target = path.normalize(String(file)); + if (!files.has(target)) + throw Object.assign(new Error(`Missing ${target}`), { code: "ENOENT" }); + return { size: Buffer.byteLength(files.get(target)) }; + }, + readFileSync(file, encoding) { + const target = path.normalize(String(file)); + calls.reads.push(target); + if (!files.has(target)) + throw Object.assign(new Error(`Missing ${target}`), { code: "ENOENT" }); + return encoding ? files.get(target) : Buffer.from(files.get(target)); + }, + mkdirSync() {}, + writeFileSync(file, value) { + const target = path.normalize(String(file)); + calls.writes.push({ file: target, value: String(value) }); + files.set(target, String(value)); + }, + }; + fs.promises = { + readFile: async (...args) => fs.readFileSync(...args), + mkdir: async (...args) => fs.mkdirSync(...args), + writeFile: async (...args) => fs.writeFileSync(...args), + }; + const github = { + rest: { + pulls: { + async get(parameters) { + const index = calls.gets.push(parameters) - 1; + return { + data: { head: { sha: heads[Math.min(index, heads.length - 1)] }, state: "open" }, + }; + }, + }, + issues: { + async listComments(parameters) { + calls.lists.push(parameters); + return { data: structuredClone(comments) }; + }, + async createComment(parameters) { + calls.creates.push(parameters); + const comment = { + id: 900 + comments.length, + user: { type: "Bot" }, + body: parameters.body, + }; + comments.push(comment); + return { data: comment }; + }, + async updateComment(parameters) { + calls.updates.push(parameters); + const comment = comments.find((item) => item.id === parameters.comment_id); + assert.ok(comment, "Only an existing comment may be updated"); + comment.body = parameters.body; + return { data: comment }; + }, + }, + }, + async paginate(method, parameters) { + return (await method(parameters)).data; + }, + }; + const core = { + info() {}, + warning() {}, + setFailed: (message) => calls.failures.push(message), + summary: { + addRaw(value) { + summary.push(value); + return this; + }, + async write() {}, + }, + }; + const context = { + repo: { owner: "example", repo: "devfeed.tech" }, + issue: { number: 304 }, + runId: 12345, + serverUrl: "https://github.com", + eventName: "pull_request", + sha, + payload: { + repository: { full_name: "example/devfeed.tech" }, + pull_request: { + number: 304, + state: "open", + base: { repo: { full_name: "example/devfeed.tech" } }, + head: { + sha, + repo: { full_name: fork ? "fork/devfeed.tech" : "example/devfeed.tech", fork }, + }, + }, + }, + }; + const process = { + env: { GITHUB_RUN_ATTEMPT: attempt, PROFILE_RESULT: "success", TEST_RESULT: "success" }, + }; + return { + files, + calls, + context, + process, + summary, + async run() { + await script()( + github, + context, + core, + (name) => (name.replace(/^node:/, "") === "fs" ? fs : require(name)), + process, + ); + return calls.creates.at(-1)?.body ?? calls.updates.at(-1)?.body ?? summary.join("\n"); + }, + }; +} + +// Expectations below exercise the exact inline script, including its GitHub API writes. +test("the summary separates after-minus-before deltas from positive savings and negative overhead", async () => { + const fixture = harness(); + const body = await fixture.run(); + assert.ok(body.startsWith(marker)); + assert.match(body, /100[,.]?000/); + assert.match(body, /300[,.]?000/); + assert.match(body, /Read result: \*\*IMPROVED\*\*/); + assert.match(body, /Write result: \*\*UNVERIFIED\*\*/); + assert.doesNotMatch(body, /\*\*PASS\b/); + assert.match(body, /\+50\.0%/); + assert.match(body, /Delta \(after - before\) \| Improvement/); + assert.match(body, /positive means less time\/work; negative means slower or higher cost/); + assert.match(body, /writer.phase|paced|pacing/i); + assert.match(body, /roll.?back|rolled.back|fsync/i); + assert.match(body, /Report pair median[^\n]+-50\.000 ms \| \+50\.0%/); + assert.match(body, /Rolled-back synthetic write median[^\n]+\+1\.000 ms[^\n]+-25\.0%/); + assert.match(body, /WAL snapshot[^\n]+\+0\.488 KiB[^\n]+-50\.0%/); + assert.match(body, /cannot establish durable write throughput or overall production benefit/); + assert.match(body, new RegExp(sha)); + assert.ok( + body.includes( + "[Workflow, artifacts and full plans](https://github.com/example/devfeed.tech/actions/runs/12345)", + ), + ); + assert.equal(fixture.calls.creates.length, 1); + assert.equal(fixture.calls.updates.length, 0); + assert.equal(fixture.calls.failures.length, 0); + assert.ok(fixture.calls.writes.some((entry) => entry.file === "summary/topic-reports.md")); +}); + +test("faster reads do not claim overall success when committed write costs are absent", async () => { + const fixture = harness(); + const body = await fixture.run(); + assert.match(body, /Read result: \*\*IMPROVED\*\*/); + assert.match(body, /Write result: \*\*UNVERIFIED\*\*/); + assert.match(body, /Committed append, settlement, correction and checkpoint costs are absent/); + assert.doesNotMatch(body, /\*\*PASS\b|overall benefit verified/i); +}); + +test("a valid read slowdown is disclosed instead of being called an improvement", async () => { + const fixture = harness(); + changeArtifacts(fixture, (data) => { + data.concurrent.facts.median_report_pair_ms = 125; + data.improvement.report_runtime_lower = false; + }); + const body = await fixture.run(); + assert.match(body, /Read result: \*\*MIXED\*\*/); + assert.match(body, /Report pair median[^\n]+\+25\.000 ms[^\n]+-25\.0%/); + assert.doesNotMatch(body, /\*\*PASS\b|Unavailable/); + assert.equal(fixture.calls.failures.length, 0); +}); + +test("zero baselines preserve absolute deltas without inventing a percentage", async () => { + const fixture = harness(); + changeArtifacts(fixture, (data) => { + data.concurrent.legacy.ingestion.median_ms = 0; + data.concurrent.legacy.ingestion.p95_ms = 0; + data.concurrent.facts.ingestion.p95_ms = 0; + }); + const body = await fixture.run(); + assert.match( + body, + /Rolled-back synthetic write median[^\n]+\+5\.000 ms[^\n]+N\/A \(baseline zero\)/, + ); + assert.match(body, /Rolled-back synthetic write p95[^\n]+\+0\.000 ms \| 0\.0%/); + assert.doesNotMatch(body, /NaN|Infinity/); +}); + +test("committed append overhead is visible even when every read check improves", async () => { + const fixture = harness(); + changeArtifacts(fixture, (data) => { + data.durable_writes = committedProfiles(); + }); + const body = await fixture.run(); + assert.match( + body, + /Read result: \*\*IMPROVED\*\*\. Write result: \*\*HIGHER LATENCY OBSERVED\*\*/, + ); + assert.match(body, /Committed topic writes: 5 calls per workflow/); + assert.match(body, /Committed topic writes: 20 calls per workflow/); + assert.match(body, /2,400 committed transactions per strategy/); + assert.match(body, /Append median \(commit included\)[^\n]+\+0\.500 ms[^\n]+-50\.0%/); + assert.match(body, /Committed writes WAL snapshot[^\n]+\+1\.000 KiB[^\n]+-100\.0%/); + assert.match(body, /WAL full-page images: 1 → 2/); + assert.match(body, /not a stable amplification ratio/); + assert.doesNotMatch(body, /\*\*PASS\b/); + assert.equal(fixture.calls.failures.length, 0); +}); + +test("lower sampled committed latency still makes no overall production claim", async () => { + const fixture = harness(); + changeArtifacts(fixture, (data) => { + data.durable_writes = committedProfiles({ slower: false }); + }); + const body = await fixture.run(); + assert.match(body, /Write result: \*\*NO HIGHER LATENCY IN THESE SAMPLES\*\*/); + assert.match(body, /Append median \(commit included\)[^\n]+-0\.500 ms \| \+50\.0%/); + assert.match( + body, + /do not measure full ingestion-pipeline capacity or establish an overall production benefit/, + ); + assert.doesNotMatch(body, /\*\*PASS\b/); +}); + +test("committed phase time and thread CPU overhead stay visible alongside rollback diagnostics", async () => { + const fixture = harness(); + changeArtifacts(fixture, (data) => { + data.durable_writes = committedProfiles(); + }); + const body = await fixture.run(); + assert.match( + body, + /Committed lifecycle phase \| 3\.000 s \| 4\.500 s[^\n]+\+1\.500 s[^\n]+-50\.0%/, + ); + assert.match(body, /Application thread CPU \| 1\.000 s \| 1\.500 s[^\n]+\+0\.500 s[^\n]+-50\.0%/); + assert.match(body, /Concurrent report diagnostics use paced writers and roll back/); + assert.match(body, /Committed lifecycle samples include COMMIT/); + assert.match( + body, + /phase elapsed time and application thread CPU compare fixed logical work, not pipeline capacity/, + ); + assert.doesNotMatch(body, /every transaction ends with rollback|\*\*PASS\b/); +}); + +test("one measured fixture cannot verify writes for a second fixture without committed evidence", async () => { + const fixture = harness(); + const target = [...fixture.files.keys()].find((file) => file.includes("100000")); + const data = JSON.parse(fixture.files.get(target)); + data.durable_writes = committedProfiles(); + fixture.files.set(target, JSON.stringify(data)); + const body = await fixture.run(); + assert.match(body, /Read result: \*\*IMPROVED\*\*\. Write result: \*\*UNVERIFIED\*\*/); + assert.match(body, /Committed topic writes: 20 calls per workflow/); + assert.match(body, /Committed append, settlement, correction and checkpoint costs are absent/); + assert.equal(fixture.calls.failures.length, 0); +}); + +for (const [label, change] of [ + [ + "missing long workflow", + (profiles) => { + profiles.pop(); + }, + ], + [ + "unequal source histories", + (profiles) => { + profiles[0].source_states_equal = false; + }, + ], + [ + "unverified projection", + (profiles) => { + profiles[0].projection_verified = false; + }, + ], + [ + "rolled-back transactions", + (profiles) => { + profiles[0].strategies.facts.committed = false; + }, + ], + [ + "missing commits", + (profiles) => { + profiles[0].strategies.facts.committed_transactions--; + }, + ], + [ + "missing operation", + (profiles) => { + delete profiles[0].strategies.facts.operations.append; + }, + ], + [ + "non-numeric WAL", + (profiles) => { + profiles[0].strategies.facts.wal_bytes = "2048"; + }, + ], + [ + "nonalternating order", + (profiles) => { + profiles[0].paired_order[1] = ["legacy", "facts"]; + }, + ], +]) { + test(`invalid committed-write evidence rejects ${label}`, async () => { + const fixture = harness(); + changeArtifacts(fixture, (data) => { + data.durable_writes = committedProfiles(); + change(data.durable_writes); + }); + const body = await fixture.run(); + assert.match(body, /Read result: \*\*UNVERIFIED\*\*/); + assert.match(body, /Invalid committed-write evidence/); + assert.equal(fixture.calls.failures.length, 1); + }); +} + +for (const [name, mutate, message] of [ + ["missing", (fixture, target) => fixture.files.delete(target), /missing|unavailable|absent/i], + [ + "malformed", + (fixture, target) => fixture.files.set(target, "{broken"), + /invalid|malformed|unavailable|parse/i, + ], + [ + "incomplete", + (fixture, target) => { + const result = JSON.parse(fixture.files.get(target)); + result.complete = false; + fixture.files.set(target, JSON.stringify(result)); + }, + /incomplete|unavailable|invalid/i, + ], +]) { + test(`the summary identifies a ${name} current-attempt artifact`, async () => { + const fixture = harness(); + const target = [...fixture.files.keys()].find((file) => file.includes("300000")); + mutate(fixture, target); + const body = await fixture.run(); + assert.match(body, message); + assert.match(body, /300[,.]?000/); + assert.equal(fixture.calls.creates.length, 1); + assert.equal(fixture.calls.failures.length, 1); + }); +} + +for (const [name, mutate] of [ + ["non-native identity", (result) => (result.client_architecture = "x86_64")], + [ + "PostgreSQL 18 identity", + (result) => (result.server.version = "PostgreSQL 18.0 on aarch64-unknown-linux-gnu"), + ], + [ + "AMD64 server identity", + (result) => (result.server.version = "PostgreSQL 17.11 on x86_64-pc-linux-gnu"), + ], + ["non-numeric metrics", (result) => (result.concurrent.facts.median_report_pair_ms = "50")], + ["missing parity", (result) => (result.reports[1].exact_output_parity = false)], + ["missing correction evidence", (result) => delete result.correction_plans.facts.Plan], + [ + "inconsistent improvement checks", + (result) => (result.improvement.report_runtime_lower = false), + ], +]) { + test(`invalid ${name} cannot produce a passing result`, async () => { + const fixture = harness(); + const target = [...fixture.files.keys()][0]; + const result = JSON.parse(fixture.files.get(target)); + mutate(result); + fixture.files.set(target, JSON.stringify(result)); + const body = await fixture.run(); + assert.match(body, /Read result: \*\*UNVERIFIED\*\*/); + assert.match(body, /Unavailable/); + assert.doesNotMatch(body, /NaN|Infinity/); + assert.equal(fixture.calls.failures.length, 1); + }); +} + +test("oversized artifacts are rejected before their contents are read", async () => { + const fixture = harness(); + const target = [...fixture.files.keys()][0]; + fixture.files.set(target, "x".repeat(4 * 1024 * 1024 + 1)); + const body = await fixture.run(); + assert.match(body, /Oversized artifact/); + assert.ok(!fixture.calls.reads.includes(target)); + assert.equal(fixture.calls.failures.length, 1); +}); + +for (const articles of attempts) { + test(`a single flat ${articles} artifact retains its results when the other matrix job fails`, async () => { + const fixture = harness(); + fixture.files.clear(); + fixture.files.set("report/topic-reports.json", JSON.stringify(artifact(articles))); + fixture.process.env.PROFILE_RESULT = "failure"; + const body = await fixture.run(); + const section = (count) => + body.split(`#### ${count.toLocaleString("en-US")} articles`)[1].split("####")[0]; + assert.match(body, /Read result: \*\*UNVERIFIED\*\*/); + assert.match(section(articles), /Report pair median/); + const missing = section(articles === 100000 ? 300000 : 100000); + assert.match(missing, /Unavailable/); + assert.doesNotMatch(missing, /Report pair median/); + assert.equal(fixture.calls.creates.length, 1); + assert.equal(fixture.calls.failures.length, 1); + }); +} + +test("an artifact from a previous attempt cannot replace a missing current result", async () => { + const fixture = harness({ attempt: "3" }); + const target = [...fixture.files.keys()].find((file) => file.includes("300000")); + fixture.files.delete(target); + fixture.files.set(target.replace("-3/", "-2/"), JSON.stringify(artifact(300000))); + const body = await fixture.run(); + assert.match(body, /missing|unavailable|absent/i); + assert.ok(fixture.calls.reads.every((file) => !file.includes("-arm64-2/"))); +}); + +test("a failed profile or test job is reported as failed even when artifacts are readable", async () => { + for (const variable of ["PROFILE_RESULT", "TEST_RESULT"]) { + const fixture = harness(); + fixture.process.env[variable] = "failure"; + const body = await fixture.run(); + assert.match(body, /fail/i); + assert.equal(fixture.calls.creates.length, 1); + } +}); + +test("an obsolete head is never published before listing comments", async () => { + const fixture = harness({ heads: ["newer-head"] }); + await fixture.run(); + assert.equal(fixture.calls.lists.length, 0); + assert.equal(fixture.calls.creates.length, 0); + assert.equal(fixture.calls.updates.length, 0); +}); + +test("an obsolete head is never published when it changes while comments are listed", async () => { + const fixture = harness({ heads: [sha, "newer-head"] }); + await fixture.run(); + assert.equal(fixture.calls.lists.length, 1); + assert.equal(fixture.calls.creates.length, 0); + assert.equal(fixture.calls.updates.length, 0); +}); + +test("fork pull requests receive a summary without attempted comment writes", async () => { + const fixture = harness({ fork: true }); + await fixture.run(); + assert.ok(fixture.calls.writes.some((entry) => entry.file === "summary/topic-reports.md")); + assert.ok(fixture.summary.length > 0); + assert.equal(fixture.calls.creates.length, 0); + assert.equal(fixture.calls.updates.length, 0); +}); + +test("the existing bot marker is updated while human comments remain untouched", async () => { + const fixture = harness({ + comments: [ + { id: 7, user: { type: "User" }, body: `${marker}\nA human note` }, + { id: 8, user: { type: "Bot" }, body: "Unrelated bot comment" }, + { id: 9, user: { type: "Bot" }, body: `${marker}\nPrior result` }, + ], + }); + await fixture.run(); + assert.equal(fixture.calls.creates.length, 0); + assert.equal(fixture.calls.updates.length, 1); + assert.equal(fixture.calls.updates[0].comment_id, 9); +}); + +test("rerunning the same result updates one marker comment instead of creating a duplicate", async () => { + const fixture = harness(); + await fixture.run(); + await fixture.run(); + assert.equal(fixture.calls.creates.length, 1); + assert.equal(fixture.calls.updates.length, 1); + assert.equal(fixture.calls.updates[0].comment_id, 900); +}); diff --git a/scripts/profile_topic_reports.py b/scripts/profile_topic_reports.py new file mode 100644 index 000000000..4f9d9ee77 --- /dev/null +++ b/scripts/profile_topic_reports.py @@ -0,0 +1,858 @@ +"""Compare historical topic reports on an explicitly disposable local PostgreSQL 17+. + +Run migrations first. --seed requires an empty database and never truncates data. +--baseline-root selects a clean checkout of the previous report implementation. +--require-postgres-major pins comparisons to a specific PostgreSQL major release. +--output creates a new JSON artifact below the current directory's reports folder. +Concurrent report diagnostics roll back their writes. Separate lifecycle samples +include COMMIT for five- and twenty-call workflows. Only the projection trigger is +temporarily disabled, in short committed transactions, for legacy comparisons. +""" + +import argparse +import hashlib +import importlib.util +import json +import os +import platform +import statistics +import sys +import threading +import time +import uuid +from concurrent.futures import ThreadPoolExecutor +from contextlib import ExitStack, suppress +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from devfeed_core import models +from sqlalchemy import create_engine, insert, text +from sqlalchemy.engine import make_url +from sqlalchemy.orm import Session + +from scripts.profile_read_models import identifier, seed_rows + +ROOT = Path(__file__).resolve().parents[1] +NOW = datetime(2026, 10, 24, 12, tzinfo=UTC) +CONCURRENT_DAYS = 90 +PRODUCTION = { + "observed_at": "2026-10-24T09:29:00Z", + "articles": 92581, + "topic_proposals": 6011, + "topic_decision_runs": 5000, + "topic_calls": 14373, + "sampled_state_bytes_mean": 20511, + "sampled_state_bytes_p95": 35444, + "sample_size": 100, +} + + +def validate_database(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(name) for name in ("PGHOSTADDR", "PGSERVICE", "PGSERVICEFILE")) + ): + raise ValueError( + "Use an explicitly local PostgreSQL *_test database without routing overrides" + ) + return url + + +def validate_server_version(value, *, required_major=None): + major = int(value) // 10000 + if major < 17: + raise ValueError("Topic report profiling requires PostgreSQL 17 or newer") + if required_major is not None and major != required_major: + raise ValueError( + f"This comparison requires PostgreSQL {required_major}; connected server is {major}" + ) + + +def report_output_path(destination): + root = Path.cwd().resolve() / "reports" + resolved = destination.resolve() + if ".." in destination.parts or not resolved.is_relative_to(root): + raise ValueError("Output must remain within the current directory's reports folder") + if resolved.suffix != ".json": + raise ValueError("Output must be a JSON artifact with a .json extension") + absolute = destination.absolute() + if any(parent.is_symlink() for parent in (absolute, *absolute.parents)): + raise ValueError("Output and its parent directories must not be symlinks") + if resolved.exists(): + raise ValueError("Output already exists; choose a new artifact filename") + return resolved + + +def write_profile_output(destination, data): + output = report_output_path(destination) + content = json.dumps(data, indent=2) + "\n" + directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW + # Anchor every child lookup to an opened directory. A parent replaced by a + # symlink after validation cannot redirect artifact creation outside reports. + with ExitStack() as descriptors: + directory = os.open(Path.cwd().resolve(), directory_flags) + descriptors.callback(os.close, directory) + for name in output.relative_to(Path.cwd().resolve()).parts[:-1]: + with suppress(FileExistsError): + os.mkdir(name, mode=0o700, dir_fd=directory) + directory = os.open(name, directory_flags, dir_fd=directory) + descriptors.callback(os.close, directory) + descriptor = os.open( + output.name, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, + dir_fd=directory, + ) + with os.fdopen(descriptor, "w", encoding="utf-8") as artifact: + artifact.write(content) + + +def report_functions(root, label): + functions = [] + for filename, function in ( + ("topic_decision_metrics.py", "decision_metrics"), + ("inference_metrics.py", "inference_charts"), + ): + path = root / "packages/core/src" / filename + name = f"topic_profile_{label}_{path.stem}" + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise ValueError(f"Cannot load report implementation {path}") + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + functions.append(getattr(module, function)) + return functions + + +class RecordingSession: + def __init__(self, session): + self.session = session + self.statements = [] + + def execute(self, statement, params=None): + sql = str(statement) + if "jsonb_array_elements" in sql or "FROM topic_decision_calls" in sql: + self.statements.append((sql, params or {})) + return self.session.execute(statement, params or {}) + + def scalar(self, statement): + return self.session.scalar(statement) + + +def plan_totals(plans): + totals = dict.fromkeys( + ( + "execution_ms", + "shared_hit_blocks", + "shared_read_blocks", + "temp_read_blocks", + "temp_written_blocks", + ), + 0, + ) + for plan in plans: + totals["execution_ms"] += plan["Execution Time"] + for output, key in ( + ("shared_hit_blocks", "Shared Hit Blocks"), + ("shared_read_blocks", "Shared Read Blocks"), + ("temp_read_blocks", "Temp Read Blocks"), + ("temp_written_blocks", "Temp Written Blocks"), + ): + totals[output] += plan["Plan"].get(key, 0) + return totals + + +def report_sample(engine, functions, days, *, explain=False): + with Session(engine) as session: + session.execute(text("SET TRANSACTION READ ONLY")) + session.execute(text("SET LOCAL statement_timeout='60s'")) + session.execute(text("SET LOCAL work_mem='4MB'")) + profile = RecordingSession(session) + started, cpu = time.perf_counter(), time.process_time() + output = [function(profile, NOW - timedelta(days=days), NOW) for function in functions] + elapsed_ms = (time.perf_counter() - started) * 1000 + cpu_ms = (time.process_time() - cpu) * 1000 + plans = [] + if explain: + for statement, params in profile.statements: + plan = session.execute( + text("EXPLAIN (ANALYZE, BUFFERS, WAL, FORMAT JSON) " + statement), params + ).scalar_one()[0] + plans.append(plan) + return { + "output": [value.model_dump(mode="json") for value in output], + "elapsed_ms": round(elapsed_ms, 3), + "application_cpu_ms": round(cpu_ms, 3), + "plan_totals": plan_totals(plans), + "plans": plans, + } + + +def seed(engine, articles, runs, history_bytes): + started = time.perf_counter() + with engine.begin() as connection: + if any( + connection.scalar(text(f"SELECT EXISTS(SELECT 1 FROM {table} LIMIT 1)")) + for table in ("articles", "topic_proposals", "topic_decision_runs", "inference_calls") + ): + raise ValueError( + "--seed requires empty fixture tables; create a new disposable database" + ) + seed_rows( + connection, + models.Article, + articles, + { + "canonical_url": "'https://topic-profile.invalid/articles/' || n", + "url_hash": "md5('topic-profile-article-' || n)", + "title": "'Topic report fixture ' || n", + "summary": "'Historical article fixture for concurrent ingestion measurements'", + }, + ) + proposal_id = identifier("topic-profile-proposal") + # A pool of distinct incompressible strings avoids a misleading tiny TOAST + # footprint from repeat('x', 20000), without millions of fixture hashes. + connection.execute( + text("""CREATE TEMP TABLE topic_profile_payloads ON COMMIT DROP AS + SELECT n AS payload_index, array_to_string(ARRAY(SELECT md5((n*10000+s)::text) + FROM generate_series(1,:parts) s),'') AS payload + FROM generate_series(0,63) n"""), + {"parts": (history_bytes + 31) // 32}, + ) + connection.execute( + text(f"""INSERT INTO topic_proposals + (id,batch_id,slug,action,origin,source_name,proposed,evidence,status, + created_at,created_by,reviewed_at,reviewed_by,applied) + SELECT {proposal_id},md5('topic-profile-batch')::uuid, + 'topic-profile-'||n,'create','import','Topic profiler','{{}}','[]', + CASE WHEN n%10=0 THEN 'pending' WHEN n%10=1 THEN 'rejected' ELSE 'approved' END, + :now - (n%365)*interval '1 day','{{}}', + CASE WHEN n%10<>0 THEN :now - (n%120)*interval '1 day' END, + CASE WHEN n%10<>0 THEN '{{}}'::jsonb END, + CASE WHEN n%10 NOT IN (0,1) THEN '{{}}'::jsonb END + FROM generate_series(1,:runs) n"""), + {"now": NOW, "runs": runs}, + ) + # First 6,000 runs remain fixed as unrelated older history grows. Repeated + # stage names span dates, so pushing the time filter before ordinality fails. + connection.execute( + text(f"""INSERT INTO topic_decision_runs + (proposal_id,input_hash,status,reason,state,created_at,updated_at) + SELECT {proposal_id},repeat('a',64), + CASE WHEN n%10=0 THEN 'deferred' ELSE 'decided' END, + CASE WHEN n%10=0 THEN 'profile_budget' END, + jsonb_build_object('evidence',payload,'calls', + (SELECT jsonb_agg(jsonb_build_object( + 'stage',(ARRAY['discovery','draft','verification','draft','verification'])[s], + 'started_at',to_char((:now - + (CASE WHEN n<=6000 THEN n%90 ELSE 180+n%185 END)*interval '1 day' + - (6-s)*interval '6 hours' - interval '1 hour') AT TIME ZONE 'UTC', + 'YYYY-MM-DD"T"HH24:MI:SS"Z"'), + 'status',CASE WHEN n%19=0 AND s=1 THEN 'failed' ELSE 'returned' END, + 'tokens',jsonb_build_object('inputTokens',1000+n%1000, + 'outputTokens',100+s,'totalTokens',1100+n%1000+s), + 'charged_tokens',1200+n%1000+s,'escalated',s>3) + ORDER BY s) FROM generate_series(1,1+n%5) s)), + :now - (n%365)*interval '1 day',:now - (n%120)*interval '1 day' + FROM generate_series(1,:runs) n + JOIN topic_profile_payloads p ON p.payload_index=n%64"""), + {"now": NOW, "runs": runs}, + ) + connection.execute( + text("""INSERT INTO inference_calls + (id,started_at,finished_at,operation,model,request_hash,status,tokens,web_searches,duration_ms) + SELECT md5('topic-profile-inference-'||n)::uuid, + :now-(n%365)*interval '1 day',:now,'topic_discovery','fixture',repeat('b',64), + 'returned',jsonb_build_object('inputTokens',1000,'outputTokens',100),0,100 + FROM generate_series(1,:rows) n"""), + {"now": NOW, "rows": articles * 5}, + ) + with engine.begin() as connection: + for table in ( + "articles", + "topic_proposals", + "topic_decision_runs", + "topic_decision_calls", + "inference_calls", + "inference_daily_usage", + ): + connection.execute(text(f"ANALYZE {table}")) + return {"elapsed_seconds": round(time.perf_counter() - started, 3)} + + +def set_projection(engine, enabled): + with engine.begin() as connection: + connection.execute(text("SET LOCAL lock_timeout='5s'")) + connection.execute( + text( + "ALTER TABLE topic_decision_runs " + + ("ENABLE" if enabled else "DISABLE") + + " TRIGGER topic_decision_calls_sync" + ) + ) + + +def maintenance_profile(engine): + """Measure a full equivalent backfill, restoring facts with transaction rollback.""" + with engine.connect() as connection: + transaction = connection.begin() + try: + connection.execute(text("TRUNCATE topic_decision_calls")) + plan = connection.execute( + text("""EXPLAIN (ANALYZE, BUFFERS, WAL, FORMAT JSON) + INSERT INTO topic_decision_calls + SELECT r.proposal_id,c.* FROM topic_decision_runs r + CROSS JOIN LATERAL topic_decision_call_values(r.state->'calls') c""") + ) + return plan.scalar_one()[0] + finally: + transaction.rollback() + + +def _correction_plan(engine, legacy): + with engine.connect() as connection: + transaction = connection.begin() + try: + proposal = connection.scalar( + text("""SELECT proposal_id FROM topic_decision_calls + GROUP BY proposal_id ORDER BY count(*) DESC,proposal_id LIMIT 1""") + ) + if legacy: + connection.execute( + text("""ALTER TABLE topic_decision_runs + DISABLE TRIGGER topic_decision_calls_sync""") + ) + before = connection.scalar(text("SELECT pg_current_wal_insert_lsn()")) + plan = connection.execute( + text("""EXPLAIN (ANALYZE, BUFFERS, WAL, FORMAT JSON) + UPDATE topic_decision_runs SET state=jsonb_set(state, + '{calls,0,charged_tokens}','50000'::jsonb) + WHERE proposal_id=:proposal"""), + {"proposal": proposal}, + ).scalar_one()[0] + plan["statement_wal_bytes"] = int( + connection.scalar( + text("SELECT pg_wal_lsn_diff(pg_current_wal_insert_lsn(),:before)"), + {"before": before}, + ) + ) + plan["statement_wal_scope"] = ( + "Server-wide LSN delta including AFTER triggers and any background WAL; " + "excludes later transaction rollback" + ) + return plan + finally: + transaction.rollback() + + +def correction_plan(engine, legacy): + # Prime the same pages first: otherwise the first strategy pays checkpoint- + # dependent full-page images and makes WAL amplification appear misleading. + _correction_plan(engine, legacy) + return _correction_plan(engine, legacy) + + +def writer(engine, iterations, barrier=None, active=None, interval=0, runs=50): + samples = [] + if active is not None: + active.set() + if barrier is not None: + barrier.wait() + cpu, started = time.thread_time(), time.perf_counter() + for index in range(iterations): + number = 1 + index % min(runs, 50) + proposal = uuid.UUID( + hashlib.sha256( + f"read-model-profile:topic-profile-proposal:{number}".encode() + ).hexdigest()[:32] + ) + with engine.connect() as connection: + transaction = connection.begin() + sample_started = time.perf_counter() + try: + value = uuid.uuid4() + connection.execute( + insert(models.Article).values( + id=value, + canonical_url=f"https://topic-profile.invalid/ingest/{value}", + url_hash=hashlib.sha256(value.bytes).hexdigest(), + title="Concurrent ingestion", + ) + ) + connection.execute( + text("""UPDATE topic_decision_runs SET state=jsonb_set(state, + '{calls,0,charged_tokens}',to_jsonb(CAST(:tokens AS bigint))),updated_at=:now + WHERE proposal_id=:proposal"""), + { + "tokens": 50000 + index, + "now": NOW, + "proposal": proposal, + }, + ) + samples.append((time.perf_counter() - sample_started) * 1000) + finally: + transaction.rollback() + if interval: + time.sleep(interval) + if active is not None: + active.clear() + return { + "iterations": iterations, + "elapsed_seconds": round(time.perf_counter() - started, 3), + "application_thread_cpu_seconds": round(time.thread_time() - cpu, 3), + "median_ms": round(statistics.median(samples), 3), + "p95_ms": round(sorted(samples)[min(len(samples) - 1, int(len(samples) * 0.95))], 3), + "committed": False, + } + + +def concurrent(engine, functions, workers, repeats, writes, interval, runs): + barrier = threading.Barrier(workers + 1) + active = threading.Event() + + def reporter(): + barrier.wait() + samples = [] + for _ in range(repeats): + during_start = active.is_set() + sample = report_sample(engine, functions, CONCURRENT_DAYS, explain=True) + sample["writer_phase_overlap"] = during_start and active.is_set() + samples.append(sample) + return samples + + started, cpu = time.perf_counter(), time.process_time() + with ThreadPoolExecutor(max_workers=workers + 1) as executor: + reports = [executor.submit(reporter) for _ in range(workers)] + ingestion = executor.submit(writer, engine, writes, barrier, active, interval, runs) + samples = [sample for future in reports for sample in future.result()] + ingestion_result = ingestion.result() + return { + "elapsed_seconds": round(time.perf_counter() - started, 3), + "application_process_cpu_seconds": round(time.process_time() - cpu, 3), + "report_workers": workers, + "days": CONCURRENT_DAYS, + "report_pairs": len(samples), + "writer_interval_seconds": round(interval, 6), + "writer_phase_overlap_pairs": sum(s["writer_phase_overlap"] for s in samples), + "median_report_pair_ms": round(statistics.median(s["elapsed_ms"] for s in samples), 3), + "report_plan_totals": plan_totals([plan for sample in samples for plan in sample["plans"]]), + "ingestion": ingestion_result, + "plans": [sample["plans"] for sample in samples], + } + + +def latency_summary(samples): + return { + "samples": len(samples), + "median_ms": round(statistics.median(samples), 3), + "p95_ms": round(sorted(samples)[min(len(samples) - 1, int(len(samples) * 0.95))], 3), + } + + +def _wal_snapshot(connection): + # Flush this backend's pending counters before reading a new statistics snapshot. + connection.execute(text("SELECT pg_stat_force_next_flush()")) + connection.commit() + connection.execute(text("SELECT pg_stat_clear_snapshot()")) + row = dict( + connection.execute( + text("SELECT pg_current_wal_insert_lsn() AS lsn, wal_records, wal_fpi FROM pg_stat_wal") + ) + .mappings() + .one() + ) + connection.commit() + return row + + +def _durable_write_phase(engine, iterations, call_count, *, legacy): + """Measure separate lifecycle commits on fresh, logically identical workflows.""" + proposals = [uuid.uuid4() for _ in range(iterations)] + articles = [uuid.uuid4() for _ in range(iterations)] + calls = [ + { + "stage": "draft" if index % 2 else "verification", + "started_at": (NOW - timedelta(hours=index + 1)).isoformat(), + "status": "returned", + "tokens": {"totalTokens": 1100}, + "charged_tokens": 1100, + "escalated": False, + } + for index in range(call_count - 1) + ] + evidence = "".join(hashlib.sha256(str(index).encode()).hexdigest() for index in range(313)) + state = {"calls": calls, "evidence": evidence} + operations = {name: [] for name in ("append", "settle", "correction", "checkpoint")} + try: + with engine.begin() as connection: + connection.execute( + insert(models.TopicProposal), + [ + { + "id": value, + "batch_id": value, + "slug": f"durable-{value}", + "action": "create", + "origin": "import", + "source_name": "Durable profiler", + "proposed": {}, + "evidence": [], + "status": "pending", + "created_by": {}, + } + for value in proposals + ], + ) + connection.execute( + insert(models.TopicDecisionRun), + [ + {"proposal_id": value, "input_hash": "d" * 64, "state": state} + for value in proposals + ], + ) + set_projection(engine, not legacy) + with engine.connect() as connection: + before = _wal_snapshot(connection) + started, cpu = time.perf_counter(), time.thread_time() + for index, proposal in enumerate(proposals): + pending = { + "stage": "verification", + "started_at": NOW.isoformat(), + "status": "running", + "charged_tokens": 40000, + "escalated": False, + } + settled = { + **pending, + "status": "returned", + "tokens": {"totalTokens": 1200}, + "charged_tokens": 1200, + } + updates = { + "append": ( + "jsonb_set(state,'{calls}',state->'calls'||CAST(:value AS jsonb))", + json.dumps([pending]), + ), + "settle": ( + f"jsonb_set(state,'{{calls,{call_count - 1}}}',CAST(:value AS jsonb))", + json.dumps(settled), + ), + "correction": ( + "jsonb_set(state,'{calls,0,charged_tokens}',CAST(:value AS jsonb))", + str(50000 + index), + ), + "checkpoint": ( + "jsonb_set(state,'{checkpoint}',CAST(:value AS jsonb))", + '"complete"', + ), + } + for name, (expression, value) in updates.items(): + transaction = connection.begin() + sample_started = time.perf_counter() + try: + if name == "append": + article = articles[index] + connection.execute( + insert(models.Article).values( + id=article, + canonical_url=f"https://durable-profile.invalid/{article}", + url_hash=hashlib.sha256(article.bytes).hexdigest(), + title="Committed ingestion", + ) + ) + connection.execute( + text( + f"UPDATE topic_decision_runs SET state={expression}," + "updated_at=:now " + "WHERE proposal_id=:proposal" + ), + {"value": value, "now": NOW, "proposal": proposal}, + ) + transaction.commit() + operations[name].append((time.perf_counter() - sample_started) * 1000) + except BaseException: + transaction.rollback() + raise + elapsed, process_cpu = time.perf_counter() - started, time.thread_time() - cpu + after = _wal_snapshot(connection) + wal_bytes = int( + connection.scalar( + text("SELECT pg_wal_lsn_diff(:after,:before)"), + {"after": after["lsn"], "before": before["lsn"]}, + ) + ) + states = connection.scalars( + text("SELECT state FROM topic_decision_runs WHERE proposal_id=ANY(:ids)"), + {"ids": proposals}, + ).all() + if not legacy: + # Compare both directions: missing, additional and stale facts all fail. + expected = """SELECT r.proposal_id,c.* FROM topic_decision_runs r + CROSS JOIN LATERAL topic_decision_call_values(r.state->'calls') c + WHERE r.proposal_id=ANY(:ids)""" + actual = "SELECT * FROM topic_decision_calls WHERE proposal_id=ANY(:ids)" + if connection.scalar( + text( + f"SELECT EXISTS(({actual} EXCEPT {expected}) UNION ALL " + f"({expected} EXCEPT {actual}))" + ), + {"ids": proposals}, + ): + raise AssertionError("Committed workflow facts differ from source") + digest = hashlib.sha256( + json.dumps(sorted(json.dumps(value, sort_keys=True) for value in states)).encode() + ).hexdigest() + return { + "operations": operations, + "elapsed_seconds": elapsed, + "application_thread_cpu_seconds": process_cpu, + "wal_bytes": wal_bytes, + "wal_records": int(after["wal_records"] - before["wal_records"]), + "wal_full_page_images": int(after["wal_fpi"] - before["wal_fpi"]), + "source_sha256": digest, + } + finally: + set_projection(engine, True) + with engine.begin() as connection: + connection.execute(text("DELETE FROM articles WHERE id=ANY(:ids)"), {"ids": articles}) + connection.execute( + text("DELETE FROM topic_proposals WHERE id=ANY(:ids)"), {"ids": proposals} + ) + # Article deletion itself emits tombstones. Remove only this phase's + # outbox rows after deletion so later pairs retain the same queue load. + for table, kind in ( + ("partner_match_jobs", "article"), + ("read_model_events", "article"), + ("search_events", "articles"), + ): + connection.execute( + text(f"DELETE FROM {table} WHERE kind=:kind AND entity_id=ANY(:ids)"), + {"kind": kind, "ids": articles}, + ) + + +def durable_write_profile(engine, iterations=200, call_count=5, repeats=3): + if not (1 <= iterations <= 1000 and 2 <= call_count <= 20 and 1 <= repeats <= 10): + raise ValueError("Committed write profiling exceeds supported bounds") + phases = {name: [] for name in ("legacy", "facts")} + orders = [] + for repeat in range(repeats): + order = ["legacy", "facts"] if repeat % 2 == 0 else ["facts", "legacy"] + orders.append(order) + for name in order: + phases[name].append( + _durable_write_phase(engine, iterations, call_count, legacy=name == "legacy") + ) + if len({phase["source_sha256"] for values in phases.values() for phase in values}) != 1: + raise AssertionError("Committed comparisons ended with different source states") + strategies = {} + for name, values in phases.items(): + operations = { + operation: [sample for value in values for sample in value["operations"][operation]] + for operation in values[0]["operations"] + } + strategies[name] = { + **latency_summary([sample for samples in operations.values() for sample in samples]), + "iterations": iterations * repeats, + "committed": True, + "committed_transactions": iterations * repeats * 4, + "operations": {key: latency_summary(samples) for key, samples in operations.items()}, + **{ + key: sum(value[key] for value in values) + for key in ( + "elapsed_seconds", + "application_thread_cpu_seconds", + "wal_bytes", + "wal_records", + "wal_full_page_images", + ) + }, + } + return { + "calls_per_workflow": call_count, + "repeats": repeats, + "paired_order": orders, + "source_states_equal": True, + "projection_verified": True, + "strategies": strategies, + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--baseline-root", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--seed", action="store_true") + parser.add_argument("--articles", type=int, default=100000) + parser.add_argument("--runs", type=int) + parser.add_argument("--history-bytes", type=int, default=20000) + parser.add_argument("--repeats", type=int, default=3) + parser.add_argument("--workers", type=int, default=4) + parser.add_argument("--writes", type=int, default=200) + parser.add_argument("--require-improvement", action="store_true") + parser.add_argument("--require-postgres-major", type=int) + args = parser.parse_args() + try: + url = validate_database(os.environ["DEVFEED_TEST_DATABASE_URL"]) + except (KeyError, ValueError) as error: + parser.error(str(error)) + if args.require_postgres_major is not None and args.require_postgres_major < 17: + parser.error("--require-postgres-major must be 17 or newer") + runs = args.runs if args.runs is not None else max(6000, args.articles * 6 // 100) + if not ( + 10 <= args.articles <= 1000000 + and 10 <= runs <= 100000 + and 0 <= args.history_bytes <= 40000 + and 1 <= args.repeats <= 10 + and 1 <= args.workers <= 8 + and 1 <= args.writes <= 1000 + ): + parser.error("Fixture and concurrency sizes exceed supported profiling bounds") + try: + args.output = report_output_path(args.output) + except (OSError, RuntimeError, ValueError) as error: + parser.error(str(error)) + functions = { + "legacy": report_functions(args.baseline_root.resolve(), "legacy"), + "facts": report_functions(ROOT, "facts"), + } + engine = create_engine(url, pool_size=12, max_overflow=0) + output = { + "production_reference": PRODUCTION, + "client_architecture": platform.machine(), + "python_version": platform.python_version(), + "harness_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), + "report_source_sha256": { + name: { + Path(sys.modules[function.__module__].__file__).name: hashlib.sha256( + Path(sys.modules[function.__module__].__file__).read_bytes() + ).hexdigest() + for function in implementations + } + for name, implementations in functions.items() + }, + } + try: + with engine.connect() as connection: + output["server"] = dict( + connection.execute( + text("""SELECT version() AS version, + current_setting('work_mem') AS default_work_mem, + current_setting('server_version_num') AS version_num""") + ) + .mappings() + .one() + ) + try: + validate_server_version( + output["server"]["version_num"], required_major=args.require_postgres_major + ) + except ValueError as error: + parser.error(str(error)) + if args.seed: + output["seed"] = seed(engine, args.articles, runs, args.history_bytes) + with engine.connect() as connection: + output["fixture"] = dict( + connection.execute( + text("""SELECT + (SELECT count(*) FROM articles) AS articles, + (SELECT count(*) FROM topic_decision_runs) AS runs, + (SELECT count(*) FROM topic_decision_calls) AS calls, + (SELECT count(*) FROM inference_calls) AS inference_calls""") + ) + .mappings() + .one() + ) + output["reports"] = [] + for days in (1, 7, 30, 90): + samples = {name: [] for name in functions} + expected = None + for repeat in range(args.repeats): + for name in ("legacy", "facts") if repeat % 2 == 0 else ("facts", "legacy"): + sample = report_sample(engine, functions[name], days, explain=repeat == 0) + actual = sample.pop("output") + if expected is not None and actual != expected: + raise AssertionError(f"{days}-day report output differs for {name}") + expected = actual + samples[name].append(sample) + output["reports"].append( + { + "days": days, + "exact_output_parity": True, + "output_sha256": hashlib.sha256( + json.dumps(expected, sort_keys=True).encode() + ).hexdigest(), + "strategies": samples, + } + ) + output["backfill_equivalent_plan"] = maintenance_profile(engine) + output["correction_plans"] = { + name: correction_plan(engine, name == "legacy") for name in functions + } + output["concurrent"] = {} + # Pace identical fixed ingestion workloads long enough to cover all report + # repetitions. Pacing waits are excluded from each transaction's latency. + comparison_window = next(row for row in output["reports"] if row["days"] == CONCURRENT_DAYS) + legacy_ms = statistics.median( + sample["elapsed_ms"] for sample in comparison_window["strategies"]["legacy"] + ) + interval = min( + 0.25, max(0.002, legacy_ms / 1000 * args.repeats * args.workers * 4 / args.writes) + ) + for name in ("legacy", "facts"): + set_projection(engine, name == "facts") + try: + output["concurrent"][name] = concurrent( + engine, + functions[name], + args.workers, + args.repeats, + args.writes, + interval, + output["fixture"]["runs"], + ) + finally: + set_projection(engine, True) + legacy, facts = (output["concurrent"][name] for name in ("legacy", "facts")) + output["improvement"] = { + "report_runtime_lower": facts["median_report_pair_ms"] + < legacy["median_report_pair_ms"], + "topic_sql_runtime_lower": facts["report_plan_totals"]["execution_ms"] + < legacy["report_plan_totals"]["execution_ms"], + "topic_sql_zero_temp_writes": facts["report_plan_totals"]["temp_written_blocks"] == 0, + "all_reports_overlap_writer_phase": all( + row["writer_phase_overlap_pairs"] == row["report_pairs"] + for row in output["concurrent"].values() + ), + } + output["durable_writes"] = [ + durable_write_profile(engine, args.writes, calls, args.repeats) for calls in (5, 20) + ] + output["complete"] = True + write_profile_output(args.output, output) + print( + json.dumps( + { + "output": str(args.output), + "fixture": output["fixture"], + "concurrent": { + name: {key: value for key, value in result.items() if key != "plans"} + for name, result in output["concurrent"].items() + }, + }, + indent=2, + ) + ) + if args.require_improvement and not all(output["improvement"].values()): + raise AssertionError(f"Profile acceptance failed: {output['improvement']}") + finally: + engine.dispose() + + +if __name__ == "__main__": + main() diff --git a/tests/analysis/test_inference_daily_usage.py b/tests/analysis/test_inference_daily_usage.py index cf8395f20..1719aec5c 100644 --- a/tests/analysis/test_inference_daily_usage.py +++ b/tests/analysis/test_inference_daily_usage.py @@ -179,17 +179,22 @@ def change(row, model): assert sum(row["calls"] for row in result) == 2 -def test_migration_backfills_history_and_old_schema_retains_exact_fallback(database): +def test_migration_backfills_history_and_preserves_ledger_activity(database): config = Config(str(REPO_ROOT / "alembic.ini")) try: command.downgrade(config, "0030") with database.begin() as session: session.add(invocation(NOW - timedelta(days=10))) with database() as session: - assert len(equal_activity(session)) == 1 + # Capture the pre-0031 ledger contract directly. The current complete + # chart also requires the later topic-call schema enforced at startup. + expected = [ + dict(row) for row in _ledger_activity(session, NOW - timedelta(days=30), NOW) + ] + assert len(expected) == 1 command.upgrade(config, "head") with database() as session: - assert len(equal_activity(session)) == 1 + assert equal_activity(session) == expected assert session.scalar(text("SELECT sum(calls) FROM inference_daily_usage")) == 1 assert session.scalar(select(func.count()).select_from(InferenceCall)) == 1 finally: diff --git a/tests/partners/catalog/test_migration.py b/tests/partners/catalog/test_migration.py index 836724ac5..b46bbe3cd 100644 --- a/tests/partners/catalog/test_migration.py +++ b/tests/partners/catalog/test_migration.py @@ -36,7 +36,8 @@ def test_partnership_revision_follows_master_schema(): scripts = ScriptDirectory.from_config(Config(str(ROOT / "alembic.ini"))) - assert scripts.get_heads() == [SCHEMA_REVISION] == ["0035"] + assert scripts.get_heads() == [SCHEMA_REVISION] == ["0036"] + assert scripts.get_revision("0036").down_revision == "0035" assert scripts.get_revision("0035").down_revision == "0034" assert scripts.get_revision("0034").down_revision == "0033" assert scripts.get_revision("0033").down_revision == "0032" @@ -51,6 +52,7 @@ def test_partnership_revision_follows_master_schema(): assert scripts.get_revision("0024").down_revision == "0023" assert scripts.get_revision("0023").down_revision == "0022" assert [revision.revision for revision in scripts.walk_revisions("0022", "head")] == [ + "0036", "0035", "0034", "0033", diff --git a/tests/taxonomy/test_topic_decision_reporting.py b/tests/taxonomy/test_topic_decision_reporting.py new file mode 100644 index 000000000..fb93ec1d2 --- /dev/null +++ b/tests/taxonomy/test_topic_decision_reporting.py @@ -0,0 +1,859 @@ +"""Reporting preserves workflow history while call facts follow transactional writes.""" + +import copy +import importlib +import json +import time +import uuid +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime, timedelta, timezone +from threading import Barrier, Event + +import pytest +from alembic import command +from alembic.config import Config +from alembic.migration import MigrationContext +from alembic.operations import Operations +from devfeed_core.inference_metrics import inference_charts +from devfeed_core.models import ( + Article, + PipelineProjection, + ReadModelEvent, + TopicDecisionRun, + TopicProposal, +) +from devfeed_core.read_models import set_state +from devfeed_core.topic_decision_metrics import decision_metrics +from sqlalchemy import delete, event, text + +from tests.support.paths import REPO_ROOT + +pytestmark = pytest.mark.integration +START = datetime(2026, 10, 20, tzinfo=UTC) +END = START + timedelta(days=2) + + +def call(stage="draft", started_at=START, tokens=10, **changes): + return { + "stage": stage, + "started_at": started_at.isoformat() if started_at is not None else None, + "tokens": tokens if isinstance(tokens, dict) else {"totalTokens": tokens}, + "charged_tokens": 0 if isinstance(tokens, dict) else tokens, + "escalated": False, + **changes, + } + + +def add_run(session, calls=(), *, status="pending", reviewed_at=None, run_status="active", **kw): + identifier = uuid.uuid4() + proposal = TopicProposal( + id=identifier, + batch_id=uuid.uuid4(), + slug=f"report-{identifier.hex}", + action="create", + origin="import", + source_name="Reporting fixture", + proposed={"name": "Reporting fixture"}, + created_by={"sub": "report-test"}, + status=status, + reviewed_at=reviewed_at, + reviewed_by={"sub": "reviewer"} if reviewed_at else None, + applied={} if status == "approved" else None, + ) + session.add(proposal) + session.flush() + if run_status is not None: + session.add( + TopicDecisionRun( + proposal_id=identifier, + input_hash="a" * 64, + status=run_status, + state={"calls": list(calls)}, + updated_at=START, + **kw, + ) + ) + session.flush() + return identifier + + +def facts(session, identifier): + return session.execute( + text("""SELECT ordinal, stage, stage_call, started_at, tokens, charged_tokens, escalated + FROM topic_decision_calls WHERE proposal_id=:id ORDER BY ordinal"""), + {"id": identifier}, + ).all() + + +def legacy_usage(session, start=START, now=END): + # Retain the released SQL as an independent oracle for well-formed history. + return dict( + session.execute( + text(""" + WITH calls AS ( + SELECT r.proposal_id, c, n, + (p.status IN ('approved', 'rejected') + AND p.reviewed_at BETWEEN :start AND :now) AS completed + FROM topic_decision_runs r JOIN topic_proposals p ON p.id=r.proposal_id, + jsonb_array_elements(r.state->'calls') WITH ORDINALITY AS items(c,n) + ), totals AS ( + SELECT *, greatest(coalesce((c->'tokens'->>'totalTokens')::bigint,0), + coalesce((c->'tokens'->>'inputTokens')::bigint,0) + + coalesce((c->'tokens'->>'outputTokens')::bigint,0)) AS tokens, + (c->>'started_at')::timestamptz BETWEEN :start AND :now AS in_window, + row_number() OVER (PARTITION BY proposal_id,c->>'stage' ORDER BY n) AS stage_call + FROM calls + ) + SELECT count(*) FILTER (WHERE in_window) AS calls, + coalesce(sum(tokens) FILTER (WHERE in_window),0) AS tokens, + coalesce(sum((c->>'charged_tokens')::bigint) FILTER (WHERE in_window),0) + AS charged_tokens, + count(*) FILTER (WHERE in_window AND tokens=0) AS unreported_calls, + count(*) FILTER (WHERE in_window AND (c->>'escalated')::boolean) AS escalations, + count(*) FILTER (WHERE in_window AND stage_call>1) AS repeat_calls, + CASE WHEN bool_and(tokens>0) FILTER (WHERE completed) + THEN sum(tokens) FILTER (WHERE completed)::float + / nullif(count(DISTINCT proposal_id) FILTER (WHERE completed),0) + ELSE NULL END AS tokens_per_completed_topic FROM totals + """), + {"start": start, "now": now}, + ) + .mappings() + .one() + ) + + +def test_reports_match_released_contract_for_statuses_and_whole_workflows(database): + with database.begin() as session: + add_run( + session, + [ + call("discovery", START - timedelta(hours=1), 12), + call( + tokens={"totalTokens": 4, "inputTokens": 10, "outputTokens": 5}, + charged_tokens=20, + status="failed", + ), + call(started_at=START + timedelta(hours=25), tokens=30, escalated=True), + call("verification", END, 40), + ], + status="approved", + reviewed_at=START + timedelta(hours=2), + run_status="decided", + ) + add_run( + session, + [ + call("draft", START - timedelta(hours=1), 5), + call("verification", START + timedelta(hours=26), 25), + ], + status="rejected", + reviewed_at=END, + run_status="decided", + ) + add_run( + session, + [call("discovery", START + timedelta(hours=27), 50)], + status="approved", + reviewed_at=START - timedelta(days=1), + run_status="decided", + ) + add_run( + session, + [ + call(started_at=START - timedelta(days=1), tokens=8), + call(started_at=START + timedelta(hours=28), tokens=11, escalated=True), + ], + ) + identifier = add_run( + session, + [call("discovery", START + timedelta(hours=30), 0, charged_tokens=100)], + run_status="deferred", + reason="rate_limit", + ) + session.get(TopicDecisionRun, identifier).updated_at = START + timedelta(hours=29) + identifier = add_run(session, run_status="deferred") + session.get(TopicDecisionRun, identifier).updated_at = START - timedelta(days=1) + add_run(session, run_status="review") + add_run(session, run_status=None) + add_run(session, status="approved", reviewed_at=START, run_status="decided") + with database() as session: + result = decision_metrics(session, START, END).model_dump() + assert {key: result[key] for key in legacy_usage(session)} == legacy_usage(session) + assert result == { + "pending": 5, + "actionable": 2, + "deferred": 2, + "awaiting_review": 1, + "approved": 2, + "rejected": 1, + "decisions_per_hour": 0.06, + "estimated_drain_hours": 33.3, + "calls": 7, + "tokens": 171, + "charged_tokens": 276, + "unreported_calls": 1, + "repeat_calls": 2, + "escalations": 2, + "escalation_percent": 28.6, + "tokens_per_completed_topic": 63.5, + "deferred_reasons": {"rate_limit": 1, "unspecified": 1}, + } + assert inference_charts(session, START, END).model_dump() == { + "first_recorded_at": None, + "activity": [], + "topic_activity": [ + { + "date": "2026-10-20", + "approved": 2, + "rejected": 0, + "deferred": 0, + "calls": 1, + "escalations": 0, + "repeated_stages": 0, + }, + { + "date": "2026-10-21", + "approved": 0, + "rejected": 0, + "deferred": 1, + "calls": 5, + "escalations": 2, + "repeated_stages": 2, + }, + { + "date": "2026-10-22", + "approved": 0, + "rejected": 1, + "deferred": 0, + "calls": 1, + "escalations": 0, + "repeated_stages": 0, + }, + ], + } + + +@pytest.mark.parametrize("unknown", [False, True]) +def test_completion_uses_undated_and_outside_window_calls_without_empty_denominators( + database, unknown +): + with database.begin() as session: + add_run( + session, + [ + call(started_at=None, tokens=0 if unknown else 20), + call(started_at=START - timedelta(days=30), tokens=30), + call("verification", START, 10), + ], + status="approved", + reviewed_at=END, + run_status="decided", + ) + add_run(session, status="rejected", reviewed_at=START, run_status="decided") + add_run( + session, + [call(tokens=1000)], + status="approved", + reviewed_at=END + timedelta(microseconds=1), + run_status="decided", + ) + with database() as session: + result = decision_metrics(session, START, END) + assert result.calls == 2 and result.tokens == 1010 + assert result.tokens_per_completed_topic == (None if unknown else 60) + assert ( + result.model_dump()["tokens_per_completed_topic"] + == legacy_usage(session)["tokens_per_completed_topic"] + ) + + +def test_chart_buckets_and_inclusive_boundaries_use_utc_for_offset_ranges(database): + offset = timezone(timedelta(hours=5, minutes=30)) + start = datetime(2026, 10, 20, 1, 30, tzinfo=offset) + end = start + timedelta(days=1) + with database.begin() as session: + add_run( + session, + [ + call(started_at=start - timedelta(microseconds=1)), + call(started_at=start), + call(started_at=end), + call(started_at=end + timedelta(microseconds=1)), + ], + status="approved", + reviewed_at=start, + run_status="decided", + ) + identifier = add_run(session, run_status="deferred") + session.get(TopicDecisionRun, identifier).updated_at = end + with database() as session: + session.execute(text("SET LOCAL TIME ZONE 'Pacific/Honolulu'")) + result = decision_metrics(session, start, end) + assert result.calls == 2 and result.repeat_calls == 2 and result.approved == 1 + assert [ + day.model_dump() for day in inference_charts(session, start, end).topic_activity + ] == [ + { + "date": "2026-10-19", + "approved": 1, + "rejected": 0, + "deferred": 0, + "calls": 1, + "escalations": 0, + "repeated_stages": 1, + }, + { + "date": "2026-10-20", + "approved": 0, + "rejected": 0, + "deferred": 1, + "calls": 1, + "escalations": 0, + "repeated_stages": 1, + }, + ] + + +def test_missing_and_malformed_calls_are_safe_unknown_facts(database): + malformed = [ + None, + "broken", + 7, + {}, + call(stage=None, started_at=None, tokens=None), + call( + stage="draft", + started_at=START, + tokens={"totalTokens": "invalid", "inputTokens": {}, "outputTokens": 1.2}, + escalated="invalid", + charged_tokens="overflow", + ), + call( + stage="draft", + started_at=START, + tokens={"totalTokens": "3", "inputTokens": 4, "outputTokens": 5}, + escalated="true", + charged_tokens="11", + ), + {"stage": "verification", "started_at": "not-a-timestamp", "tokens": {}}, + call( + "verification", + START, + {"totalTokens": 10**40, "inputTokens": str(2**63), "outputTokens": False}, + ), + ] + with database.begin() as session: + identifier = add_run(session, malformed, status="approved", reviewed_at=START) + for value in ({}, None, "not-an-array", 3): + empty = add_run(session) + session.get(TopicDecisionRun, empty).state = {"calls": value} + empty = add_run(session) + session.get(TopicDecisionRun, empty).state = {} + with database() as session: + rows = facts(session, identifier) + assert [row.ordinal for row in rows] == list(range(1, 10)) + assert [row.stage_call for row in rows] == [1, 2, 3, 4, 5, 1, 2, 1, 2] + assert [int(row.tokens) for row in rows] == [0, 0, 0, 0, 0, 0, 9, 0, 0] + assert rows[5].charged_tokens is None and rows[6].charged_tokens == 11 + assert not rows[5].escalated and rows[6].escalated + assert rows[7].started_at is None + result = decision_metrics(session, START, END) + assert (result.calls, result.tokens, result.charged_tokens, result.unreported_calls) == ( + 3, + 9, + 11, + 2, + ) + assert result.repeat_calls == 2 and result.escalations == 1 + assert result.tokens_per_completed_topic is None + assert inference_charts(session, START, END).topic_activity[0].calls == 3 + + +def test_special_timestamps_stay_undated_and_large_token_addition_cannot_overflow(database): + timestamps = ["now", "today", "infinity", "-infinity", "2026-02-30T12:00:00Z"] + calls = [{**call(tokens=10), "started_at": value} for value in timestamps] + calls.append(call(tokens={"inputTokens": 2**63 - 1, "outputTokens": 2**63 - 1})) + with database.begin() as session: + session.execute(text("SET LOCAL TIME ZONE 'Pacific/Honolulu'")) + identifier = add_run(session, calls, status="approved", reviewed_at=START) + naive = add_run(session, [{**call(), "started_at": "2026-10-20T00:00:00"}]) + with database() as session: + rows = facts(session, identifier) + assert all(row.started_at is None for row in rows[:-1]) + assert rows[-1].tokens == 2 * (2**63 - 1) + assert facts(session, naive)[0].started_at == START + result = decision_metrics(session, START, END) + assert result.calls == 2 and result.tokens == 2 * (2**63 - 1) + 10 + assert result.tokens_per_completed_topic == float(2 * (2**63 - 1) + 50) + + +def test_report_plans_read_the_window_without_expanding_or_ranking_old_calls(database): + with database.begin() as session: + add_run( + session, + [call(started_at=START - timedelta(days=365)) for _ in range(1000)] + + [call(started_at=START)], + ) + session.execute(text("ANALYZE topic_decision_calls")) + session.execute(text("ANALYZE topic_proposals")) + captured = [] + + def capture(connection, cursor, statement, parameters, context, executemany): + if "topic_decision_calls" in statement: + captured.append((statement, parameters)) + + with database() as session: + engine = session.get_bind() + event.listen(engine, "before_cursor_execute", capture) + try: + assert decision_metrics(session, START, END).calls == 1 + assert inference_charts(session, START, END).topic_activity[0].calls == 1 + finally: + event.remove(engine, "before_cursor_execute", capture) + assert len(captured) == 2 + for statement, parameters in captured: + document = ( + session.connection() + .exec_driver_sql("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) " + statement, parameters) + .scalar_one()[0] + ) + pending = [document["Plan"]] + nodes = [] + while pending: + node = pending.pop() + nodes.append(node) + pending.extend(node.get("Plans", [])) + assert not {"Function Scan", "WindowAgg"}.intersection( + node["Node Type"] for node in nodes + ) + scans = [node for node in nodes if node.get("Relation Name") == "topic_decision_calls"] + assert scans and all(node["Actual Rows"] <= 1 for node in scans) + assert any("started_at" in node.get("Index Cond", "") for node in nodes) + assert document["Plan"]["Temp Written Blocks"] == 0 + + +def test_facts_follow_append_settlement_correction_reorder_removal_and_noop(database): + initial = [call("draft", START, 10), call("verification", START, 20)] + with database.begin() as session: + identifier = add_run(session, initial) + with database.begin() as session: + run = session.get(TopicDecisionRun, identifier) + run.state = {"calls": [*initial, call("draft", END, 0, charged_tokens=100)]} + with database() as session: + assert [row.stage_call for row in facts(session, identifier)] == [1, 1, 2] + before = dict( + session.execute( + text("""SELECT ordinal,xmin::text FROM topic_decision_calls + WHERE proposal_id=:id"""), + {"id": identifier}, + ).all() + ) + settled = [*initial, call("draft", END, 30, charged_tokens=30)] + with database.begin() as session: + session.get(TopicDecisionRun, identifier).state = {"calls": settled} + with database() as session: + after = dict( + session.execute( + text("""SELECT ordinal,xmin::text FROM topic_decision_calls + WHERE proposal_id=:id"""), + {"id": identifier}, + ).all() + ) + assert after[1] == before[1] and after[2] == before[2] and after[3] != before[3] + assert decision_metrics(session, START, END).tokens == 60 + with database.begin() as session: + # Checkpoints and exact replay must not rewrite facts on every state save. + session.execute( + text("""UPDATE topic_decision_runs + SET state=state || '{"checkpoint":"saved"}'::jsonb WHERE proposal_id=:id"""), + {"id": identifier}, + ) + session.execute( + text("UPDATE topic_decision_runs SET state=state WHERE proposal_id=:id"), + {"id": identifier}, + ) + with database() as session: + assert ( + dict( + session.execute( + text("""SELECT ordinal,xmin::text FROM topic_decision_calls + WHERE proposal_id=:id"""), + {"id": identifier}, + ).all() + ) + == after + ) + corrected = [settled[2], {**settled[0], "stage": "verification"}, settled[1]] + with database.begin() as session: + session.get(TopicDecisionRun, identifier).state = {"calls": corrected} + with database() as session: + assert [ + (row.ordinal, row.stage, row.stage_call, int(row.tokens)) + for row in facts(session, identifier) + ] == [(1, "draft", 1, 30), (2, "verification", 1, 10), (3, "verification", 2, 20)] + with database() as session: + session.get(TopicDecisionRun, identifier).state = {"calls": []} + session.flush() + assert facts(session, identifier) == [] + session.rollback() + with database.begin() as session: + assert len(facts(session, identifier)) == 3 + session.get(TopicDecisionRun, identifier).state = {"calls": corrected[:1]} + with database() as session: + assert len(facts(session, identifier)) == 1 + assert decision_metrics(session, START, END).tokens == 30 + with database.begin() as session: + session.get(TopicDecisionRun, identifier).state = {} + with database() as session: + assert facts(session, identifier) == [] + + +@pytest.mark.parametrize( + ("changes", "expected"), + [ + ({"tokens": {"totalTokens": 99}}, {"tokens": 99}), + ({"started_at": END.isoformat()}, {"started_at": END}), + ({"stage": "verification"}, {"stage": "verification", "stage_call": 1}), + ({"charged_tokens": None}, {"charged_tokens": None}), + ({"escalated": True}, {"escalated": True}), + ({"status": "returned"}, {}), + ], +) +def test_correcting_one_call_does_not_lock_unchanged_call_facts(database, changes, expected): + initial = [call(tokens=index + 1) for index in range(20)] + corrected = [*initial[:-1], {**initial[-1], **changes}] + with database.begin() as session: + identifier = add_run(session, initial) + with database() as session: + before = facts(session, identifier) + engine = session.get_bind() + with engine.connect() as holder, holder.begin(): + holder.execute( + text("""SELECT ordinal FROM topic_decision_calls + WHERE proposal_id=:id AND ordinal=1 FOR UPDATE"""), + {"id": identifier}, + ) + with database.begin() as session: + session.execute(text("SET LOCAL lock_timeout='2s'")) + run = session.get(TopicDecisionRun, identifier) + run.state = {"calls": corrected} + # The former upsert attempted to lock ordinal 1 before checking that + # it was unchanged, so a correction of ordinal 20 timed out here. + session.flush() + with database() as session: + after = facts(session, identifier) + assert after[:-1] == before[:-1] + assert after[-1]._asdict() == {**before[-1]._asdict(), **expected} + assert session.get(TopicDecisionRun, identifier).state["calls"] == corrected + + +def test_concurrent_atomic_appends_never_duplicate_or_lose_fact_occurrences(database): + with database.begin() as session: + identifier = add_run(session, [call(tokens=1)]) + writers = 4 + barrier = Barrier(writers) + + def append(index): + with database.begin() as session: + barrier.wait(timeout=10) + session.execute( + text("""UPDATE topic_decision_runs + SET state=jsonb_set(state,'{calls}',(state->'calls') || CAST(:call AS jsonb)) + WHERE proposal_id=:id"""), + {"id": identifier, "call": json.dumps([call(tokens=index + 2)])}, + ) + + with ThreadPoolExecutor(max_workers=writers) as pool: + list(pool.map(append, range(writers))) + with database() as session: + rows = facts(session, identifier) + assert [row.ordinal for row in rows] == [1, 2, 3, 4, 5] + assert [row.stage_call for row in rows] == [1, 2, 3, 4, 5] + assert sorted(int(row.tokens) for row in rows) == [1, 2, 3, 4, 5] + result = decision_metrics(session, START, END) + assert (result.calls, result.tokens, result.repeat_calls) == (5, 15, 4) + + +@pytest.mark.parametrize("deleted", ["run", "proposal", "truncate"]) +def test_source_deletion_and_transactional_truncate_remove_call_facts(database, deleted): + with database.begin() as session: + identifier = add_run(session, [call()]) + with database() as session: + if deleted == "truncate": + session.execute(text("TRUNCATE topic_decision_runs CASCADE")) + elif deleted == "run": + session.delete(session.get(TopicDecisionRun, identifier)) + else: + session.delete(session.get(TopicProposal, identifier)) + session.flush() + assert facts(session, identifier) == [] + session.rollback() + with database() as session: + assert len(facts(session, identifier)) == 1 + + +def test_migration_backfill_matches_live_projection_and_downgrade_preserves_json(database): + migration = importlib.import_module("migrations.versions.0036_topic_decision_calls") + calls = [ + call("draft", START - timedelta(days=1), 10), + call("draft", START, 20, escalated=True), + {"stage": "verification", "started_at": "invalid", "tokens": {}}, + None, + ] + with database.begin() as session: + identifier = add_run(session, calls) + with database() as session: + expected = facts(session, identifier) + original = copy.deepcopy(session.get(TopicDecisionRun, identifier).state) + engine = session.get_bind() + with engine.connect() as connection, connection.begin() as transaction: + with Operations.context(MigrationContext.configure(connection)): + migration.downgrade() + assert connection.scalar(text("SELECT to_regclass('topic_decision_calls')")) is None + assert ( + connection.scalar( + text("SELECT state FROM topic_decision_runs WHERE proposal_id=:id"), + {"id": identifier}, + ) + == original + ) + migration.upgrade() + actual = connection.execute( + text("""SELECT ordinal,stage,stage_call,started_at, + tokens,charged_tokens,escalated FROM topic_decision_calls + WHERE proposal_id=:id ORDER BY ordinal"""), + {"id": identifier}, + ).all() + assert actual == expected + migration.downgrade() + migration.upgrade() + assert connection.scalar(text("SELECT count(*) FROM topic_decision_calls")) == len( + calls + ) + transaction.rollback() + + +def test_topic_migration_round_trip_keeps_scheduler_clock_capture_and_source_history(database): + config = Config(str(REPO_ROOT / "alembic.ini")) + capture_sql = text("""SELECT tgname, pg_get_triggerdef(oid) FROM pg_trigger + WHERE tgrelid='articles'::regclass + AND tgname IN ('read_model_changed','read_model_article_clock_changed') ORDER BY tgname""") + with database.begin() as session: + proposal_id = add_run(session, [call(tokens=10), call(tokens=20)]) + article = Article( + canonical_url=f"https://publisher.example/migration/{proposal_id}", + url_hash=proposal_id.hex, + title="Keep the scheduler clock during topic rollback", + automation_next_check_at=START, + ) + session.add(article) + session.flush() + article_id = article.id + set_state(session, "pipeline", {"target_policy": "retained", "serving_policy": "retained"}) + session.add( + PipelineProjection( + policy_version="retained", + kind="article", + entity_id=article_id, + payload={"next_check_at": START.isoformat(), "title": article.title}, + stage="analysis", + state="waiting", + created_at=START, + activity_at=START, + evaluated_at=START, + ) + ) + session.flush() + session.execute(delete(ReadModelEvent)) + retained_triggers = session.execute(capture_sql).all() + assert len(retained_triggers) == 2 + source_state = copy.deepcopy(session.get(TopicDecisionRun, proposal_id).state) + expected_facts = facts(session, proposal_id) + try: + command.downgrade(config, "0035") + with database.begin() as session: + assert session.scalar(text("SELECT version_num FROM alembic_version")) == "0035" + assert session.scalar(text("SELECT to_regclass('topic_decision_calls')")) is None + assert session.execute(capture_sql).all() == retained_triggers + assert session.get(TopicDecisionRun, proposal_id).state == source_state + clock = START + timedelta(hours=1) + session.execute( + text("UPDATE articles SET automation_next_check_at=:clock WHERE id=:id"), + {"clock": clock, "id": article_id}, + ) + assert session.scalar(text("SELECT count(*) FROM read_model_events")) == 0 + card = session.get(PipelineProjection, ("retained", "article", article_id)) + assert datetime.fromisoformat(card.payload["next_check_at"]) == clock + assert card.evaluated_at == START + command.upgrade(config, "head") + with database.begin() as session: + assert session.scalar(text("SELECT version_num FROM alembic_version")) == "0036" + assert facts(session, proposal_id) == expected_facts + assert session.get(TopicDecisionRun, proposal_id).state == source_state + assert session.execute(capture_sql).all() == retained_triggers + clock = START + timedelta(hours=2) + session.execute( + text("UPDATE articles SET automation_next_check_at=:clock WHERE id=:id"), + {"clock": clock, "id": article_id}, + ) + card = session.get(PipelineProjection, ("retained", "article", article_id)) + assert datetime.fromisoformat(card.payload["next_check_at"]) == clock + assert card.evaluated_at == START + assert session.scalar(text("SELECT count(*) FROM read_model_events")) == 0 + finally: + command.upgrade(config, "head") + + +@pytest.fixture +def isolated_reporting_schema(database): + with database() as session: + engine = session.get_bind() + schema = f"topic_reporting_lock_{uuid.uuid4().hex}" + with engine.begin() as connection: + original_head = connection.scalar(text("SELECT version_num FROM alembic_version")) + connection.execute(text(f"CREATE SCHEMA {schema}")) + connection.execute( + text( + f"CREATE TABLE {schema}.topic_proposals (id uuid PRIMARY KEY, " + "reviewed_at timestamptz)" + ) + ) + connection.execute( + text( + f"CREATE TABLE {schema}.topic_decision_runs (proposal_id uuid PRIMARY KEY, " + "state jsonb NOT NULL)" + ) + ) + connection.execute( + text(f"INSERT INTO {schema}.topic_proposals VALUES (:id, :reviewed)"), + {"id": uuid.UUID(int=1), "reviewed": START}, + ) + connection.execute( + text(f"INSERT INTO {schema}.topic_decision_runs VALUES (:id, CAST(:state AS jsonb))"), + {"id": uuid.UUID(int=1), "state": json.dumps({"calls": [call()]})}, + ) + try: + yield engine, schema, original_head + finally: + with engine.begin() as connection: + connection.execute(text(f"DROP SCHEMA {schema} CASCADE")) + + +def wait_for_migration_retry(connection, started, migration_pid): + assert started.wait(timeout=5) + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + connection.execute(text("SELECT pg_stat_clear_snapshot()")) + if ( + connection.scalar( + text("SELECT wait_event FROM pg_stat_activity WHERE pid=:pid"), + {"pid": migration_pid[0]}, + ) + == "PgSleep" + ): + return + time.sleep(0.01) + pytest.fail("Migration did not release its partial locks and retry admission") + + +@pytest.mark.parametrize("flushed", [False, True]) +def test_migration_retries_source_lock_admission_without_blocking_reverse_order_writer( + isolated_reporting_schema, flushed +): + engine, schema, original_head = isolated_reporting_schema + migration = importlib.import_module("migrations.versions.0036_topic_decision_calls") + started = Event() + migration_pid = [] + + def upgrade(): + with engine.connect() as connection, connection.begin() as transaction: + connection.execute(text(f"SET LOCAL search_path TO {schema}")) + migration_pid.append(connection.scalar(text("SELECT pg_backend_pid()"))) + started.set() + with Operations.context(MigrationContext.configure(connection)): + migration.upgrade() + assert connection.scalar(text("SELECT count(*) FROM topic_decision_calls")) == 1 + transaction.rollback() + + with engine.connect() as writer, writer.begin(), ThreadPoolExecutor(max_workers=1) as pool: + writer.execute(text(f"SET LOCAL search_path TO {schema}")) + writer.execute(text("SET LOCAL lock_timeout = '2s'")) + writer.execute(text("SELECT * FROM topic_decision_runs FOR UPDATE")) + if flushed: + writer.execute(text("UPDATE topic_decision_runs SET state=state")) + task = pool.submit(upgrade) + try: + wait_for_migration_retry(writer, started, migration_pid) + # This is the ORM's run -> proposal flush order. It must finish + # while migration admission retries, rather than forming a cycle. + writer.execute(text("UPDATE topic_decision_runs SET state=state")) + writer.execute(text("UPDATE topic_proposals SET reviewed_at=reviewed_at")) + finally: + writer.rollback() + task.result(timeout=10) + with engine.connect() as connection: + assert ( + connection.scalar( + text("SELECT to_regclass(:table)"), + {"table": f"{schema}.topic_decision_calls"}, + ) + is None + ) + assert connection.scalar(text("SELECT version_num FROM alembic_version")) == original_head + + +@pytest.mark.parametrize("blocker", ["writer", "report"]) +def test_downgrade_retries_all_table_locks_without_blocking_existing_work( + isolated_reporting_schema, blocker +): + engine, schema, original_head = isolated_reporting_schema + migration = importlib.import_module("migrations.versions.0036_topic_decision_calls") + with engine.begin() as connection: + connection.execute(text(f"SET LOCAL search_path TO {schema}")) + with Operations.context(MigrationContext.configure(connection)): + migration.upgrade() + started = Event() + migration_pid = [] + + def downgrade(): + with engine.connect() as connection, connection.begin() as transaction: + connection.execute(text(f"SET LOCAL search_path TO {schema}")) + migration_pid.append(connection.scalar(text("SELECT pg_backend_pid()"))) + started.set() + with Operations.context(MigrationContext.configure(connection)): + migration.downgrade() + assert ( + connection.scalar( + text("SELECT to_regclass(:table)"), {"table": f"{schema}.topic_decision_calls"} + ) + is None + ) + assert connection.scalar(text("SELECT count(*) FROM topic_decision_runs")) == 1 + transaction.rollback() + + with engine.connect() as active, active.begin(), ThreadPoolExecutor(max_workers=1) as pool: + active.execute(text(f"SET LOCAL search_path TO {schema}")) + active.execute(text("SET LOCAL lock_timeout = '2s'")) + if blocker == "writer": + active.execute(text("UPDATE topic_decision_runs SET state=state")) + else: + active.execute(text("SELECT * FROM topic_decision_calls")) + task = pool.submit(downgrade) + try: + wait_for_migration_retry(active, started, migration_pid) + if blocker == "writer": + active.execute(text("UPDATE topic_proposals SET reviewed_at=reviewed_at")) + else: + # A report can retain facts AccessShare before asking for sources. + assert active.scalar(text("SELECT count(*) FROM topic_decision_runs")) == 1 + assert active.scalar(text("SELECT count(*) FROM topic_proposals")) == 1 + finally: + active.rollback() + task.result(timeout=10) + with engine.connect() as connection: + assert connection.scalar(text(f"SELECT count(*) FROM {schema}.topic_decision_calls")) == 1 + assert connection.scalar(text(f"SELECT state FROM {schema}.topic_decision_runs")) == { + "calls": [call()] + } + assert connection.scalar(text("SELECT version_num FROM alembic_version")) == original_head diff --git a/tests/tooling/profiling/test_topic_report_profile_script.py b/tests/tooling/profiling/test_topic_report_profile_script.py new file mode 100644 index 000000000..c5f640d4e --- /dev/null +++ b/tests/tooling/profiling/test_topic_report_profile_script.py @@ -0,0 +1,664 @@ +"""The topic profiler bounds disposable work and measures the actual report SQL.""" + +import json +import os +import stat +import sys +from datetime import timedelta +from pathlib import Path +from types import SimpleNamespace + +import pytest +from devfeed_core.inference_metrics import inference_charts +from devfeed_core.topic_decision_metrics import decision_metrics +from sqlalchemy import event, text + +from scripts import profile_topic_reports as profile + + +@pytest.fixture(autouse=True) +def clear_libpq_routes(monkeypatch): + for name in ("PGHOSTADDR", "PGSERVICE", "PGSERVICEFILE"): + monkeypatch.delenv(name, raising=False) + + +@pytest.mark.parametrize( + "destination", + [ + "outside.json", + "../escape.json", + "reports/../../escape.json", + "reports/sub/../escape.json", + "reports/report.txt", + ], +) +def test_output_rejects_traversal_outside_paths_and_non_json_artifacts( + monkeypatch, tmp_path, destination +): + monkeypatch.chdir(tmp_path) + with pytest.raises(ValueError, match="reports folder|.json extension"): + profile.report_output_path(Path(destination)) + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize("absolute", [False, True]) +def test_output_creates_private_json_inside_reports_without_overwriting( + monkeypatch, tmp_path, absolute +): + monkeypatch.chdir(tmp_path) + relative = Path("reports/nested/topic-reports.json") + output = tmp_path / relative + destination = output if absolute else relative + assert profile.report_output_path(destination) == output + profile.write_profile_output(destination, {"complete": True}) + assert json.loads(output.read_text()) == {"complete": True} + assert output.read_text().endswith("\n") + assert stat.S_IMODE(output.stat().st_mode) == 0o600 + with pytest.raises(ValueError, match="already exists"): + profile.write_profile_output(destination, {"replaced": True}) + assert json.loads(output.read_text()) == {"complete": True} + + +@pytest.mark.parametrize("kind", ["file", "directory", "symlink", "parent-symlink"]) +def test_output_rejects_existing_files_directories_and_symlinks(monkeypatch, tmp_path, kind): + monkeypatch.chdir(tmp_path) + reports = tmp_path / "reports" + reports.mkdir() + output = reports / "profile.json" + if kind == "file": + output.write_text("Retain the earlier report") + elif kind == "directory": + output.mkdir() + elif kind == "symlink": + original = reports / "original.json" + original.write_text("Retain the earlier report") + output.symlink_to(original) + else: + original = reports / "original" + original.mkdir() + (reports / "alias").symlink_to(original, target_is_directory=True) + output = reports / "alias" / "profile.json" + with pytest.raises(ValueError, match="already exists|must not be symlinks"): + profile.write_profile_output(output, {"replaced": True}) + if kind == "file": + assert output.read_text() == "Retain the earlier report" + elif kind == "symlink": + assert original.read_text() == "Retain the earlier report" + elif kind == "parent-symlink": + assert not (original / "profile.json").exists() + + +def test_output_cannot_follow_a_parent_replaced_after_validation(monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + output = tmp_path / "reports" / "nested" / "profile.json" + output.parent.mkdir(parents=True) + outside = tmp_path / "outside" + outside.mkdir() + validate = profile.report_output_path + + def swap_directory(destination): + validated = validate(destination) + output.parent.rmdir() + output.parent.symlink_to(outside, target_is_directory=True) + return validated + + monkeypatch.setattr(profile, "report_output_path", swap_directory) + with pytest.raises(OSError): + profile.write_profile_output(output, {"complete": True}) + assert not (outside / "profile.json").exists() + + +@pytest.mark.parametrize("symlink", [False, True]) +def test_output_exclusive_creation_preserves_a_target_added_after_validation( + monkeypatch, tmp_path, symlink +): + monkeypatch.chdir(tmp_path) + output = tmp_path / "reports" / "profile.json" + victim = tmp_path / "earlier.json" + victim.write_text("Retain the existing file") + original_open = os.open + + def competing_open(path, flags, mode=0o777, *, dir_fd=None): + if path == output.name and flags & os.O_EXCL: + if symlink: + os.symlink(victim, path, dir_fd=dir_fd) + else: + descriptor = original_open( + path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600, dir_fd=dir_fd + ) + with os.fdopen(descriptor, "w") as earlier: + earlier.write("Retain the earlier report") + return original_open(path, flags, mode, dir_fd=dir_fd) + + monkeypatch.setattr(os, "open", competing_open) + with pytest.raises(FileExistsError): + profile.write_profile_output(output, {"replaced": True}) + assert victim.read_text() == "Retain the existing file" + assert output.read_text() == ( + "Retain the existing file" if symlink else "Retain the earlier report" + ) + + +@pytest.mark.parametrize("existing", [False, True]) +def test_cli_rejects_invalid_output_before_loading_code_or_opening_database( + monkeypatch, tmp_path, capsys, existing +): + monkeypatch.chdir(tmp_path) + output = tmp_path / "reports" / "existing.json" if existing else tmp_path / "outside.json" + if existing: + output.parent.mkdir() + output.write_text("Retain the earlier report") + monkeypatch.setenv("DEVFEED_TEST_DATABASE_URL", "postgresql://test@localhost/devfeed_test") + monkeypatch.setattr( + sys, + "argv", + ["topic-profiler", "--baseline-root", str(profile.ROOT), "--output", str(output)], + ) + monkeypatch.setattr( + profile, "report_functions", lambda *_args: pytest.fail("Loaded report code") + ) + monkeypatch.setattr(profile, "create_engine", lambda *_a, **_kw: pytest.fail("Opened database")) + with pytest.raises(SystemExit) as error: + profile.main() + assert error.value.code == 2 + assert "Output" in capsys.readouterr().err + if existing: + assert output.read_text() == "Retain the earlier report" + + +@pytest.mark.parametrize( + "url", + [ + "postgresql://test@production.invalid/devfeed_test", + "postgresql://test@localhost/devfeed", + "postgresql://test@localhost", + "postgresql://test@/devfeed_test", + "sqlite://localhost/devfeed_test", + "postgresql://test@localhost/devfeed_test?host=production.invalid", + "postgresql://test@localhost/devfeed_test?hostaddr=192.0.2.1", + "postgresql://test@localhost/devfeed_test?service=production", + "postgresql://test@localhost/devfeed_test?options=-csearch_path%3Dother", + ], +) +def test_profiler_rejects_nonlocal_databases_and_connection_query_overrides(url): + with pytest.raises(ValueError, match="explicitly local PostgreSQL"): + profile.validate_database(url) + + +@pytest.mark.parametrize("name", ["PGHOSTADDR", "PGSERVICE", "PGSERVICEFILE"]) +def test_profiler_rejects_libpq_routing_overrides(monkeypatch, name): + monkeypatch.setenv(name, "unexpected-route") + with pytest.raises(ValueError, match="routing overrides"): + profile.validate_database("postgresql+psycopg://test@127.0.0.1/devfeed_test") + + +@pytest.mark.parametrize("host", ["localhost", "127.0.0.1"]) +def test_profiler_accepts_explicit_local_disposable_database(host): + value = f"postgresql+psycopg://test@{host}:55499/devfeed_test" + assert str(profile.validate_database(value)) == value + + +@pytest.mark.parametrize( + "version,required_major", + [("170000", None), ("170011", 17), ("180000", None), ("180001", 18), ("190001", None)], +) +def test_profiler_accepts_supported_servers_and_matching_explicit_major(version, required_major): + profile.validate_server_version(version, required_major=required_major) + + +@pytest.mark.parametrize( + "version,required_major,message", + [ + ("160011", None, "17 or newer"), + ("90624", None, "17 or newer"), + ("180000", 17, "requires PostgreSQL 17"), + ("170011", 18, "requires PostgreSQL 18"), + ], +) +def test_profiler_rejects_unsupported_servers_and_explicit_major_mismatch( + version, required_major, message +): + with pytest.raises(ValueError, match=message): + profile.validate_server_version(version, required_major=required_major) + + +def test_cli_checks_explicit_server_major_before_seeding_or_profiling( + monkeypatch, tmp_path, capsys +): + server = {"version": "PostgreSQL 18", "default_work_mem": "4MB", "version_num": "180000"} + + class Connection: + def __enter__(self): + return self + + def __exit__(self, *_args): + pass + + def execute(self, _statement): + return SimpleNamespace(mappings=lambda: SimpleNamespace(one=lambda: server)) + + disposed = [] + engine = SimpleNamespace(connect=Connection, dispose=lambda: disposed.append(True)) + monkeypatch.setenv("DEVFEED_TEST_DATABASE_URL", "postgresql://test@localhost/devfeed_test") + monkeypatch.setattr(profile, "create_engine", lambda *_args, **_kw: engine) + monkeypatch.setattr(profile, "seed", lambda *_args: pytest.fail("Seeded mismatched server")) + monkeypatch.setattr(profile, "report_sample", lambda *_a, **_kw: pytest.fail("Profiled server")) + monkeypatch.chdir(tmp_path) + output = tmp_path / "reports" / "profile.json" + monkeypatch.setattr( + sys, + "argv", + [ + "topic-profiler", + "--baseline-root", + str(profile.ROOT), + "--output", + str(output), + "--seed", + "--require-postgres-major", + "17", + ], + ) + with pytest.raises(SystemExit) as error: + profile.main() + assert error.value.code == 2 + assert "requires PostgreSQL 17; connected server is 18" in capsys.readouterr().err + assert not output.exists() + assert disposed == [True] + + +@pytest.mark.parametrize( + "options", + [ + ["--articles", "9"], + ["--articles", "1000001"], + ["--runs", "9"], + ["--runs", "100001"], + ["--history-bytes", "-1"], + ["--history-bytes", "40001"], + ["--repeats", "0"], + ["--repeats", "11"], + ["--workers", "0"], + ["--workers", "9"], + ["--writes", "0"], + ["--writes", "1001"], + ], +) +def test_cli_rejects_unbounded_work_before_loading_code_or_connecting( + monkeypatch, tmp_path, capsys, options +): + monkeypatch.setenv("DEVFEED_TEST_DATABASE_URL", "postgresql://test@localhost/devfeed_test") + monkeypatch.setattr( + sys, + "argv", + [ + "topic-profiler", + "--baseline-root", + str(tmp_path), + "--output", + str(tmp_path / "result.json"), + *options, + ], + ) + monkeypatch.setattr(profile, "create_engine", lambda *_a, **_kw: pytest.fail("Opened database")) + monkeypatch.setattr(profile, "report_functions", lambda *_a: pytest.fail("Loaded report code")) + with pytest.raises(SystemExit) as error: + profile.main() + assert error.value.code == 2 + assert "profiling bounds" in capsys.readouterr().err + assert not (tmp_path / "result.json").exists() + + +def test_cli_rejects_unsupported_required_major_before_connecting(monkeypatch, tmp_path, capsys): + monkeypatch.setenv("DEVFEED_TEST_DATABASE_URL", "postgresql://test@localhost/devfeed_test") + monkeypatch.setattr( + sys, + "argv", + [ + "topic-profiler", + "--baseline-root", + str(tmp_path), + "--output", + str(tmp_path / "profile.json"), + "--require-postgres-major", + "16", + ], + ) + monkeypatch.setattr(profile, "create_engine", lambda *_a, **_kw: pytest.fail("Opened database")) + monkeypatch.setattr(profile, "report_functions", lambda *_a: pytest.fail("Loaded report code")) + with pytest.raises(SystemExit) as error: + profile.main() + assert error.value.code == 2 + assert "--require-postgres-major must be 17 or newer" in capsys.readouterr().err + + +def fixture_snapshot(session): + return { + "articles": session.scalar(text("SELECT count(*) FROM articles")), + "runs": session.execute( + text("""SELECT proposal_id,state,updated_at + FROM topic_decision_runs ORDER BY proposal_id""") + ).all(), + "facts": session.execute( + text("""SELECT * FROM topic_decision_calls + ORDER BY proposal_id,ordinal""") + ).all(), + "inference": session.scalar(text("SELECT count(*) FROM inference_calls")), + } + + +@pytest.mark.integration +def test_seed_refuses_existing_history_and_report_sample_captures_actual_plans(database): + with database() as session: + engine = session.get_bind() + profile.seed(engine, articles=20, runs=20, history_bytes=64) + with database() as session: + before = fixture_snapshot(session) + with pytest.raises(ValueError, match="requires empty fixture tables"): + profile.seed(engine, articles=20, runs=20, history_bytes=64) + loaded = profile.report_functions(profile.ROOT, "test-direct") + assert loaded[0] is not decision_metrics and loaded[1] is not inference_charts + actual = profile.report_sample(engine, loaded, days=30, explain=True) + expected = profile.report_sample(engine, [decision_metrics, inference_charts], days=30) + assert actual["output"] == expected["output"] + assert actual["output"][0]["calls"] == 60 + assert len(actual["output"][1]["topic_activity"]) == 31 + assert len(actual["plans"]) == 2 + assert actual["plan_totals"]["execution_ms"] >= 0 + assert actual["plan_totals"]["temp_written_blocks"] == 0 + assert all("Execution Time" in plan and "Plan" in plan for plan in actual["plans"]) + assert any("topic_decision_calls" in json.dumps(plan) for plan in actual["plans"]) + with database() as session: + assert fixture_snapshot(session) == before + + +@pytest.mark.integration +def test_correction_wal_measurement_includes_after_triggers_and_rolls_back(database): + with database() as session: + engine = session.get_bind() + profile.seed(engine, articles=10, runs=10, history_bytes=64) + with database() as session: + before = fixture_snapshot(session) + for legacy in (True, False): + plan = profile.correction_plan(engine, legacy=legacy) + assert isinstance(plan["statement_wal_bytes"], int) + assert plan["statement_wal_bytes"] >= plan["Plan"]["WAL Bytes"] > 0 + assert "AFTER triggers" in plan["statement_wal_scope"] + if not legacy: + assert any( + trigger["Trigger Name"] == "topic_decision_calls_sync" and trigger["Calls"] == 1 + for trigger in plan["Triggers"] + ) + assert plan["statement_wal_bytes"] > plan["Plan"]["WAL Bytes"] + with database() as session: + assert fixture_snapshot(session) == before + + +def committed_fixture_snapshot(session): + return { + **fixture_snapshot(session), + "table_counts": { + table.name: session.scalar(text(f'SELECT count(*) FROM "{table.name}"')) + for table in profile.models.Base.metadata.sorted_tables + }, + } + + +@pytest.mark.parametrize( + "iterations,call_count,repeats", + [(0, 5, 1), (1001, 5, 1), (1, 1, 1), (1, 21, 1), (1, 5, 0), (1, 5, 11)], +) +def test_durable_writes_reject_unbounded_work_before_connecting( + monkeypatch, iterations, call_count, repeats +): + monkeypatch.setattr( + profile, "_durable_write_phase", lambda *_a, **_kw: pytest.fail("Started durable writes") + ) + with pytest.raises(ValueError, match="exceeds supported bounds"): + profile.durable_write_profile( + object(), iterations=iterations, call_count=call_count, repeats=repeats + ) + + +@pytest.mark.integration +@pytest.mark.parametrize("call_count", [5, 20]) +def test_durable_writes_commit_lifecycle_steps_and_clean_up_each_strategy(database, call_count): + with database() as session: + engine = session.get_bind() + profile.seed(engine, articles=10, runs=10, history_bytes=64) + with database() as session: + before = committed_fixture_snapshot(session) + visible_steps = [] + final_states = [] + + def observe_commits(_connection, _cursor, statement, parameters, _context, _many): + normalized = " ".join(statement.lower().split()) + if normalized.startswith("update topic_decision_runs set state="): + value = json.loads(parameters["value"]) + with engine.connect() as observer: + state = observer.scalar( + text("SELECT state FROM topic_decision_runs WHERE proposal_id=:proposal"), + {"proposal": parameters["proposal"]}, + ) + if isinstance(value, list): + operation = "append" + assert len(state["calls"]) == call_count - 1 + elif isinstance(value, dict): + operation = "settle" + assert len(state["calls"]) == call_count + assert state["calls"][-1]["status"] == "running" + assert state["calls"][-1]["charged_tokens"] == 40000 + elif isinstance(value, int): + operation = "correction" + assert state["calls"][-1]["status"] == "returned" + assert state["calls"][-1]["tokens"] == {"totalTokens": 1200} + assert state["calls"][0]["charged_tokens"] == 1100 + else: + operation = "checkpoint" + assert value == "complete" + assert state["calls"][0]["charged_tokens"] in {50000, 50001} + assert "checkpoint" not in state + visible_steps.append(operation) + elif normalized.startswith("delete from articles where id="): + with engine.connect() as observer: + assert ( + observer.scalar( + text("SELECT count(*) FROM articles WHERE id=ANY(:ids)"), parameters + ) + == 2 + ) + states = observer.execute( + text("""SELECT r.proposal_id,r.state FROM topic_decision_runs r + JOIN topic_proposals p ON p.id=r.proposal_id + WHERE p.slug LIKE 'durable-%' ORDER BY r.state->>'evidence', + r.state#>>'{calls,0,charged_tokens}'""") + ).all() + assert len(states) == 2 + for proposal, state in states: + assert len(state["calls"]) == call_count + assert state["calls"][-1]["status"] == "returned" + assert state["checkpoint"] == "complete" + facts = ( + observer.execute( + text("""SELECT charged_tokens FROM topic_decision_calls + WHERE proposal_id=:proposal ORDER BY ordinal"""), + {"proposal": proposal}, + ) + .scalars() + .all() + ) + if len(final_states) == 0: + assert facts == [1100] * (call_count - 1) + else: + assert facts == [call["charged_tokens"] for call in state["calls"]] + final_states.append([state for _proposal, state in states]) + + event.listen(engine, "before_cursor_execute", observe_commits) + try: + result = profile.durable_write_profile( + engine, iterations=2, call_count=call_count, repeats=1 + ) + finally: + event.remove(engine, "before_cursor_execute", observe_commits) + assert visible_steps == ["append", "settle", "correction", "checkpoint"] * 4 + assert final_states[0] == final_states[1] + assert result["calls_per_workflow"] == call_count + assert result["paired_order"] == [["legacy", "facts"]] + assert result["source_states_equal"] is True + assert result["projection_verified"] is True + for strategy in result["strategies"].values(): + assert strategy["committed"] is True + assert strategy["iterations"] == 2 + assert strategy["committed_transactions"] == 8 + assert strategy["samples"] == 8 + assert strategy["wal_bytes"] > 0 + assert strategy["wal_records"] > 0 + assert strategy["wal_full_page_images"] >= 0 + assert strategy["elapsed_seconds"] > 0 + assert strategy["application_thread_cpu_seconds"] > 0 + assert set(strategy["operations"]) == {"append", "settle", "correction", "checkpoint"} + for operation in strategy["operations"].values(): + assert operation["samples"] == 2 + assert 0 < operation["median_ms"] <= operation["p95_ms"] + with database() as session: + assert committed_fixture_snapshot(session) == before + assert ( + session.scalar( + text("""SELECT tgenabled FROM pg_trigger + WHERE tgrelid='topic_decision_runs'::regclass + AND tgname='topic_decision_calls_sync'""") + ) + == "O" + ) + + +@pytest.mark.integration +def test_durable_writes_clean_up_partial_commits_and_restore_projection_after_failure(database): + with database() as session: + engine = session.get_bind() + profile.seed(engine, articles=10, runs=10, history_bytes=64) + with database() as session: + before = committed_fixture_snapshot(session) + observed_committed_append = [] + + def fail_settlement(_connection, _cursor, statement, parameters, _context, _many): + if " ".join(statement.lower().split()).startswith("update topic_decision_runs set state="): + value = json.loads(parameters["value"]) + if isinstance(value, dict): + with engine.connect() as observer: + state = observer.scalar( + text("SELECT state FROM topic_decision_runs WHERE proposal_id=:proposal"), + {"proposal": parameters["proposal"]}, + ) + assert len(state["calls"]) == 5 + assert state["calls"][-1]["status"] == "running" + observed_committed_append.append(True) + raise RuntimeError("Settlement failed after the append committed") + + event.listen(engine, "before_cursor_execute", fail_settlement) + try: + with pytest.raises(RuntimeError, match="after the append committed"): + profile.durable_write_profile(engine, iterations=2, call_count=5, repeats=1) + finally: + event.remove(engine, "before_cursor_execute", fail_settlement) + assert observed_committed_append == [True] + with database() as session: + assert committed_fixture_snapshot(session) == before + assert ( + session.scalar( + text("""SELECT tgenabled FROM pg_trigger + WHERE tgrelid='topic_decision_runs'::regclass + AND tgname='topic_decision_calls_sync'""") + ) + == "O" + ) + + +@pytest.mark.integration +def test_seed_repeated_stages_cross_report_window_and_utc_day_boundaries(database): + with database() as session: + engine = session.get_bind() + profile.seed(engine, articles=10, runs=90, history_bytes=64) + with database() as session: + for number, days in ((29, 30), (89, 90)): + rows = session.execute( + text("""SELECT c.stage,c.stage_call,c.started_at + FROM topic_decision_calls c JOIN topic_proposals p ON p.id=c.proposal_id + WHERE p.slug=:slug ORDER BY c.ordinal"""), + {"slug": f"topic-profile-{number}"}, + ).all() + draft = [row for row in rows if row.stage == "draft"] + verification = [row for row in rows if row.stage == "verification"] + start = profile.NOW - timedelta(days=days) + assert [row.stage_call for row in draft] == [1, 2] + assert draft[0].started_at < start < draft[1].started_at <= profile.NOW + assert [row.stage_call for row in verification] == [1, 2] + assert verification[0].started_at.date() < verification[1].started_at.date() + + +@pytest.mark.integration +@pytest.mark.parametrize("fail", [False, True]) +def test_cli_preserves_facts_and_restores_projection_after_concurrent_samples( + database, monkeypatch, tmp_path, capsys, fail +): + with database() as session: + engine = session.get_bind() + profile.seed(engine, articles=10, runs=10, history_bytes=64) + with database() as session: + before = fixture_snapshot(session) + monkeypatch.chdir(tmp_path) + output = tmp_path / "reports" / "profile.json" + monkeypatch.setattr( + sys, + "argv", + [ + "topic-profiler", + "--baseline-root", + str(profile.ROOT), + "--output", + str(output), + "--repeats", + "1", + "--workers", + "2", + "--writes", + "2", + ], + ) + if fail: + + def failure(*_args, **_kwargs): + raise RuntimeError("Concurrent reporter failed") + + monkeypatch.setattr(profile, "concurrent", failure) + with pytest.raises(RuntimeError, match="Concurrent reporter failed"): + profile.main() + assert not output.exists() + else: + profile.main() + assert json.loads(capsys.readouterr().out)["fixture"]["articles"] == 10 + result = json.loads(output.read_text()) + assert [report["days"] for report in result["reports"]] == [1, 7, 30, 90] + assert all(report["exact_output_parity"] for report in result["reports"]) + for sample in result["concurrent"].values(): + assert sample["ingestion"]["committed"] is False + assert sample["ingestion"]["iterations"] == 2 + assert sample["report_pairs"] == 2 + assert len(sample["plans"]) == 2 + assert [sample["calls_per_workflow"] for sample in result["durable_writes"]] == [5, 20] + for sample in result["durable_writes"]: + assert sample["source_states_equal"] is True + assert sample["projection_verified"] is True + for strategy in sample["strategies"].values(): + assert strategy["committed"] is True + assert strategy["committed_transactions"] == 8 + with database() as session: + assert fixture_snapshot(session) == before + assert ( + session.scalar( + text("""SELECT tgenabled FROM pg_trigger + WHERE tgrelid='topic_decision_runs'::regclass + AND tgname='topic_decision_calls_sync'""") + ) + == "O" + )