Stack/graceful deletion/03 payment drain

docs/plans/2026-08-10-graceful-content-lock-deletion.md

@@ -14,7 +14,7 @@
 
 ## Status and provenance
 
-- Plan status: **accepted product design; Tasks 1–5 committed and Task 6 implemented pending commit; Tasks 7–10 remain**.
+- Plan status: **accepted product design; Tasks 1, 2, 4, 5, and 6 committed; the omitted Task 3 prerequisite is implemented pending commit; Tasks 7–10 remain**.
 - Repository inspected: `/home/u/Projects/Synonym/Pubky/locks-public`.
 - Planning base when written: clean `master` at `ba49a77`.
 - There has been no production deployment. New persistence may require a clean pre-production database; no historical backfill is required.
@@ -153,11 +153,12 @@ Both drain endpoints return `200` with the same closed aggregate body:
   "status": "active",
   "accepted_count": 0,
   "terminal_count": 0,
-  "cancellation_enqueued_count": 0
+  "cancellation_enqueued_count": 0,
+  "cleanup_token": "<43-character-unpadded-base64url>"
 }
 ```
 
-`status` is exactly `active` or `completed`. The response contains no drain ID, replay flag, Bundle ID, reader, Payment Request ID, address, payment reference, or raw error. Exact replay returns the same aggregate body.
+`status` is exactly `active` or `completed`. The response contains no drain ID, replay flag, Bundle ID, reader, Payment Request ID, address, payment reference, or raw error. Replay preserves the frozen drain identity, cancellation count, and cleanup token while returning its latest monotonic aggregate progress.
 
 ### Per-Bundle status
 
@@ -200,6 +201,8 @@ The canonical persisted `request_state` is one of these exact closed snake-case
 
 `expired` is returned when the invoice has a durable `payment_expired_at`; otherwise the persisted observation state maps one-to-one to `undetected`, `detected`, or `confirmed`. `confirmations` and `amount_matched` remain orthogonal factual fields.
 
+Locks maps `rejected`, `canceled`, and `proposal_expired` requests to `VerificationTaskStatus::Expired`. An `accepted` request whose `payment_state` is `expired` also maps to `Expired`. These terminal outcomes carry no failure message and never map to `Failed`.
+
 Drain classification uses the persisted state without inference from invoice delivery or Bitcoin observation:
 
 - `accepted` is accepted and blocking;
@@ -420,14 +423,21 @@ cargo test --workspace --no-run
 - Modify: `locks-server/src/paykit_http_client.rs`
 - Modify: `locks-server/src/app_state/mod.rs`
 - Create: `locks-service/src/application/ports/payment_drain.rs`
+- Create: `locks-service/src/application/ports/payment_drain_repository.rs`
 - Create: `locks-service/src/application/use_cases/drain_lock_payments.rs`
+- Create: `locks-service/src/infrastructure/postgres/payment_drains.rs`
+- Create: `locks-service/migrations/0015_content_lock_payment_drains.sql`
+- Modify: `locks-service/src/infrastructure/postgres/content_lock_deletions.rs`
+- Modify: `locks-service/src/infrastructure/postgres/verification_task_claims.rs`
 - Test: `locks-server/src/paykit_http_client.rs`
 - Test: deletion use-case tests and HTTP integration fixtures
 
 **Dependency gate:** Patch both plans with exact per-Bundle enums, error mappings, and drain-cleanup route before RED tests. Then implement Paykit Server routes first.
 
 **RED:** Test exact signed JSON, no `minimum_confirmations` leak, aggregate redaction, local application of confirmations, canceled/rejected/expired transitions, timely matched confirmation continuation, and retryable transport errors.
 
+**GREEN:** Freeze Paykit obligation identity, criterion, cutoff status, and authoritative invoice window in the deletion snapshot; persist the opaque drain token and aggregate under the deletion claim fence. Paykit aggregate progress is monotonic: `accepted_count` may only decrease to zero, `terminal_count` may only increase by the same amount, and `cancellation_enqueued_count` plus the opaque cleanup token remain immutable. `completed` requires `accepted_count == 0` and cannot regress to `active`. Locks persists each newer aggregate under the live deletion lease, but still requires every frozen local obligation to become terminal before phase advancement; aggregate completion never bypasses Locks-local confirmation or reorg reconciliation. Exclude ordinary verification workers for every surviving deletion snapshot. Immediately before external entitlement storage, an ordinary claimed worker durably marks entitlement publication under its exact lease. Deletion admission locks and owns every matching task row before snapshot/reset: a committed publication marker blocks deletion, while committed deletion ownership blocks publication and every ordinary claim/retry/terminal write. Ambiguous entitlement publication retains the marker until an ordinary retry reconciles an equivalent entitlement and terminalizes the task. Publish entitlements before terminalizing their task and reconcile only an equivalent existing entitlement. Compose the client and repository in `AppState`. Task 9 supplies the queue polling/supervision that invokes this phase use case.
+
 **Suggested commit:** `feat(paykit): drain deleting lock payments`
 
 ### Task 8: Implement final credential/read draining

locks-e2e/tests/creator_publishing_http.rs

@@ -827,6 +827,7 @@ impl FakePaykitServer {
             Some(json!({
                 "bundle_id": BUNDLE_ID,
                 "lock_resource": lock_resource,
+                "payment_in": 24,
                 "reader": creator().to_string(),
             }))
         );
@@ -872,14 +873,20 @@ async fn fake_invoice_handler(
     State(state): State<Arc<Mutex<FakePaykitState>>>,
     headers: HeaderMap,
     body: Bytes,
-) -> StatusCode {
+) -> (StatusCode, Json<serde_json::Value>) {
     let mut state = state.lock().await;
     state.invoice_count += 1;
     state.invoice_body = Some(serde_json::from_slice(&body).unwrap());
     state.invoice_signature = headers
         .get("X-Paykit-Signature")
         .map(|value| value.to_str().unwrap().to_owned());
-    StatusCode::CREATED
+    (
+        StatusCode::CREATED,
+        Json(json!({
+            "invoice_created_at": "2026-08-12T10:00:00Z",
+            "payment_deadline": "2026-08-13T10:00:00Z",
+        })),
+    )
 }
 
 async fn fake_status_handler(

locks-e2e/tests/postgres_runtime.rs

@@ -10,7 +10,7 @@ use axum::extract::ConnectInfo;
 use axum::http::{Request, StatusCode, header};
 use axum::routing::post;
 use locks_core::ids::{
-    BundleId, CreatorPubky, GuardedResourceHash, LockServerPubky, PubkyLockResource,
+    BundleId, CreatorPubky, GuardedResourceHash, LockServerPubky, PubkyLockResource, TaskId,
 };
 use locks_core::lock_policy::{
     AccessPolicy, CONTENT_LOCK_VERSION, ContentLock, Criterion, GuardedResource, LockLogic,
@@ -29,16 +29,19 @@ use locks_server::worker::{VerificationWorker, WorkerTick};
 use locks_service::application::models::{
     AccessCredential, AccessCredentialLookupKey, ContentLockDeletionJob,
     ContentLockOwnershipStatus, CreatorAuthorityAuthKind, CreatorAuthorityRecord,
-    CreatorAuthoritySecret, VerificationTaskStatus,
+    CreatorAuthoritySecret, VerificationTaskRecord, VerificationTaskStatus,
+};
+use locks_service::application::ports::{
+    ContentLockDeletionRepository, VerificationTaskRepository,
 };
-use locks_service::application::ports::ContentLockDeletionRepository;
 use locks_service::infrastructure::memory::{
     content_locks::InMemoryContentLockRepository, entitlements::InMemoryEntitlementRepository,
     guarded_resources::InMemoryGuardedResourceRepository,
     lock_service_pointers::InMemoryLockServicePointerRepository,
 };
 use locks_service::infrastructure::postgres::{
-    CreatorAuthoritySecretCipher, PostgresContentLockDeletionRepository, run_migrations,
+    CreatorAuthoritySecretCipher, PostgresContentLockDeletionRepository,
+    PostgresVerificationTaskRepository, run_migrations,
 };
 use serde_json::{Value, json};
 use sqlx::postgres::PgPoolOptions;
@@ -111,6 +114,61 @@ async fn postgres_runtime_state_survives_app_state_recreation() {
     database.cleanup().await;
 }
 
+#[tokio::test]
+async fn manual_completion_hides_legacy_paykit_admission_without_authoritative_window() {
+    let Some(database) = TestDatabase::create().await else {
+        return;
+    };
+    let submitted = paykit_submission_for(&paykit_content_lock());
+    let task = VerificationTaskRecord {
+        task_id: TaskId::from_str(&uuid::Uuid::new_v4().to_string()).unwrap(),
+        creator: submitted.pubky_lock_resource.creator().clone(),
+        submitted_proof_bundle: submitted.clone(),
+        status: VerificationTaskStatus::Pending,
+        submitted_at: datetime!(2026-08-12 06:00:00 UTC),
+        started_at: None,
+        completed_at: None,
+        failure_message: None,
+    };
+    PostgresVerificationTaskRepository::new(database.pool().clone())
+        .insert_verification_task(task.clone())
+        .await
+        .unwrap();
+    sqlx::query(
+        "INSERT INTO paykit_task_admissions
+             (verification_task_id, ready, ready_at)
+         VALUES ($1::uuid, TRUE, now())",
+    )
+    .bind(task.task_id.to_string())
+    .execute(database.pool())
+    .await
+    .unwrap();
+
+    let response = router(app_state(database.pool().clone()))
+        .oneshot(json_request(
+            "POST",
+            "/verification-task-completions",
+            json!({
+                "creator": submitted.pubky_lock_resource.creator(),
+                "bundle_id": submitted.bundle_id,
+            }),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(response.status(), StatusCode::NOT_FOUND);
+    assert_eq!(
+        response_json(response).await,
+        json!({
+            "error": {
+                "code": "verification_task_not_found",
+                "message": "verification task not found"
+            }
+        })
+    );
+
+    database.cleanup().await;
+}
+
 #[tokio::test]
 async fn postgres_runtime_readyz_returns_ready_without_leaking_runtime_details() {
     let Some(database) = TestDatabase::create().await else {
@@ -275,9 +333,18 @@ async fn snapshotted_unready_paykit_replay_ignores_tombstoned_lock_and_reader_re
             let call = paykit_state.fetch_add(1, Ordering::SeqCst);
             async move {
                 if call == 0 {
-                    StatusCode::INTERNAL_SERVER_ERROR
+                    (
+                        StatusCode::INTERNAL_SERVER_ERROR,
+                        axum::Json(json!({ "error": "injected" })),
+                    )
                 } else {
-                    StatusCode::OK
+                    (
+                        StatusCode::OK,
+                        axum::Json(json!({
+                            "invoice_created_at": "2026-08-12T10:00:00Z",
+                            "payment_deadline": "2026-08-13T10:00:00Z",
+                        })),
+                    )
                 }
             }
         }),

locks-server/src/api/verification.rs

@@ -20,7 +20,9 @@ use locks_service::application::use_cases::submit_proof_bundle::{
 use locks_service::application::use_cases::validate_paykit_payment_submission::{
     ValidatePaykitPaymentSubmissionRequest, ValidatePaykitPaymentSubmissionUseCase,
 };
-use locks_service::infrastructure::postgres::PostgresPaykitTaskAdmissionRepository;
+use locks_service::infrastructure::postgres::{
+    PaykitInvoiceWindow, PostgresPaykitTaskAdmissionRepository,
+};
 use locks_service::infrastructure::verifiers::registry::StaticCriterionVerifierRegistry;
 
 use crate::api::dtos::{
@@ -116,13 +118,20 @@ async fn maybe_prepare_paykit_submission(
         let admissions = PostgresPaykitTaskAdmissionRepository::new(pool.clone());
         if let Some(admission) = admissions.find_existing(submitted).await? {
             if admission.requires_paykit {
-                create_paykit_invoice(state, &admission.task.submitted_proof_bundle).await?;
-                admissions.mark_ready(&admission.task).await?;
+                let invoice_window = create_paykit_invoice(
+                    state,
+                    &admission.task.submitted_proof_bundle,
+                    admission.payment_in,
+                )
+                .await?;
+                admissions
+                    .mark_ready(&admission.task, invoice_window)
+                    .await?;
             }
             return Ok(Some(admission.task.into()));
         }
     }
-    ValidatePaykitPaymentSubmissionUseCase::new(state.content_locks().as_ref())
+    let validated = ValidatePaykitPaymentSubmissionUseCase::new(state.content_locks().as_ref())
         .execute(ValidatePaykitPaymentSubmissionRequest {
             submitted_proof_bundle: submitted.clone(),
         })
@@ -140,10 +149,17 @@ async fn maybe_prepare_paykit_submission(
     if let Some(pool) = state.postgres_pool() {
         let task = submit_use_case.prepare_task(submitted.clone()).await?;
         let admissions = PostgresPaykitTaskAdmissionRepository::new(pool.clone());
-        let admission = admissions.reserve(task).await?;
+        let admission = admissions.reserve(task, validated.payment_in).await?;
         if admission.requires_paykit {
-            create_paykit_invoice(state, &admission.task.submitted_proof_bundle).await?;
-            admissions.mark_ready(&admission.task).await?;
+            let invoice_window = create_paykit_invoice(
+                state,
+                &admission.task.submitted_proof_bundle,
+                admission.payment_in,
+            )
+            .await?;
+            admissions
+                .mark_ready(&admission.task, invoice_window)
+                .await?;
         }
         return Ok(Some(admission.task.into()));
     }
@@ -160,6 +176,7 @@ async fn maybe_prepare_paykit_submission(
         .create_invoice(&PaykitInvoiceRequest {
             bundle_id: submitted.bundle_id.to_string(),
             lock_resource: submitted.pubky_lock_resource.to_string(),
+            payment_in: validated.payment_in,
             reader: reader.to_string(),
         })
         .await
@@ -170,14 +187,15 @@ async fn maybe_prepare_paykit_submission(
 async fn create_paykit_invoice(
     state: &AppState,
     submitted: &SubmittedProofBundle,
-) -> Result<(), ApiError> {
+    payment_in: u64,
+) -> Result<PaykitInvoiceWindow, ApiError> {
     let reader = submitted.reader_public_key.as_ref().ok_or_else(|| {
         ApiError::new(
             ApiErrorCode::InvalidRequest,
             "paykit-payment requires reader_public_key",
         )
     })?;
-    state
+    let response = state
         .paykit_http_client()
         .ok_or_else(|| {
             ApiError::new(
@@ -188,10 +206,15 @@ async fn create_paykit_invoice(
         .create_invoice(&PaykitInvoiceRequest {
             bundle_id: submitted.bundle_id.to_string(),
             lock_resource: submitted.pubky_lock_resource.to_string(),
+            payment_in,
             reader: reader.to_string(),
         })
         .await
-        .map_err(map_paykit_invoice_error)
+        .map_err(map_paykit_invoice_error)?;
+    Ok(PaykitInvoiceWindow {
+        invoice_created_at: response.invoice_created_at,
+        payment_deadline: response.payment_deadline,
+    })
 }
 
 fn map_paykit_invoice_error(error: PaykitClientError) -> ApiError {