Fence payout intent claims with persistent leases

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"]]
 
 

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()"
+        )

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

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,
+    }

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()
     ]

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"