diff --git a/.forgejo/workflows/ci-smoke.yaml b/.forgejo/workflows/ci-smoke.yaml index 908adfc..43ca7d2 100644 --- a/.forgejo/workflows/ci-smoke.yaml +++ b/.forgejo/workflows/ci-smoke.yaml @@ -22,7 +22,7 @@ jobs: pytest-smoke: runs-on: self-hosted steps: - - name: Test enforcement, inventory and installed package + - name: Test enforcement, inventory, package and PostgreSQL run: | set -eu REF="${GITHUB_SHA:?commit SHA required}" @@ -39,4 +39,4 @@ jobs: fi uv sync --locked --group dev --extra runtime make ci-check - echo "hub-core source, enforcement, inventory and package checks passed @ ${REF}" + echo "hub-core source, enforcement, inventory, package and PostgreSQL checks passed @ ${REF}" diff --git a/Makefile b/Makefile index 03a47de..8556ec2 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: install test runtime-run conformance container-build ecosystem-regression inventory-check package-check ci-check +.PHONY: install test runtime-run conformance container-build ecosystem-regression inventory-check package-check ci-check postgres-test UV ?= uv ECOSYSTEM_REGRESSION ?= /home/worsch/the-custodian/scripts/hub-ecosystem-regression.sh @@ -31,4 +31,7 @@ package-check: $(UV) build --out-dir dist/ci @set -eu; set -- dist/ci/*.whl; test "$$#" -eq 1; $(UV) run --no-project --isolated --refresh-package hub-core --with "$$1" python -I tools/check_installed_package.py -ci-check: test inventory-check package-check +postgres-test: + HUB_CORE_TEST_POSTGRES=1 $(UV) run --locked --extra runtime python -m pytest -q tests/test_postgres_integration.py + +ci-check: test inventory-check package-check postgres-test diff --git a/docs/conformance.md b/docs/conformance.md index 5c43916..b9ee16a 100644 --- a/docs/conformance.md +++ b/docs/conformance.md @@ -50,12 +50,13 @@ harness reports only the implemented profile above. ## Access enforcement and CI gates `make ci-check` runs the test suite, checks the reviewed access inventory for -source drift, builds distributions and validates an installed wheel outside the +source drift, runs disposable PostgreSQL integration tests, builds distributions +and validates an installed wheel outside the checkout's import path. The wheel check uses a fresh environment and refreshes the Hub package so rebuilding the same version cannot reuse an older installation. Forgejo runs these gates for `main` pushes and manual runs, using the full commit SHA and a unique temporary checkout. CI installs the locked development and runtime dependencies first. Individual gates are -`make test`, `make inventory-check` and `make package-check`. +`make test`, `make inventory-check`, `make package-check` and `make postgres-test`. `tests/test_enforced_conformance.py` runs all twelve existing Tier 2/3 checks through an explicitly enforced runtime using a real signed IAM JWT and synthetic @@ -71,3 +72,27 @@ It makes no owner requests and starts no service. The separate optional Audit Core interoperability suite still requires `HUB_CORE_AUDIT_CORE_SOURCE` and is not silently represented as covered by ordinary CI. Local CI-equivalent success is not a deployed Forgejo receipt or live owner/platform acceptance. + + +## Disposable PostgreSQL gate + +`make postgres-test` enables `tests/test_postgres_integration.py`. It creates its +own Docker container and a separate database for each test, then removes both. +It never accepts an operator database URL. The container uses a random fixture +password, a loopback-only ephemeral port and tmpfs data storage. The PostgreSQL +16 Alpine image is pinned by digest in the test file. Readiness waits for TCP so +the image's temporary initialization server cannot be mistaken for the final one. + +The tests run the entire packaged Alembic chain through `0006_outcome_outbox`, +check the pending-outcome downgrade guard and re-upgrade, demonstrate concurrent +`SKIP LOCKED` delivery, roll back business and ledger writes after an outbox +failure, and verify persisted retry state. A child worker exits with `os._exit` +after writing a synthetic receiver receipt but before committing its local +acknowledgement; a new connection then replays the identical envelope. + +This is a required gate of `make ci-check`, including Forgejo. The runner needs +Docker daemon access and the pinned image cached or registry pull access. Missing +Docker/image access fails this gate; it does not silently pass. Ordinary pytest +skips this module unless `HUB_CORE_TEST_POSTGRES=1`; the dedicated gate enables it +explicitly. Tests use disposable superuser credentials and synthetic custody, +so production runtime grants and live receiver admission remain separate checks. diff --git a/docs/operation-outcome-audit.md b/docs/operation-outcome-audit.md index 754e5e4..fcfe0aa 100644 --- a/docs/operation-outcome-audit.md +++ b/docs/operation-outcome-audit.md @@ -82,8 +82,10 @@ The real Audit Core receiver/storage source accepts the eight-field envelope and returns a duplicate receipt on replay. Its operational-readiness classification is explicitly a test fixture; this is not live custody evidence. -PostgreSQL multiworker lock scheduling, production grants, retention and deployed -failure-detection/receiver acceptance still require integration receipts. +Disposable PostgreSQL tests now verify multiworker lock scheduling, the full +migration chain, rollback, persisted retries and process-exit replay. See the +[PostgreSQL gate](conformance.md#disposable-postgresql-gate). Production grants, +retention and deployed failure-detection/receiver acceptance still require receipts. Validation on 2026-09-28: 366 tests pass with the opt-in owner-source suite enabled; inventory drift, package build and isolated installed-wheel checks pass. The diff --git a/tests/test_postgres_integration.py b/tests/test_postgres_integration.py new file mode 100644 index 0000000..9805338 --- /dev/null +++ b/tests/test_postgres_integration.py @@ -0,0 +1,258 @@ +"""Opt-in, real PostgreSQL tests. Always creates its own disposable container. + +Run make postgres-test. No caller-provided database URL is accepted. +""" +import asyncio +import json +import os +from pathlib import Path +import secrets +import subprocess +import sys +import time +from uuid import uuid4 + +import pytest +import sqlalchemy as sa +from sqlalchemy.engine import URL + +if os.getenv('HUB_CORE_TEST_POSTGRES') != '1': + pytest.skip('run make postgres-test for disposable PostgreSQL', allow_module_level=True) + +from alembic import command +from alembic.config import Config +from hub_core.runtime.models import MessageCommand +from hub_core.runtime.postgres_store import PostgresPortStore +from hub_core.runtime.tables import runtime_messages, runtime_audit_ledger, runtime_outcome_outbox +from hub_core.security.context import current_authorization +from test_access_boundary import Owners + +IMAGE = 'postgres@sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f58fddf30ffc07777' +HEAD = '0006_outcome_outbox' + + +def docker(*args): + return subprocess.run(['docker',*args], check=True, capture_output=True, text=True, timeout=120).stdout.strip() + + +@pytest.fixture(scope='module') +def postgres(): + password = secrets.token_hex(16) + env = {**os.environ,'POSTGRES_PASSWORD':password} + process = subprocess.run(['docker','run','--detach','--rm', + '--label','hub-core.test=postgres-integration','--tmpfs','/var/lib/postgresql/data:rw', + '--publish','127.0.0.1::5432','--env','POSTGRES_PASSWORD',IMAGE], + env=env,check=True,capture_output=True,text=True,timeout=120) + container = process.stdout.strip() + try: + port = int(docker('port',container,'5432/tcp').rsplit(':',1)[1]) + deadline = time.monotonic()+40 + while True: + result = subprocess.run(['docker','exec',container,'pg_isready','-h','127.0.0.1','-U','postgres'], + capture_output=True,timeout=5) + if result.returncode == 0: + break + if time.monotonic() >= deadline: + pytest.fail('disposable PostgreSQL did not become ready') + time.sleep(.2) + yield URL.create('postgresql+psycopg2',username='postgres',password=password, + host='127.0.0.1',port=port,database='postgres') + finally: + docker('rm','--force',container) + + +@pytest.fixture +def database(postgres,monkeypatch): + name = 'hub_test_' + uuid4().hex + admin = sa.create_engine(postgres,isolation_level='AUTOCOMMIT') + with admin.connect() as connection: + connection.exec_driver_sql('CREATE DATABASE ' + name) + url = postgres.set(database=name) + # Alembic's environment can override config. Bind it explicitly to this owned DB. + monkeypatch.setenv('DATABASE_URL',url.render_as_string(hide_password=False)) + monkeypatch.delenv('HUB_CORE_MIGRATION_ROLE',raising=False) + monkeypatch.delenv('HUB_CORE_MIGRATION_SCHEMA',raising=False) + config = Config() + config.set_main_option('script_location',str(Path('hub_core/migrations').resolve())) + config.set_main_option('sqlalchemy.url',url.render_as_string(hide_password=False)) + try: + command.upgrade(config,HEAD) + yield url, config + finally: + with admin.connect() as connection: + connection.exec_driver_sql('DROP DATABASE ' + name + ' WITH (FORCE)') + admin.dispose() + + +def store_for(url): + from sqlalchemy.ext.asyncio import create_async_engine + return PostgresPortStore(create_async_engine(url.set(drivername='postgresql+asyncpg'))) + + +async def commit_message(store): + owners = Owners() + context = await owners.controller().authorize('verified-root','hub.message.write', + '/ports/messaging/messages',str(uuid4()),'a'*64) + binding = current_authorization.set(context) + try: + return await store.send_message(MessageCommand(schema_version='0.1.0',correlation_id=uuid4(), + from_address='agent:root',to_addresses=['agent:reader'],body='disposable fixture')) + finally: + current_authorization.reset(binding) + + +async def pending(store): + async with store.sessions() as session: + return [dict(r) for r in (await session.execute(sa.select(runtime_outcome_outbox))).mappings()] + + +def test_full_migration_chain_and_guarded_downgrade(database): + url,config = database + engine = sa.create_engine(url) + try: + with engine.connect() as connection: + assert connection.exec_driver_sql('SELECT version_num FROM alembic_version').scalar_one() == HEAD + assert sa.inspect(connection).get_foreign_keys('runtime_outcome_outbox')[0]['referred_table'] == 'runtime_audit_ledger' + async def write(): + store = store_for(url) + try: + assert await store.readiness_checks() == {'database':'ok'} + await commit_message(store) + finally: + await store.aclose() + asyncio.run(write()) + with pytest.raises(RuntimeError,match='drained outbox'): + command.downgrade(config,'0005_message_identity_aliases') + with engine.begin() as connection: + # Synthetic custody receipt permits testing the schema rollback path. + connection.execute(runtime_outcome_outbox.update().values(delivered_at=time.time())) + command.downgrade(config,'0005_message_identity_aliases') + command.upgrade(config,HEAD) + with engine.connect() as connection: + assert connection.execute(sa.select(sa.func.count()).select_from(runtime_messages)).scalar_one() == 1 + assert connection.execute(sa.select(sa.func.count()).select_from(runtime_audit_ledger)).scalar_one() == 1 + finally: + engine.dispose() + + +def test_two_workers_skip_locked_row_without_duplicate_claim(database): + url,_ = database + async def run(): + first,second = store_for(url),store_for(url) + held,release = asyncio.Event(),asyncio.Event() + seen = [] + class HoldingSink: + async def append_outcome(self,event): + seen.append(event['id']) + held.set() + await release.wait() + class OtherSink: + async def append_outcome(self,event): + seen.append(event['id']) + task = None + try: + await commit_message(first) + await commit_message(first) + task = asyncio.create_task(first.deliver_outcomes(HoldingSink(),limit=1)) + await asyncio.wait_for(held.wait(),2) + # Must finish while first worker still holds its transaction's row lock. + assert await asyncio.wait_for(second.deliver_outcomes(OtherSink(),limit=1),2) == 1 + assert not task.done() + release.set() + assert await task == 1 + assert len(seen) == len(set(seen)) == 2 + assert all(row['delivered_at'] is not None for row in await pending(first)) + finally: + release.set() + if task is not None and not task.done(): + task.cancel() + await asyncio.gather(task,return_exceptions=True) + await first.aclose() + await second.aclose() + asyncio.run(run()) + + +def test_outbox_failure_rolls_back_real_postgres_transaction(database): + url,_ = database + async def run(): + store = store_for(url) + def fail(connection,cursor,statement,parameters,context,many): + if statement.startswith('INSERT INTO runtime_outcome_outbox'): + raise RuntimeError('injected outbox insert failure') + sa.event.listen(store.engine.sync_engine,'before_cursor_execute',fail) + try: + with pytest.raises(RuntimeError,match='injected outbox'): + await commit_message(store) + async with store.sessions() as session: + for table in (runtime_messages,runtime_audit_ledger,runtime_outcome_outbox): + assert (await session.execute(sa.select(sa.func.count()).select_from(table))).scalar_one() == 0 + finally: + await store.aclose() + asyncio.run(run()) + + +def test_worker_process_exit_after_receipt_replays_same_envelope(database,tmp_path): + url,_ = database + async def prepare(): + store = store_for(url) + try: + await commit_message(store) + finally: + await store.aclose() + asyncio.run(prepare()) + receipt = tmp_path/'accepted.json' + script = ''' +import asyncio,json,os +from hub_core.runtime.postgres_store import PostgresPortStore +class Receiver: + async def append_outcome(self,event): + with open(os.environ['HUB_TEST_RECEIPT'],'w') as stream: + json.dump(event,stream) + stream.flush() + os.fsync(stream.fileno()) + os._exit(73) # Receiver accepted; local DB delivery mark never committed. +async def main(): + store=PostgresPortStore.from_url(os.environ['HUB_TEST_DATABASE']) + await store.deliver_outcomes(Receiver(),limit=1) +asyncio.run(main()) +''' + env = {**os.environ,'HUB_TEST_RECEIPT':str(receipt), + 'HUB_TEST_DATABASE':url.set(drivername='postgresql+asyncpg').render_as_string(hide_password=False)} + result = subprocess.run([sys.executable,'-c',script],env=env,capture_output=True,timeout=15) + assert result.returncode == 73 + accepted = json.loads(receipt.read_text()) + async def recover(): + store = store_for(url) + class DuplicateReceiver: + async def append_outcome(self,event): + assert event == accepted + try: + row, = await pending(store) + assert row['attempts'] == 0 and row['delivered_at'] is None + assert await asyncio.wait_for(store.deliver_outcomes(DuplicateReceiver(),limit=1),3) == 1 + assert await store.deliver_outcomes(DuplicateReceiver()) == 0 + finally: + await store.aclose() + asyncio.run(recover()) + + +def test_receiver_failure_persists_backoff_across_connection_reopen(database): + url,_ = database + async def run(): + store = store_for(url) + class FailingReceiver: + async def append_outcome(self,event): + raise OSError('fixture receiver unavailable') + try: + await commit_message(store) + assert await store.deliver_outcomes(FailingReceiver()) == 0 + finally: + await store.aclose() + reopened = store_for(url) + try: + row, = await pending(reopened) + assert row['attempts'] == 1 and row['next_attempt'] >= row['created_at'] + 1 + assert row['delivered_at'] is None + finally: + await reopened.aclose() + asyncio.run(run()) diff --git a/workplans/HUB-WP-0012-netkingdom-platform-root-access.md b/workplans/HUB-WP-0012-netkingdom-platform-root-access.md index 420634f..fc9a98a 100644 --- a/workplans/HUB-WP-0012-netkingdom-platform-root-access.md +++ b/workplans/HUB-WP-0012-netkingdom-platform-root-access.md @@ -389,6 +389,25 @@ suite. Inventory drift, distribution builds, isolated installed-wheel checks and current migration/context/store wheel contents pass. No database migration or deployment was applied to a live service. +## Real PostgreSQL verification — 2026-09-28 + +Added a required `make postgres-test` gate to `make ci-check` and therefore the +existing Forgejo workflow. A digest-pinned, loopback-only disposable PostgreSQL +container hosts isolated test databases. Operator database URLs are not accepted; +the fixture removes its own databases and container after the run. + +Five integration tests exercise the full Alembic chain to `0006`, guarded +schema downgrade/re-upgrade, two concurrent `SKIP LOCKED` delivery workers, +atomic rollback, persisted retry state and replay after an actual child worker +process exits after a synthetic custody receipt but before local acknowledgement. + +Final `make ci-check` passed: **360 ordinary tests** (two optional modules +skipped in that phase), inventory/build/installed-wheel checks, then **five real +PostgreSQL tests** with the opt-in enabled. Fixture-container cleanup was verified. +Production grants, deployment and live receiver acceptance remain open; no live +database was contacted or migrated. +See [test details](../docs/conformance.md#disposable-postgresql-gate). + ## Acceptance checkpoints - [x] Architecture/source/runtime review captured; new implementation owner is hub-core