From 87e76c62d63cf3f0365a032dba5f3ad6a609cf1b Mon Sep 17 00:00:00 2001 From: exzent Date: Fri, 6 Nov 2026 10:10:59 +0800 Subject: [PATCH] feat: add fenced payout delivery claims --- backend/app/cli/payout_delivery.py | 97 +++- .../app/repositories/payout_delivery_repo.py | 68 ++- backend/app/schemas/payout_delivery.py | 28 ++ .../app/services/payout_delivery_service.py | 32 ++ .../tests/test_deployment_control_plane.py | 1 + backend/tests/test_deployment_queue_schema.py | 22 +- backend/tests/test_payout_claims_db.py | 441 ++++++++++++++++++ .../R__payout_delivery_claim_identity.sql | 24 + .../V112__payout_delivery_claims.sql | 9 + docs/PAYOUT_RECEIPTS.md | 28 +- 10 files changed, 720 insertions(+), 30 deletions(-) create mode 100644 backend/tests/test_payout_claims_db.py create mode 100644 db/migrations/R__payout_delivery_claim_identity.sql create mode 100644 db/migrations/V112__payout_delivery_claims.sql diff --git a/backend/app/cli/payout_delivery.py b/backend/app/cli/payout_delivery.py index d6262841..10638602 100644 --- a/backend/app/cli/payout_delivery.py +++ b/backend/app/cli/payout_delivery.py @@ -7,13 +7,26 @@ import httpx from asyncpg import exceptions as pg_errors -from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.exc import DBAPIError, SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from app.core.config import get_settings from app.repositories.payout_delivery_repo import PayoutDeliveryRepository -from app.schemas.payout_delivery import AllocationIntent, AllocationKey, ReceiptIntent -from app.services.payout_delivery_service import admit_intent, show_intent, verify_receipt +from app.schemas.payout_delivery import ( + AllocationIntent, + AllocationKey, + ClaimFence, + ClaimKey, + ClaimRequest, + ReceiptIntent, + RenewRequest, +) +from app.services.payout_delivery_service import ( + admit_intent, + manage_claim, + show_intent, + verify_receipt, +) # Supported configuration, SQLAlchemy, raw driver and transport failures stay private. ADMISSION_ERRORS = ( @@ -45,49 +58,83 @@ def run(source: TextIO, destination: TextIO, transport: httpx.BaseTransport | No ] +async def execute_admission( + command: str, + intent: AllocationKey | AllocationIntent | ClaimKey, + session_factory: async_sessionmaker[AsyncSession] | None, +) -> dict: + """Own transaction completion and dispose only a locally created engine.""" + engine = None + try: + if session_factory is None: + engine = create_async_engine(get_settings().database_url, echo=False) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + async with session_factory.begin() as db: + repo = PayoutDeliveryRepository(db) + if isinstance(intent, ClaimKey): + result = await manage_claim(repo, command, intent) + elif isinstance(intent, AllocationIntent): + result = await admit_intent(repo, intent) + else: + result = await show_intent(repo, intent) + except DBAPIError as error: + state = getattr(error.orig, "sqlstate", None) + result = { + "verdict": "busy" + if isinstance(intent, ClaimKey) and state in {"55P03", "57014"} + else "storage_error" + } + except ADMISSION_ERRORS: + # The process boundary must never expose SQL, connection URLs or payloads. + result = {"verdict": "storage_error"} + finally: + if engine is not None: + try: + await engine.dispose() + except ADMISSION_ERRORS: + result = {"verdict": "storage_error"} + return result + + async def run_admission( command: str, source: TextIO, destination: TextIO, session_factory: async_sessionmaker[AsyncSession] | None = None, ) -> int: - """Own validation, transaction completion and the lifetime of an owned engine.""" - engine = None + """Validate bounded input and emit only facts from a completed transaction.""" try: payload = source.read(65537) - model = {"submit": AllocationIntent, "show": AllocationKey}[command] + model = { + "submit": AllocationIntent, + "show": AllocationKey, + "claim": ClaimRequest, + "renew": RenewRequest, + "release": ClaimFence, + "recover": ClaimFence, + "lease-show": ClaimKey, + }[command] intent = model.model_validate_json(payload if len(payload) <= 65536 else "") except (KeyError, ValueError, TypeError, OSError, RecursionError): result = {"verdict": "invalid_input"} else: - try: - if session_factory is None: - engine = create_async_engine(get_settings().database_url, echo=False) - session_factory = async_sessionmaker(engine, expire_on_commit=False) - async with session_factory.begin() as db: - repo = PayoutDeliveryRepository(db) - if isinstance(intent, AllocationIntent): - result = await admit_intent(repo, intent) - else: - result = await show_intent(repo, intent) - except ADMISSION_ERRORS: - # The process boundary must never expose SQL, connection URLs or payloads. - result = {"verdict": "storage_error"} - finally: - if engine is not None: - try: - await engine.dispose() - except ADMISSION_ERRORS: - result = {"verdict": "storage_error"} + result = await execute_admission(command, intent, session_factory) destination.write(json.dumps(result) + "\n") return { "accepted": 0, "replayed": 0, "found": 0, + "claimed": 0, + "renewed": 0, + "released": 0, + "recovered": 0, "invalid_input": 5, "not_found": 6, "conflict": 7, "storage_error": 8, + "busy": 9, + "stale": 10, + "unavailable": 11, }[result["verdict"]] diff --git a/backend/app/repositories/payout_delivery_repo.py b/backend/app/repositories/payout_delivery_repo.py index 476aadd9..d6aa8dde 100644 --- a/backend/app/repositories/payout_delivery_repo.py +++ b/backend/app/repositories/payout_delivery_repo.py @@ -3,7 +3,7 @@ from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession -from app.schemas.payout_delivery import AllocationIntent, AllocationKey +from app.schemas.payout_delivery import AllocationIntent, AllocationKey, ClaimKey class PayoutDeliveryRepository: @@ -54,3 +54,69 @@ async def read(self, key: AllocationKey) -> dict | None: .first() ) return dict(row) if row else None + + async def claim_budget(self) -> None: + """Bound transaction-local database statements and lock waits.""" + await self.db.execute(text("SET LOCAL lock_timeout = '250ms'")) + await self.db.execute(text("SET LOCAL statement_timeout = '1000ms'")) + + async def read_claim(self, key: ClaimKey) -> dict | None: + """Inspect the stored fence without repairing an expired claim.""" + await self.claim_budget() + row = ( + ( + await self.db.execute( + text("SELECT * FROM payout_delivery_claims WHERE intent_id = :intent_id"), + key.model_dump(), + ) + ) + .mappings() + .first() + ) + return dict(row) if row else None + + async def mutate_claim(self, command: str, request: ClaimKey) -> tuple[bool, dict | None]: + """Lock the immutable parent before applying one bounded child mutation.""" + await self.claim_budget() + values = request.model_dump() + parent = await self.db.scalar( + text( + "SELECT intent_id FROM payout_delivery_intents " + "WHERE intent_id = :intent_id FOR UPDATE NOWAIT" + ), + values, + ) + if parent is None: + return False, None + if command == "claim": + sql = ( + "INSERT INTO " + "payout_delivery_claims(intent_id, epoch, state, worker_id, expires_at) " + "VALUES (:intent_id, 1, 'claimed', :worker_id, " + "clock_timestamp() + :ttl_seconds * interval '1 second') " + "ON CONFLICT (intent_id) DO UPDATE SET epoch = payout_delivery_claims.epoch + 1, " + "state = 'claimed', worker_id = " + "EXCLUDED.worker_id, expires_at = EXCLUDED.expires_at " + "WHERE payout_delivery_claims.state = 'ready'" + ) + else: + sql = self.fenced_claim_sql(command) + row = (await self.db.execute(text(sql + " RETURNING *"), values)).mappings().first() + return True, dict(row) if row else None + + @staticmethod + def fenced_claim_sql(command: str) -> str: + """Keep state and complementary live/expired predicates on every fenced write.""" + change = ( + "expires_at = GREATEST(expires_at, clock_timestamp() + " + ":ttl_seconds * interval '1 second')" + if command == "renew" + else "state = 'ready', worker_id = NULL, expires_at = NULL" + ) + comparison = "<=" if command == "recover" else ">" + return ( + "UPDATE payout_delivery_claims SET " + + change + + " WHERE intent_id = :intent_id AND worker_id = :worker_id AND epoch = :epoch " + "AND state = 'claimed' AND expires_at " + comparison + " clock_timestamp()" + ) diff --git a/backend/app/schemas/payout_delivery.py b/backend/app/schemas/payout_delivery.py index c2b7ca53..a3b57ba4 100644 --- a/backend/app/schemas/payout_delivery.py +++ b/backend/app/schemas/payout_delivery.py @@ -1,6 +1,7 @@ """Strict input and balance evidence for payout receipt verification.""" from typing import Annotated, Literal, Self +from uuid import UUID from pydantic import AfterValidator, AnyHttpUrl, BaseModel, ConfigDict, Field, model_validator from solders.hash import Hash, ParseHashError @@ -98,3 +99,30 @@ class TokenBalance(BaseModel): owner: PublicKey program_id: PublicKey = Field(alias="programId") token_amount: TokenAmount = Field(alias="uiTokenAmount") + + +class ClaimKey(BaseModel): + """Identify one admitted intent for lease coordination.""" + + model_config = ConfigDict(strict=True, extra="forbid") + intent_id: UUID + + +class ClaimRequest(ClaimKey): + """Require an explicit coordinator and bounded operational TTL.""" + + worker_id: UUID + ttl_seconds: Annotated[int, Field(ge=1, le=300)] + + +class ClaimFence(ClaimKey): + """Identify one exact persisted ownership generation.""" + + worker_id: UUID + epoch: Annotated[int, Field(ge=1, le=9223372036854775807)] + + +class RenewRequest(ClaimFence): + """Renew one live generation without changing its identity.""" + + ttl_seconds: Annotated[int, Field(ge=1, le=300)] diff --git a/backend/app/services/payout_delivery_service.py b/backend/app/services/payout_delivery_service.py index 520b025c..225ed290 100644 --- a/backend/app/services/payout_delivery_service.py +++ b/backend/app/services/payout_delivery_service.py @@ -1,6 +1,7 @@ """Read-only verification of the deliberately narrow legacy payout receipt shape.""" import struct +from datetime import UTC from typing import Any import base58 @@ -11,6 +12,7 @@ from app.schemas.payout_delivery import ( AllocationIntent, AllocationKey, + ClaimKey, ReceiptIntent, TokenBalance, solana_hash, @@ -220,3 +222,33 @@ async def show_intent(repo: PayoutDeliveryRepository, key: AllocationKey) -> dic if row is None: return {"verdict": "not_found"} return {"verdict": "found", "intent_id": str(row["intent_id"]), "state": row["state"]} + + +async def manage_claim(repo: PayoutDeliveryRepository, command: str, request: ClaimKey) -> dict: + """Return observed lease facts without owning the caller's transaction.""" + if command == "lease-show": + row = await repo.read_claim(request) + verdict = "found" + missing = "not_found" + else: + parent, row = await repo.mutate_claim(command, request) + if not parent: + return {"verdict": "not_found"} + verdict = { + "claim": "claimed", + "renew": "renewed", + "release": "released", + "recover": "recovered", + }[command] + missing = "unavailable" if command == "claim" else "stale" + if row is None: + return {"verdict": missing} + expiry = row["expires_at"] + return { + "verdict": verdict, + "intent_id": str(row["intent_id"]), + "state": row["state"], + "epoch": row["epoch"], + "worker_id": str(row["worker_id"]) if row["worker_id"] else None, + "expires_at": expiry.astimezone(UTC).isoformat() if expiry else None, + } diff --git a/backend/tests/test_deployment_control_plane.py b/backend/tests/test_deployment_control_plane.py index 7eabff7e..4467be46 100644 --- a/backend/tests/test_deployment_control_plane.py +++ b/backend/tests/test_deployment_control_plane.py @@ -61,6 +61,7 @@ async def deployment_engine(deployment_postgres) -> AsyncIterator[AsyncEngine]: "V107__project_deployments_quota_identity.sql", "V110__project_deployments_lifecycle.sql", "V111__payout_delivery.sql", + "V112__payout_delivery_claims.sql", ) if (MIGRATIONS / name).is_file() ] diff --git a/backend/tests/test_deployment_queue_schema.py b/backend/tests/test_deployment_queue_schema.py index 584a84a9..28367879 100644 --- a/backend/tests/test_deployment_queue_schema.py +++ b/backend/tests/test_deployment_queue_schema.py @@ -38,5 +38,25 @@ async def test_shared_bootstrap_installs_payout_immutability(request): } with pytest.raises(DBAPIError) as error: async with engine.begin() as connection: - await connection.execute(text("TRUNCATE payout_delivery_intents")) + await connection.execute(text("TRUNCATE payout_delivery_intents CASCADE")) + assert getattr(error.value.orig, "sqlstate") == "23514" + + +@pytest.mark.usefixtures("deployment_engine") +async def test_shared_bootstrap_installs_claim_guards(request): + """Require the shared migration loop to install both enabled claim guards.""" + engine = request.getfixturevalue("deployment_engine") + async with engine.connect() as connection: + rows = ( + await connection.execute( + text( + "SELECT tgname, tgenabled::text FROM pg_trigger " + "WHERE tgrelid = 'public.payout_delivery_claims'::regclass AND NOT tgisinternal" + ) + ) + ).all() + assert set(rows) == {("payout_claim_identity", "O"), ("payout_claim_no_truncate", "O")} + with pytest.raises(DBAPIError) as error: + async with engine.begin() as connection: + await connection.execute(text("TRUNCATE payout_delivery_claims")) assert getattr(error.value.orig, "sqlstate") == "23514" diff --git a/backend/tests/test_payout_claims_db.py b/backend/tests/test_payout_claims_db.py new file mode 100644 index 00000000..942fcc02 --- /dev/null +++ b/backend/tests/test_payout_claims_db.py @@ -0,0 +1,441 @@ +"""Prove exact-intent leases through actual CLI, services and PostgreSQL.""" + +import asyncio +import io +import json +import os +import sys +from pathlib import Path +from uuid import uuid4 + +import pytest +import test_payout_delivery_db as admission +from asyncpg import UndefinedTableError +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.exc import DBAPIError + +from app.cli.payout_delivery import run_admission +from app.repositories.payout_delivery_repo import PayoutDeliveryRepository +from app.schemas.payout_delivery import ClaimFence, ClaimRequest, RenewRequest +from app.services.payout_delivery_service import admit_intent, manage_claim + +allocation = admission.allocation +delivery_db = admission.delivery_db +marketplace_pg = admission.marketplace_pg +MIGRATIONS = Path(__file__).resolve().parents[2] / "db/migrations" + + +@pytest.fixture +async def claims_db(delivery_db, allocation): + maker, url, schema = delivery_db + async with maker.begin() as db: + conn = await db.connection() + raw = await conn.get_raw_connection() + assert raw.driver_connection is not None + for name in ("V112__payout_delivery_claims.sql", "R__payout_delivery_claim_identity.sql"): + await raw.driver_connection.execute((MIGRATIONS / name).read_text()) + intent = await admit_intent(PayoutDeliveryRepository(db), allocation) + return maker, url, schema, {"intent_id": intent["intent_id"], "worker_id": str(uuid4())} + + +async def invoke_claim(maker, command, payload): + """Execute the CLI transaction boundary and parse its fixed JSON output.""" + out = io.StringIO() + code = await run_admission(command, io.StringIO(json.dumps(payload)), out, maker) + return code, json.loads(out.getvalue()) + + +async def seed_claim(maker, key, expired): + """Insert a permitted initial epoch using actual server time.""" + async with maker.begin() as db: + await db.execute( + text( + "INSERT INTO payout_delivery_claims VALUES " + "(:intent_id, 1, 'claimed', :worker_id, " + "clock_timestamp() + :offset * interval '1 second')" + ), + key | {"offset": -1 if expired else 300}, + ) + + +async def test_cli_lost_response_generations_and_nonshortening(claims_db): + maker, _, _, key = claims_db + code, first = await invoke_claim(maker, "claim", key | {"ttl_seconds": 300}) + assert code == 0 and first["epoch"] == 1 + assert (await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}))[ + 1 + ] == first | {"verdict": "found"} + assert (await invoke_claim(maker, "renew", key | {"epoch": 1, "ttl_seconds": 1}))[1][ + "expires_at" + ] == first["expires_at"] + assert await invoke_claim(maker, "claim", key | {"ttl_seconds": 1}) == ( + 11, + {"verdict": "unavailable"}, + ) + for epoch in (1, 2, 3): + assert (await invoke_claim(maker, "release", key | {"epoch": epoch}))[1] == { + "verdict": "released", + "intent_id": key["intent_id"], + "state": "ready", + "epoch": epoch, + "worker_id": None, + "expires_at": None, + } + assert (await invoke_claim(maker, "claim", key | {"ttl_seconds": 300}))[1][ + "epoch" + ] == epoch + 1 + for command in ("renew", "release", "recover"): + payload = key | {"epoch": epoch} | ({"ttl_seconds": 1} if command == "renew" else {}) + before = await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}) + assert await invoke_claim(maker, command, payload) == (10, {"verdict": "stale"}) + assert ( + await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}) == before + ) + + +@pytest.mark.parametrize("command", ["renew", "release", "recover"]) +@pytest.mark.parametrize("cause", ["worker", "epoch", "missing", "ready", "wrong_time"]) +async def test_fenced_zero_rows_observable(claims_db, command, cause): + maker, _, _, key = claims_db + expired = (command == "recover") != (cause == "wrong_time") + if cause != "missing": + await seed_claim(maker, key, expired) + if cause == "ready": + await invoke_claim(maker, "recover" if expired else "release", key | {"epoch": 1}) + payload = key | {"epoch": 2 if cause == "epoch" else 1} + if cause == "worker": + payload["worker_id"] = str(uuid4()) + if command == "renew": + payload["ttl_seconds"] = 30 + before = await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}) + assert await invoke_claim(maker, command, payload) == (10, {"verdict": "stale"}) + assert await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}) == before + + +async def test_server_clock_expiry_in_one_transaction(claims_db): + maker, _, _, key = claims_db + async with maker.begin() as db: + start = await db.scalar(text("SELECT now()")) + expiry = await db.scalar( + text( + "INSERT INTO payout_delivery_claims VALUES " + "(:intent_id, 1, 'claimed', :worker_id, clock_timestamp()) RETURNING expires_at" + ), + key, + ) + current = await db.scalar(text("SELECT clock_timestamp()")) + assert start < expiry <= current + repo = PayoutDeliveryRepository(db) + request = ClaimFence.model_validate_json(json.dumps(key | {"epoch": 1})) + assert (await manage_claim(repo, "release", request))["verdict"] == "stale" + assert (await manage_claim(repo, "recover", request))["verdict"] == "recovered" + assert (await invoke_claim(maker, "claim", key | {"ttl_seconds": 1}))[1]["epoch"] == 2 + + +@pytest.mark.parametrize("command", ["claim", "renew", "release", "recover"]) +async def test_caller_rollback_each_mutation(claims_db, command): + maker, _, _, key = claims_db + if command != "claim": + await seed_claim(maker, key, command == "recover") + before = await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}) + model = {"claim": ClaimRequest, "renew": RenewRequest}.get(command, ClaimFence) + payload = key | ({"epoch": 1} if command != "claim" else {}) + if command in {"claim", "renew"}: + payload["ttl_seconds"] = 300 + with pytest.raises(RuntimeError, match="caller abort"): + async with maker.begin() as db: + result = await manage_claim( + PayoutDeliveryRepository(db), + command, + model.model_validate_json(json.dumps(payload)), + ) + assert result["verdict"] in {"claimed", "renewed", "released", "recovered"} + if command == "renew": + assert result["expires_at"] > before[1]["expires_at"] + raise RuntimeError("caller abort") + assert await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}) == before + + +@pytest.mark.parametrize("rollback", [False, True]) +async def test_parent_nowait_independent_intents(claims_db, allocation, rollback): + maker, _, _, key = claims_db + async with maker.begin() as db: + other = await admit_intent( + PayoutDeliveryRepository(db), allocation.model_copy(update={"allocation_id": "other"}) + ) + async with maker() as first, maker() as second: + request = ClaimRequest.model_validate_json(json.dumps(key | {"ttl_seconds": 300})) + await manage_claim(PayoutDeliveryRepository(first), "claim", request) + pid1 = await first.scalar(text("SELECT pg_backend_pid()")) + pid2 = await second.scalar(text("SELECT pg_backend_pid()")) + assert pid1 != pid2 + with pytest.raises(DBAPIError) as error: + await asyncio.wait_for( + manage_claim(PayoutDeliveryRepository(second), "claim", request), 3 + ) + assert getattr(error.value.orig, "sqlstate") == "55P03" + await second.rollback() + assert await invoke_claim(maker, "claim", key | {"ttl_seconds": 300}) == ( + 9, + {"verdict": "busy"}, + ) + assert ( + await invoke_claim( + maker, "claim", key | {"intent_id": other["intent_id"], "ttl_seconds": 300} + ) + )[0] == 0 + await first.rollback() if rollback else await first.commit() + code, result = await invoke_claim(maker, "claim", key | {"ttl_seconds": 300}) + assert code == (0 if rollback else 11) + if rollback: + assert result["epoch"] == 1 + + +async def test_access_exclusive_lock_budget(claims_db): + maker, _, _, key = claims_db + async with maker() as holder, maker() as waiter: + waiting_pid = await waiter.scalar(text("SELECT pg_backend_pid()")) + await holder.execute(text("LOCK TABLE payout_delivery_intents IN ACCESS EXCLUSIVE MODE")) + holding_pid = await holder.scalar(text("SELECT pg_backend_pid()")) + assert waiting_pid != holding_pid + request = ClaimRequest.model_validate_json(json.dumps(key | {"ttl_seconds": 300})) + task = asyncio.create_task(manage_claim(PayoutDeliveryRepository(waiter), "claim", request)) + try: + async with asyncio.timeout(3): + while True: + blockers = await holder.scalar( + text("SELECT pg_blocking_pids(:pid)"), {"pid": waiting_pid} + ) + if holding_pid in blockers: + break + await asyncio.sleep(0) + assert ( + await holder.scalar( + text("SELECT count(*) FROM pg_locks WHERE pid = :pid AND NOT granted"), + {"pid": waiting_pid}, + ) + > 0 + ) + with pytest.raises(DBAPIError) as error: + await asyncio.wait_for(task, 3) + assert getattr(error.value.orig, "sqlstate") == "55P03" + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await waiter.rollback() + await holder.rollback() + async with maker.begin() as db: + await PayoutDeliveryRepository(db).claim_budget() + assert await db.scalar(text("SHOW lock_timeout")) == "250ms" + assert await db.scalar(text("SHOW statement_timeout")) == "1s" + async with maker() as db: + assert await db.scalar(text("SHOW lock_timeout")) == "0" + + +@pytest.mark.parametrize( + "statement", + [ + "DELETE FROM payout_delivery_claims", + "TRUNCATE payout_delivery_claims", + "UPDATE payout_delivery_claims SET intent_id = gen_random_uuid()", + "UPDATE payout_delivery_claims SET epoch = 0", + "UPDATE payout_delivery_claims SET epoch = epoch + 1", + "UPDATE payout_delivery_claims SET worker_id = gen_random_uuid()", + "UPDATE payout_delivery_claims SET expires_at = expires_at - interval '1 second'", + ], +) +async def test_claim_dml_guards(claims_db, statement): + maker, _, _, key = claims_db + await invoke_claim(maker, "claim", key | {"ttl_seconds": 300}) + with pytest.raises(DBAPIError): + async with maker.begin() as db: + await db.execute(text(statement)) + + +async def test_ready_guards_and_repeatable_replay(claims_db): + maker, _, _, key = claims_db + _, claimed = await invoke_claim(maker, "claim", key | {"ttl_seconds": 300}) + async with maker.begin() as db: + conn = await db.connection() + raw = await conn.get_raw_connection() + assert raw.driver_connection is not None + for name in ( + "V112__payout_delivery_claims.sql", + "R__payout_delivery_claim_identity.sql", + "R__payout_delivery_claim_identity.sql", + ): + await raw.driver_connection.execute((MIGRATIONS / name).read_text()) + _, stored = await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}) + assert stored == claimed | {"verdict": "found"} + await invoke_claim(maker, "release", key | {"epoch": 1}) + for change in ( + "state = state", + "state = 'claimed', epoch = 3, worker_id = " + "gen_random_uuid(), expires_at = clock_timestamp()", + ): + with pytest.raises(DBAPIError): + async with maker.begin() as db: + await db.execute(text("UPDATE payout_delivery_claims SET " + change)) + assert (await invoke_claim(maker, "claim", key | {"ttl_seconds": 300}))[1]["epoch"] == 2 + async with maker.begin() as db: + await db.execute(text("DROP TABLE payout_delivery_claims")) + assert await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}) == ( + 8, + {"verdict": "storage_error"}, + ) + with pytest.raises(UndefinedTableError): + async with maker.begin() as db: + conn = await db.connection() + raw = await conn.get_raw_connection() + assert raw.driver_connection is not None + await raw.driver_connection.execute( + (MIGRATIONS / "R__payout_delivery_claim_identity.sql").read_text() + ) + + +@pytest.mark.parametrize( + "epoch,state", [(2, "claimed"), (9223372036854775807, "claimed"), (1, "ready")] +) +async def test_insert_generation_guards(claims_db, epoch, state): + maker, _, _, key = claims_db + with pytest.raises(DBAPIError): + async with maker.begin() as db: + await db.execute( + text( + "INSERT INTO payout_delivery_claims VALUES " + "(:intent_id, :epoch, :state, :worker_id, clock_timestamp())" + ), + key | {"epoch": epoch, "state": state}, + ) + + +@pytest.mark.parametrize( + "sqlstate,expected", + [("23514", (8, {"verdict": "storage_error"})), ("57014", (9, {"verdict": "busy"}))], +) +async def test_real_commit_failure_suppresses_fence(claims_db, sqlstate, expected, capsys): + maker, _, _, key = claims_db + async with maker.begin() as db: + await db.execute( + text( + "CREATE FUNCTION fail_claim_commit() RETURNS trigger AS $$ " + "BEGIN RAISE EXCEPTION 'private-marker' USING ERRCODE = '" + + sqlstate + + "'; END; $$ LANGUAGE plpgsql" + ) + ) + await db.execute( + text( + "CREATE CONSTRAINT TRIGGER " + "fail_claim_commit AFTER INSERT ON payout_delivery_claims " + "DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION fail_claim_commit()" + ) + ) + assert await invoke_claim(maker, "claim", key | {"ttl_seconds": 1}) == expected + assert await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}) == ( + 6, + {"verdict": "not_found"}, + ) + assert capsys.readouterr().err == "" + + +@pytest.mark.parametrize( + "command,change", + [("claim", {"ttl_seconds": v}) for v in [True, 1.5, 0, 301]] + + [("renew", {"epoch": v}) for v in [True, 1.5, 0, 9223372036854775808]] + + [("claim", {"intent_id": "private-marker"}), ("claim", {"rpc_endpoint": "private-marker"})], +) +async def test_strict_private_input(command, change): + key = {"intent_id": str(uuid4()), "worker_id": str(uuid4()), "ttl_seconds": 1} + if command == "renew": + key["epoch"] = 1 + assert await invoke_claim(None, command, key | change) == (5, {"verdict": "invalid_input"}) + + +async def test_subprocess_all_commands(claims_db): + maker, url, schema, key = claims_db + await seed_claim(maker, key, True) + scoped = make_url(url).update_query_dict({"dsn": "postgresql:///?search_path=" + schema}) + env = os.environ | {"DATABASE_URL": scoped.render_as_string(hide_password=False)} + commands = [ + ("recover", key | {"epoch": 1}, "recovered"), + ("claim", key | {"ttl_seconds": 300}, "claimed"), + ("lease-show", {"intent_id": key["intent_id"]}, "found"), + ("renew", key | {"epoch": 2, "ttl_seconds": 1}, "renewed"), + ("release", key | {"epoch": 2}, "released"), + ] + for command, payload, verdict in commands: + process = await asyncio.create_subprocess_exec( + sys.executable, + "-m", + "app.cli.payout_delivery", + command, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=env, + ) + stdout, stderr = await asyncio.wait_for( + process.communicate(json.dumps(payload).encode()), 20 + ) + assert process.returncode == 0 and stderr == b"" + result = json.loads(stdout) + assert result["verdict"] == verdict and result["intent_id"] == key["intent_id"] + assert (await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}))[1][ + "state" + ] == "ready" + + +@pytest.mark.parametrize("command", ["claim", "renew", "release", "recover", "lease-show"]) +async def test_missing_parent(claims_db, command): + maker, _, _, key = claims_db + payload = key | {"intent_id": str(uuid4())} + if command != "claim": + payload["epoch"] = 1 + if command in {"claim", "renew"}: + payload["ttl_seconds"] = 1 + if command == "lease-show": + payload = {"intent_id": payload["intent_id"]} + assert await invoke_claim(maker, command, payload) == (6, {"verdict": "not_found"}) + + +async def test_expired_claim_requires_explicit_recovery(claims_db): + maker, _, _, key = claims_db + await seed_claim(maker, key, True) + assert await invoke_claim(maker, "claim", key | {"ttl_seconds": 1}) == ( + 11, + {"verdict": "unavailable"}, + ) + _, stored = await invoke_claim(maker, "lease-show", {"intent_id": key["intent_id"]}) + assert stored["state"] == "claimed" and stored["epoch"] == 1 + assert stored["worker_id"] == key["worker_id"] + + +@pytest.mark.parametrize("state", ["55P03", "57014"]) +async def test_existing_admission_errors_keep_storage_verdict(delivery_db, allocation, state): + """Preserve submit failure codes while claim contention gains its own busy result.""" + maker, _, _ = delivery_db + async with maker.begin() as db: + await db.execute( + text( + "CREATE FUNCTION fail_existing_intake_commit() RETURNS trigger AS $$ " + "BEGIN RAISE EXCEPTION 'synthetic-private-marker' USING ERRCODE = '" + + state + + "'; END; $$ LANGUAGE plpgsql" + ) + ) + await db.execute( + text( + "CREATE CONSTRAINT TRIGGER fail_existing_intake_commit " + "AFTER INSERT ON payout_delivery_intents DEFERRABLE INITIALLY DEFERRED " + "FOR EACH ROW EXECUTE FUNCTION fail_existing_intake_commit()" + ) + ) + output = io.StringIO() + assert ( + await run_admission("submit", io.StringIO(allocation.model_dump_json()), output, maker) == 8 + ) + assert json.loads(output.getvalue()) == {"verdict": "storage_error"} + async with maker() as db: + assert await db.scalar(text("SELECT count(*) FROM payout_delivery_intents")) == 0 diff --git a/db/migrations/R__payout_delivery_claim_identity.sql b/db/migrations/R__payout_delivery_claim_identity.sql new file mode 100644 index 00000000..abb9bc6a --- /dev/null +++ b/db/migrations/R__payout_delivery_claim_identity.sql @@ -0,0 +1,24 @@ +-- INVARIANT(189): retain generations permanently; ownership cannot change within a live epoch. +CREATE OR REPLACE FUNCTION protect_payout_delivery_claim() RETURNS TRIGGER AS $$ +BEGIN + IF TG_OP = 'INSERT' THEN + IF NEW.state = 'claimed' AND NEW.epoch = 1 THEN RETURN NEW; END IF; + ELSIF TG_OP = 'UPDATE' AND NEW.intent_id = OLD.intent_id THEN + IF (OLD.state = 'ready' AND NEW.state = 'claimed' AND NEW.epoch = OLD.epoch + 1) + OR (OLD.state = 'claimed' AND NEW.state = 'claimed' AND NEW.epoch = OLD.epoch + AND NEW.worker_id = OLD.worker_id AND OLD.expires_at > clock_timestamp() + AND NEW.expires_at >= OLD.expires_at) + OR (OLD.state = 'claimed' AND NEW.state = 'ready' AND NEW.epoch = OLD.epoch) + THEN RETURN NEW; END IF; + END IF; + RAISE EXCEPTION 'Payout claim transition is invalid' USING ERRCODE = '23514'; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE TRIGGER payout_claim_identity + BEFORE INSERT OR UPDATE OR DELETE ON payout_delivery_claims + FOR EACH ROW EXECUTE FUNCTION protect_payout_delivery_claim(); + +CREATE OR REPLACE TRIGGER payout_claim_no_truncate + BEFORE TRUNCATE ON payout_delivery_claims + FOR EACH STATEMENT EXECUTE FUNCTION protect_payout_delivery_claim(); diff --git a/db/migrations/V112__payout_delivery_claims.sql b/db/migrations/V112__payout_delivery_claims.sql new file mode 100644 index 00000000..11f2d1b0 --- /dev/null +++ b/db/migrations/V112__payout_delivery_claims.sql @@ -0,0 +1,9 @@ +CREATE TABLE IF NOT EXISTS payout_delivery_claims ( + intent_id UUID PRIMARY KEY REFERENCES payout_delivery_intents(intent_id) ON DELETE RESTRICT, + epoch BIGINT NOT NULL CHECK (epoch >= 1), + state TEXT NOT NULL CHECK (state IN ('ready', 'claimed')), + worker_id UUID, + expires_at TIMESTAMPTZ, + CHECK ((state = 'ready' AND worker_id IS NULL AND expires_at IS NULL) + OR (state = 'claimed' AND worker_id IS NOT NULL AND expires_at IS NOT NULL)) +); diff --git a/docs/PAYOUT_RECEIPTS.md b/docs/PAYOUT_RECEIPTS.md index 2dccca07..3ca60a51 100644 --- a/docs/PAYOUT_RECEIPTS.md +++ b/docs/PAYOUT_RECEIPTS.md @@ -1,8 +1,8 @@ -2026-11-05 · CLI contract · Issues #187 and #188 +2026-11-05 · CLI contract · Issues #187, #188 and #189 -# Payout receipts and immutable intents +# Payout receipts, immutable intents and leases -The no-argument command compares an explicit payout intent with a trusted Solana RPC response without writing data. The `submit` and `show` commands store and read immutable intents in PostgreSQL. A matching receipt describes the trusted response; it does not independently prove a real payment, create an entitlement, or send funds. Automatic activity accrual and the durable sender remain incomplete. +The no-argument command compares an explicit payout intent with a trusted Solana RPC response without writing data. The `submit` and `show` commands store and read immutable intents in PostgreSQL. Claim commands coordinate exact intent UUIDs with retained epochs; they do not send payments. A matching receipt describes the trusted response; it does not independently prove a real payment, create an entitlement, or send funds. Automatic activity accrual and the durable sender remain incomplete. | When | Input | Output | Excluded | |---|---|---|---| @@ -56,3 +56,25 @@ R188.4: Exit codes are 0 for accepted/replayed/found, 5 for invalid input, 6 for R188.5: The additive table migration is `db/migrations/V111__payout_delivery.sql`; immutability functions and triggers are installed by `db/migrations/R__payout_delivery_identity.sql`. Existing receipt-only applications ignore this table. Application rollback retains admitted data and the new table; no automatic drop or down migration is provided. Reproduce synthetic PostgreSQL intake QA with `SECRET_KEY=ci-placeholder-value uv run --frozen python -m pytest -q tests/test_payout_delivery_db.py`; preserve generated evidence outside Git. These checks do not establish entitlement, live chain activity or production mint compatibility. PostgreSQL 16 documents the [separate READ COMMITTED command snapshots](https://www.postgresql.org/docs/16/transaction-iso.html) and [conflict-safe insertion and RETURNING](https://www.postgresql.org/docs/16/sql-insert.html). + +## Exact-intent leases (issue #189) + +R189.1: A privileged backend operator supplies a committed intent UUID and a worker UUID. Worker identity coordinates processing; it does not authenticate a signer. The existing CLI adds these JSON commands without changing receipt verification or immutable admission. + +| Command | Required JSON fields | Successful verdict | +|---|---|---| +| `claim` | `intent_id`, `worker_id`, `ttl_seconds` | `claimed` | +| `renew` | `intent_id`, `worker_id`, `epoch`, `ttl_seconds` | `renewed` | +| `release` | `intent_id`, `worker_id`, `epoch` | `released` | +| `recover` | `intent_id`, `worker_id`, `epoch` | `recovered` | +| `lease-show` | `intent_id` | `found` | + +R189.2: UUIDs are required; epoch is a strict integer from 1 to 9223372036854775807, TTL from 1 to 300 seconds. TTL is an operational lease duration, not reward cadence. Unknown fields, bools and floats are rejected. Success exits 0 and returns only `verdict`, `intent_id`, `state`, `epoch`, `worker_id` and UTC ISO8601 `expires_at`; ready rows have null worker/expiry. Use `lease-show` after a lost claim response to recover the persisted fence. + +R189.3: Initial claim is epoch 1; a new claim from ready increments it, including for the same worker. Claimed rows are unavailable even when expired. Renew/release require the exact live fence; recover requires the exact expired fence. Renew never shortens or revives a lease. The server clock governs expiry. Mutators lock the exact immutable parent before the child; independent intents do not share a slot. Ordinary invalid transitions and deletion/truncation are rejected; database owners can alter these protections. + +R189.4: Each operation installs transaction-local lock_timeout=250ms and statement_timeout=1000ms. These bound individual database waits/statements, not connection setup or total process duration. Success is emitted after commit. Error output contains only a verdict: invalid_input=5, not_found=6, storage_error=8, busy=9, stale=10, unavailable=11. Existing conflict=7 remains. Service/repository callers own rollback; the configured backend bootstrap prerequisite still applies. + +R189.5: No automatic worker, signed attempt, prepared/finalized state, broadcast or payment confirmation exists here. A later signed-attempt stage must atomically persist the exact attempt and leave claimed under its live fence before broadcast. Current release/recover filter only claimed. Backlog selection, signing configuration, sender, reconciliation and automatic rewards remain incomplete. Synthetic verification evidence stays outside Git; it does not establish any real payout or signed-attempt guarantee. + +R189.6: The additive claim migration is `db/migrations/V112__payout_delivery_claims.sql`; the installed V111 intent table and its repeatable guards are retained. Application rollback retains the child relation and its monotonic generations. PostgreSQL16 defines [the advancing server clock](https://www.postgresql.org/docs/16/functions-datetime.html) and [row-lock NOWAIT behavior](https://www.postgresql.org/docs/16/sql-select.html); relation-lock waits additionally use the stated local timeouts.