worker: autonomously reclaim expired File imports through bounded retries

applications/worker.py

@@ -57,7 +57,7 @@ def complete_noop(
 
 
 class FileDispatchAuthority(Protocol):
-    """Application port for database-selected first-attempt File work."""
+    """Application port for database-selected bounded-attempt File work."""
 
     def claim(self) -> FileDispatchLease | FileDispatchNoWork: ...
 

docs/decisions/0060-reclaim-expired-file-imports-with-bounded-retries.md

@@ -0,0 +1,73 @@
+---
+name: adr-0060-reclaim-expired-file-imports-with-bounded-retries
+version: "1.0.0"
+description: >
+  Reclaim expired scheduler-owned File imports with database-timed bounded
+  backoff while preserving the existing exact WorkerLease recovery path.
+---
+
+# 0060. Reclaim expired File imports with bounded retries
+
+- Status: accepted
+- Date: 2026-07-26
+- Refines: ADR-0029, ADR-0041, ADR-0043, ADR-0055, ADR-0059
+
+## Context
+
+Autonomous File dispatch can stop after leasing or between the acquired,
+prepared, and indexed publication boundaries. The durable job retains enough
+lineage to resume, but ADR-0059 deliberately leaves every expired scheduler
+lease untouched. Without a bounded reclaim owner, one ordinary process stop can
+leave an accepted upsert unpublished indefinitely. Letting the process choose a
+tenant, job, retry time, or generation would weaken the function-only scheduler
+boundary and duplicate the existing WorkerLease authority.
+
+## Decision
+
+The existing scheduler claim function first considers expired scheduler-owned
+File imports, then delegates to its unchanged first-attempt selector. PostgreSQL
+selects one globally eligible recovery with `FOR UPDATE SKIP LOCKED`, using the
+database clock and a fixed exponential delay of 30, 60, then 120 seconds after
+lease expiry. It may mint only generations two through four, so one first
+attempt has at most three automatic reclaims. Generation four remains untouched
+after expiry; this decision adds no dead-letter transition or operator requeue.
+
+Recovery revalidates the same exact current page, upsert observation,
+SourceVersion/root capability, Membership version, receiver, and scan epoch as
+first dispatch. It also holds the existing per-Source progress lock and advances
+only the locked job to the exact next generation. The transaction preserves the
+durable recovery boundary, clears prior redemption, replaces the nonce and
+database timestamps, and appends one immutable digest-only `reclaimed` event.
+The scheduler still has no table access and receives only the existing closed
+lease/no-work result. A missing configured root or failed current authority
+check yields no work and no mutation.
+
+The independent Supply worker consumes the replacement WorkerLease through the
+existing publication-recovery state machine. Every older token is stale because
+generation, nonce, and current job state are checked together. No new process,
+queue, Runtime authority, or content-bearing retry path is introduced.
+
+## Consequences
+
+- Concurrent schedulers can advance an expired generation only once.
+- Recovery work is preferred over fresh first attempts once its database-owned
+  delay has elapsed; ineligible or exhausted work does not block fresh work.
+- A crash at any durable publication boundary resumes toward one active Revision
+  and one publication effect; the replaced lease has zero effect.
+- Downgrade is refused while a scheduler-owned higher-generation lease remains,
+  because the previous schema cannot preserve its automatic-reclaim provenance.
+- Retry exhaustion remains durable but operationally unresolved.
+- The supporting partial-index build remains transactional with the migration
+  fences and therefore takes a relation lock. Before a production rollout, the
+  operator must measure the retained `file_import_job` population and index-build
+  lock duration against the deployment drain budget, using a maintenance window
+  when the measured bound is not acceptable.
+
+## Revisit trigger
+
+Revisit before adding terminal-failure retries, dead-letter state, exhaustion
+transitions, operator remediation/requeue, configurable backoff, provider
+polling, automatic page acceptance, or automatic upsert/delete ordering. Any
+revision must preserve database-selected tenant routing, exact-generation lease
+fencing, current authority revalidation, content-free scheduler results, and one
+publication authority.

docs/decisions/README.md

@@ -49,6 +49,7 @@ kernel, capability separation, and publication visibility model.
 | File delete observation execution | [0057 — Execute current File deletes through tombstone authority](0057-execute-current-file-deletes-through-tombstone-authority.md) | One trusted exact locator is revalidated against the current complete v4 scan, derives effect identity server-side, calls the sole tombstone authority, and records one immutable exact binding | Provider deletion authority, caller-authored path/effect identity, stale scan deletion, a second visibility authority, or Runtime trust in Supply metadata |
 | Mixed File page upsert scheduling | [0058 — Schedule only upserts from mixed File pages](0058-schedule-only-upserts-from-mixed-file-pages.md) | Validate one complete current v4 page, then atomically schedule and replay its nonempty ordered upsert projection through the existing import path while preserving original ordinals | Scheduler delete/tombstone authority, implicit audience, automatic upsert/delete ordering, a second queue, or Runtime trust in Supply metadata |
 | Autonomous File dispatch | [0059 — Dispatch scheduled File imports through exact leases](0059-dispatch-scheduled-file-imports-through-exact-leases.md) | A function-only scheduler login atomically claims the oldest current page-scheduled upsert and mints the existing exact first-attempt WorkerLease | Caller tenant/job routing, broad Control credentials, direct scheduler table access, retry/reclaim, delete execution, or a second queue/process |
+| Bounded File reclaim | [0060 — Reclaim expired File imports with bounded retries](0060-reclaim-expired-file-imports-with-bounded-retries.md) | The same function-only scheduler prefers database-timed expired work, revalidates exact current authority, and advances at most three higher WorkerLease generations through durable-boundary recovery | Caller-selected retry routing/timing/generation, terminal-failure retry, dead-letter or requeue authority, delete execution, Runtime authority, or a new process/queue |
 | Private delivery ingress | [0045 — Redeem private delivery evidence at ingress](0045-redeem-private-delivery-evidence-at-ingress.md) | One digest-only service/request/asker/audience/epoch-bound DeliveryEvidenceRef constructs private TrustedDeliveryContext inside the current UserActor transaction before content work | Raw trusted delivery facts on the wire, bearer persistence, application-role minting/table reads, alternate Runtime paths, or claiming later M2 carriers |
 | Exact Package egress | [0046 — Bind egress to one exact Package hop](0046-bind-egress-to-one-exact-package-hop.md) | One digest-only grant binds one exact audience-bound Package to one model or channel preflight hop and redeems atomically | Treating Package construction as disclosure authority, arbitrary content at egress, cross-hop reuse, or bypassing final policy |
 | Public OpenAPI v0 | [0047 — Freeze OpenAPI v0 through one Runtime path](0047-freeze-openapi-v0-through-one-runtime-path.md) | One public `/v0/resolve` schema and a hidden provisional v1 bridge share the same sealed Runtime; Package release lineage is read-only from the Learning-published active manifest | Two authorization compositions, caller-authored release facts, Runtime publication/fallback, or in-place mutation of historical snapshots |

engine/persistence/__init__.py

@@ -81,6 +81,7 @@
 )
 from engine.persistence.worker_jobs import (
     DEFAULT_WORKER_LEASE_TTL_SECONDS,
+    FILE_DISPATCH_MAX_LEASE_GENERATION,
     MAX_WORKER_LEASE_TTL_SECONDS,
     FileDispatchLease,
     FileDispatchNoWork,
@@ -162,6 +163,7 @@
     "WorkerNoOpCompletion",
     "WorkerNoOpOutcome",
     "DEFAULT_WORKER_LEASE_TTL_SECONDS",
+    "FILE_DISPATCH_MAX_LEASE_GENERATION",
     "MAX_WORKER_LEASE_TTL_SECONDS",
     "assert_runtime_role",
     "assert_action_role",

engine/persistence/schema_security_manifest.yaml

@@ -1,5 +1,5 @@
 {
-  "manifestVersion": "31.0.0",
+  "manifestVersion": "32.0.0",
   "controlOperations": [
     {
       "name": "register_file_source",
@@ -314,15 +314,24 @@
       "serverCapabilityInput": "all-or-nothing configured logical File root assertion applied only after globally oldest eligible selection and again during final post-lock revalidation; an omitted root yields no-work and cannot redirect selection",
       "databaseOwnedTime": "full-precision clock_timestamp authority check refreshed after every progress-lock wait; protocol lease timestamps and immediate worker verification use the PostgreSQL clock normalized to whole seconds",
       "fixedTtlSeconds": 300,
-      "selection": "oldest current v3/v4 page-scheduled upsert ordered by database acceptance time, checkpoint sequence, page ordinal, change ordinal, Organization, Source, and job id; the claim holds the per-Source progress advisory lock, refreshes its statement snapshot, revalidates latest accepted scan epoch, and uses FOR UPDATE SKIP LOCKED authority-row fencing before leasing",
+      "automaticReclaimPolicy": {
+        "maximumReclaims": 3,
+        "maximumLeaseGeneration": 4,
+        "baseBackoffSeconds": 30,
+        "backoff": "30 * 2^(current lease generation - 1), measured after database-issued lease expiry",
+        "eligibleStates": ["leased", "running", "prepared", "ready"],
+        "exhaustedBehavior": "content-free no-work; later dead-letter/operator owner remains NOT_ACTIVE"
+      },
+      "selection": "eligible expired recovery attempts are selected before first attempts; recovery orders by database-owned retry eligibility then original acceptance lineage and exact tenant/job tie-breakers, while first attempts retain the Issue #91 global order; both paths hold the per-Source progress advisory lock, refresh the statement snapshot, revalidate latest accepted scan epoch, and use FOR UPDATE SKIP LOCKED authority-row fencing before leasing",
       "currentAuthority": [
         "active exact SourceVersion",
         "latest accepted scan epoch",
         "active exact Membership/version",
         "enabled exact File import ServicePrincipal"
       ],
       "atomicWrites": [
-        "file_import_job"
+        "file_import_job",
+        "file_import_job_event"
       ]
     },
     {
@@ -5840,6 +5849,7 @@
           "context_worker_activate_recoverable_file_publication",
           "context_worker_record_file_import_interruption"
           ,"context_scheduler_claim_file_import"
+          ,"context_scheduler_claim_first_file_import"
         ],
         "definerRoles": [
           "context_engine_worker_lease_definer",
@@ -5884,7 +5894,7 @@
         ],
         "context_engine_file_dispatch_definer": [
           "SELECT",
-          "UPDATE state, signing_key_version, lease_nonce_digest, lease_issued_at, lease_expires_at, lease_generation, dispatch_claimed"
+          "UPDATE state, signing_key_version, lease_nonce_digest, lease_issued_at, lease_expires_at, lease_generation, dispatch_claimed, recovery_from_state, lease_redeemed_at"
         ]
       },
       "partitions": [],
@@ -5905,6 +5915,8 @@
         "PG-FILE-CHANGE-SUPERSESSION-083"
         ,"PG-FILE-DISPATCH-091"
         ,"PG-FILE-DISPATCH-CONCURRENCY-091"
+        ,"PG-FILE-RECLAIM-093"
+        ,"PG-FILE-RECLAIM-CONCURRENCY-093"
       ]
     },
     {
@@ -6171,6 +6183,22 @@
               "context_engine_worker_lease_definer"
             ],
             "withCheck": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid"
+          },
+          {
+            "name": "file_import_job_event_file_dispatch_definer_select",
+            "command": "SELECT",
+            "roles": [
+              "context_engine_file_dispatch_definer"
+            ],
+            "using": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid"
+          },
+          {
+            "name": "file_import_job_event_file_dispatch_definer_insert",
+            "command": "INSERT",
+            "roles": [
+              "context_engine_file_dispatch_definer"
+            ],
+            "withCheck": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid"
           }
         ]
       },
@@ -6191,8 +6219,12 @@
           "context_worker_index_file_publication",
           "context_worker_activate_recoverable_file_publication",
           "context_worker_record_file_import_interruption"
+          ,"context_scheduler_claim_file_import"
+        ],
+        "definerRoles": [
+          "context_engine_worker_lease_definer",
+          "context_engine_file_dispatch_definer"
         ],
-        "definerRole": "context_engine_worker_lease_definer",
         "directTableMutationAllowed": false
       },
       "permittedOperations": {
@@ -6210,6 +6242,10 @@
         "context_engine_worker_lease_definer": [
           "SELECT",
           "INSERT"
+        ],
+        "context_engine_file_dispatch_definer": [
+          "SELECT",
+          "INSERT"
         ]
       },
       "retention": {
@@ -6228,7 +6264,9 @@
         "DB-001",
         "DB-004",
         "DB-008",
-        "PG-FILE-RECOVERY-027"
+        "PG-FILE-RECOVERY-027",
+        "PG-FILE-RECLAIM-093",
+        "PG-FILE-RECLAIM-CONCURRENCY-093"
       ]
     },
     {

engine/persistence/worker_jobs.py

@@ -39,6 +39,7 @@
 
 DEFAULT_WORKER_LEASE_TTL_SECONDS: Final = 300
 MAX_WORKER_LEASE_TTL_SECONDS: Final = 3600
+FILE_DISPATCH_MAX_LEASE_GENERATION: Final = 4
 
 
 def _utc_now() -> datetime:
@@ -165,14 +166,14 @@ class WorkerLeaseAuthorityUnavailable(RuntimeError):
 
 @dataclass(frozen=True, slots=True)
 class FileDispatchNoWork:
-    """Closed, content-free result when no first-attempt File job is eligible."""
+    """Closed, content-free result when no automatic File work is eligible."""
 
     status: Literal["no_work"] = field(default="no_work", init=False)
 
 
 @dataclass(frozen=True, slots=True, repr=False)
 class FileDispatchLease:
-    """Scheduler-minted exact first-attempt lease and internal routing facts."""
+    """Scheduler-minted exact bounded-attempt lease and internal routing facts."""
 
     token: WorkerLeaseToken = field(repr=False)
     organization_id: UUID = field(repr=False)
@@ -191,8 +192,15 @@ def __post_init__(self) -> None:
         if type(self.source_ref) is not SourceRef:
             raise TypeError("File dispatch source must be SourceRef")
         _require_uuid("service_principal_id", self.service_principal_id)
-        if self.lease_generation != 1:
-            raise ValueError("first-attempt File dispatch requires generation one")
+        if (
+            type(self.lease_generation) is not int
+            or not 1
+            <= self.lease_generation
+            <= FILE_DISPATCH_MAX_LEASE_GENERATION
+        ):
+            raise ValueError(
+                "File dispatch lease exceeds the automatic generation budget"
+            )
         issued_at = _require_utc("issued_at", self.issued_at)
         expires_at = _require_utc("expires_at", self.expires_at)
         if expires_at <= issued_at:
@@ -219,7 +227,7 @@ def __repr__(self) -> str:
 
 
 class PostgreSQLFileDispatchAuthority:
-    """Claim the oldest eligible scheduled File import without tenant input."""
+    """Claim the oldest eligible first or bounded recovery File attempt."""
 
     __slots__ = ("_codec", "_configured_root_refs", "_scheduler_engine")
 
@@ -245,7 +253,7 @@ def __init__(
         self._configured_root_refs = configured_root_refs
 
     def claim(self) -> FileDispatchClaim:
-        """Atomically fence one first attempt, then mint its exact existing token."""
+        """Atomically fence one bounded attempt, then mint its exact token."""
 
         nonce = generate_worker_lease_nonce()
         try:

eval/catalogs/m0-security-evidence.yaml

@@ -540,6 +540,21 @@
       "layer": "runtime",
       "selector": "tests/integration/test_file_dispatch.py::test_independent_worker_process_dispatches_and_publishes_one_job"
     },
+    {
+      "id": "PG-FILE-RECLAIM-093",
+      "layer": "postgres",
+      "selector": "tests/integration/test_file_dispatch.py::test_scheduler_reclaims_one_expired_dispatch_after_database_backoff"
+    },
+    {
+      "id": "PG-FILE-RECLAIM-CONCURRENCY-093",
+      "layer": "postgres",
+      "selector": "tests/integration/test_file_dispatch.py::test_concurrent_schedulers_reclaim_one_expired_generation_once"
+    },
+    {
+      "id": "PROC-FILE-RECLAIM-093",
+      "layer": "runtime",
+      "selector": "tests/integration/test_file_dispatch.py::test_scheduler_recovers_interrupted_publication_and_stales_old_lease"
+    },
     {
       "id": "HTTP-FILE-MIXED-UPSERT-NO-DELETE-089",
       "layer": "runtime",
@@ -686,12 +701,15 @@
           "PG-FILE-MIXED-UPSERT-SCHEDULE-089",
           "PG-FILE-MIXED-UPSERT-REPLAY-089",
           "PG-FILE-DISPATCH-091",
-          "PG-FILE-DISPATCH-CONCURRENCY-091"
+          "PG-FILE-DISPATCH-CONCURRENCY-091",
+          "PG-FILE-RECLAIM-093",
+          "PG-FILE-RECLAIM-CONCURRENCY-093"
         ],
         "runtime": [
           "RUNTIME-WORKER-LEASE-007",
           "HTTP-FILE-MIXED-UPSERT-NO-DELETE-089",
-          "PROC-FILE-DISPATCH-091"
+          "PROC-FILE-DISPATCH-091",
+          "PROC-FILE-RECLAIM-093"
         ]
       }
     },