Keep outbox deliveries and native tasks retryable after a capacity refusal

CHANGELOG.md

@@ -38,6 +38,25 @@ SPDX-License-Identifier: Apache-2.0
   than this release cannot read `weave connections test` answers for HTTP
   connections from an upgraded server; upgrade the CLI and SDK together with the
   server.
+- Keep an integration event delivery recoverable when Weave refuses one of its
+  database transactions for capacity (`WV-OPERATION-CAPACITY`). The delivery
+  stays `leased` instead of moving to `retry` with `DELIVERY_FAILED` or to an
+  `AUTHORITY_REVOKED` incident. When the lease expires, the attempt is recorded
+  as `ACK_UNKNOWN` and the same event ID is delivered again. The attempt still
+  counts, so a delivery refused on every attempt ends in a `DELIVERY_EXHAUSTED`
+  incident.
+- Native connector tasks no longer fail with `HANDLER_FAILED` when Weave refuses
+  one of their platform calls for capacity (`WV-OPERATION-CAPACITY` or
+  `WV-REQUEST-CAPACITY`). The invocation check before the connector starts is
+  sent again while the task's lease is valid; authority and credential checks
+  made while the connector runs get up to three attempts within one second.
+- Native connector calls no longer share the two execution work slots of API
+  requests, so a burst of requests no longer refuses them. They run up to the
+  `capacity` configured for each `WEAVE_NATIVE_EXECUTORS` entry.
+- `email_receipts.dispatch` answers HTTP 429 (`WV-OPERATION-CAPACITY` or
+  `WV-REQUEST-CAPACITY`) when Weave refuses the dispatch for capacity, where it
+  used to answer HTTP 200 with state `blocked`. The receipt keeps its state, and
+  the next pending scan dispatches it.
 - Draw "weave" in the Firefly Weave logo as a wordmark whose w is woven from two
   strands, one amber, instead of typed text, in Studio, on the desktop launch
   page and installer, and in the README banner and social preview. NOTICE names

docs/reference/http-and-webhooks.md

@@ -356,6 +356,18 @@ startup compatibility checks still report the runtime as restricted, an
 in-process executor claims nothing and keeps polling; tasks queued in that window
 are claimed once the API becomes operational.
 
+**Busy platform.** In-process executors run connector calls up to the
+`capacity` configured for each entry, apart from the execution capacity that API
+requests share, so a burst of requests does not refuse them. A capacity rejection
+(`WV-OPERATION-CAPACITY` or `WV-REQUEST-CAPACITY`) of a platform call no longer
+fails a claimed task at once. If one turns away the invocation check before the
+connector starts, the executor tries again while the task's lease stays valid,
+as the SDK does for context and credentials (see
+[the worker protocol](worker-protocol.md#retries-and-lost-responses)). Authority
+and credential checks made while the connector runs make at most three attempts
+within one second, so a refusal cannot hold the connector's own resources open.
+If the lease or deadline runs out first, the task fails as before.
+
 **Replicas.** API-only replicas may disable dispatch while a compatible peer runs
 connector actions. Lease recovery is a separate scheduler with limited database
 privileges, so keep one scheduler-enabled replica for recovery, waits, timers, and

docs/reference/integration-events.md

@@ -226,6 +226,12 @@ A delivery's status is `pending`, `leased`, `retry`, `delivered`, or `incident`.
   5 seconds; after later ones, 30 seconds. An expired lease is recorded as
   `ACK_UNKNOWN` before the next attempt. Attempts accumulate and are never
   erased.
+- **Busy platform.** If Weave turns away a database transaction during an
+  attempt for capacity (`WV-OPERATION-CAPACITY`), the attempt is recorded
+  neither as failed nor as `AUTHORITY_REVOKED`. The lease stays fenced and
+  expires, and the next attempt records it as `ACK_UNKNOWN`. The attempt still
+  counts toward the three, so sustained refusals end in the usual exhaustion
+  incident.
 - **Incidents.** When attempts run out or authority is revoked, the delivery
   becomes an operations incident with a controlled code and an audit record. It
   does not create or suspend a run.

src/firefly_weave/connectors/execution.py

@@ -18,6 +18,9 @@
 
 import asyncio
 import json
+import math
+from collections.abc import Awaitable, Callable
+from functools import partial
 from typing import TYPE_CHECKING, Any, cast
 from uuid import NAMESPACE_URL, UUID, uuid5
 
@@ -42,12 +45,51 @@
 from firefly_weave.operations.execution import ReservedSlots, request_execution
 from firefly_weave.persistence.uow import UnitOfWork
 from firefly_weave.runtime.repository import RuntimeRepository
+from firefly_weave.sdk._settlement import check_settlement, settlement_deadline
 from firefly_weave.workers.leases import TaskService
 
 if TYPE_CHECKING:
     from firefly_weave.providers.teams.references import TeamsReferences
 
 
+async def _admitted[Result](proof: LeaseProof, call: Callable[[], Awaitable[Result]], *, waits: bool = False) -> Result:
+    """Run a platform call of a native handler, again while the platform rejects it for capacity.
+
+    A rejection committed nothing, so the identical call may be sent again. Before the connector
+    starts (`waits`), the owning handler holds nothing else and retries while its lease watchdog
+    remains authoritative, like a remote worker's context and credentials (sdk/transport.py).
+    Checks made while the connector runs, which may hold its own resources, and calls from any
+    other task make at most three attempts within one second.
+    """
+    try:
+        return await call()
+    except Exception as error:
+        if not capacity_rejected(error):
+            raise
+        rejected = error
+    deadline = settlement_deadline(proof) if waits else None
+    if deadline is None:
+        attempts, window = 3, asyncio.timeout(1)
+    else:
+        attempts = math.ceil(max(0, deadline - asyncio.get_running_loop().time()) / 0.05) + 1
+        window = asyncio.timeout_at(deadline)
+    try:
+        async with window:
+            for retry in range(attempts - 1):
+                await asyncio.sleep(min(0.05 * 2 ** min(retry, 3), 0.25))
+                check_settlement(proof)
+                try:
+                    return await call()
+                except Exception as error:
+                    if not capacity_rejected(error):
+                        raise
+                    rejected = error
+    except TimeoutError:
+        if not window.expired():
+            raise
+    raise rejected
+
+
 @service
 class ConnectorExecutionService:
     def __init__(
@@ -78,34 +120,54 @@ async def execute(
             return await self._execute(scope, principal_id, lease)
 
     async def _execute(self, scope: Scope, principal_id: UUID, lease: TaskLease) -> JsonValue:
-        adapter, invocation = await self.operation("invocation", scope, principal_id, lease.proof)
+        invocation_check = partial(self.operation, "invocation", scope, principal_id, lease.proof)
+        adapter, invocation = await _admitted(lease.proof, invocation_check, waits=True)
         active = True
 
+        async def current[Result](call: Callable[[], Awaitable[Result]], refusal: str) -> Result:
+            async def attempt() -> Result:
+                # Every attempt rechecks, so a retry cannot outlive the handler.
+                if not active:
+                    raise ValueError(refusal)
+                return await call()
+
+            return await _admitted(lease.proof, attempt)
+
         async def credentials(slot: str) -> ResolvedSecret:
             if not active or slot not in invocation.connection.secret_refs:
                 raise ValueError("Credential unavailable")
             actor = await self.access.load_principal(principal_id)
-            return await self.tasks.credentials(
-                CredentialRequest(lease=lease.proof, connection_revision_id=invocation.connection.id, slot=slot),
-                actor=actor,
-                scope=scope,
-                context=AuditContext(),
+            return await current(
+                partial(
+                    self.tasks.credentials,
+                    CredentialRequest(lease=lease.proof, connection_revision_id=invocation.connection.id, slot=slot),
+                    actor=actor,
+                    scope=scope,
+                    context=AuditContext(),
+                ),
+                "Credential unavailable",
             )
 
         async def authorize() -> None:
             self.registry.require_operational()
             if not active:
                 raise ValueError("Connector authority unavailable")
-            current_adapter, current = await self.operation("invocation", scope, principal_id, lease.proof)
-            if current_adapter != adapter or current != invocation:
+            current_adapter, checked = await current(invocation_check, "Connector authority unavailable")
+            if current_adapter != adapter or checked != invocation:
                 raise ValueError("Connector authority changed")
             actor = await self.access.load_principal(principal_id)
             for slot in invocation.connection.secret_refs:
-                await self.tasks.credential_authority(
-                    CredentialRequest(lease=lease.proof, connection_revision_id=invocation.connection.id, slot=slot),
-                    actor=actor,
-                    scope=scope,
-                    context=AuditContext(),
+                await current(
+                    partial(
+                        self.tasks.credential_authority,
+                        CredentialRequest(
+                            lease=lease.proof, connection_revision_id=invocation.connection.id, slot=slot
+                        ),
+                        actor=actor,
+                        scope=scope,
+                        context=AuditContext(),
+                    ),
+                    "Connector authority unavailable",
                 )
 
         async def reference(identifier: UUID, generation: int, activity_id: str | None) -> JsonObject:

src/firefly_weave/operations/event_delivery.py

@@ -50,7 +50,7 @@
     Subscription,
     SubscriptionRequest,
 )
-from firefly_weave.definitions.models import CatalogError
+from firefly_weave.definitions.models import CatalogError, capacity_rejected
 from firefly_weave.operations.outbox import OutboxService
 from firefly_weave.operations.subscriptions import SubscriptionService
 from firefly_weave.persistence.paging import page_ids
@@ -331,17 +331,20 @@ async def check(tx: Transaction) -> UUID:
             return "retry"
         except asyncio.CancelledError:
             raise
-        except Exception:
+        except Exception as error:
             # An unavailable database cannot prove revocation. Leave the fenced
             # attempt recoverable if checking or settling cannot finish in budget.
-            if asyncio.get_running_loop().time() >= deadline:
+            # A capacity refusal is not a delivery failure: lease expiry recovers it.
+            if capacity_rejected(error) or asyncio.get_running_loop().time() >= deadline:
                 return "retry"
             try:
                 async with asyncio.timeout_at(deadline):
                     try:
                         async with self.uow.open(scope) as tx:
                             await self._checked(tx, identifier, token)
-                    except (AccessDenied, CatalogError, AuthenticationFailed):
+                    except (AccessDenied, CatalogError, AuthenticationFailed) as refused:
+                        if capacity_rejected(refused):
+                            return "retry"
                         return await self._settle(scope, identifier, token, "AUTHORITY_REVOKED", False, terminal=True)
                     return await self._settle(scope, identifier, token, "DELIVERY_FAILED", False)
             except Exception:

tests/integration/test_outbox.py

@@ -736,3 +736,68 @@ async def expired(*args, **kwargs):
     monkeypatch.setattr(s.subscriptions.bindings, "resolve", expired)
     assert (await s.dispatch()).retry == 1
     assert not s.remote.requests
+
+
+async def test_capacity_refusal_before_send_leaves_the_lease_for_recovery(setup, monkeypatch):
+    s = setup
+    await s.append()
+    original = s.subscriptions.bindings.resolve
+    held, released = asyncio.Event(), asyncio.Event()
+    tasks = []
+
+    async def hold():
+        # Another writer owns the project operations fence past the 250 ms admission timeout.
+        async with s.access_db[1].begin() as session:
+            await session.execute(
+                text(
+                    "SELECT pg_advisory_xact_lock(hashtextextended('weave.operations:' "
+                    "|| cast(:tenant AS text) || ':' || cast(:project AS text),0))"
+                ),
+                {"tenant": str(s.scope.tenant_id), "project": str(s.scope.project_id)},
+            )
+            held.set()
+            await released.wait()
+
+    async def watch():
+        # Release the fence once the refused writer gives up or another transaction queues behind it.
+        first = None
+        while not released.is_set():
+            async with s.access_db[1].begin() as session:
+                waiting = set(
+                    (
+                        await session.execute(
+                            text("SELECT virtualtransaction FROM pg_locks WHERE locktype='advisory' AND NOT granted")
+                        )
+                    ).scalars()
+                )
+            if first is None:
+                first = waiting or None
+            elif waiting != first:
+                released.set()
+                return
+            await asyncio.sleep(0.01)
+
+    async def contended(*args, **kwargs):
+        result = await original(*args, **kwargs)
+        tasks.append(asyncio.create_task(hold()))
+        async with asyncio.timeout(5):
+            await held.wait()
+        tasks.append(asyncio.create_task(watch()))
+        return result
+
+    monkeypatch.setattr(s.subscriptions.bindings, "resolve", contended)
+    report = await s.dispatch()
+    released.set()
+    async with asyncio.timeout(5):
+        await asyncio.gather(*tasks)
+    assert report.retry == 1 and report.incident == 0
+    assert not s.remote.requests
+    row = (await s.rows("event_deliveries"))[0]
+    # Nothing was settled: the fenced lease waits for expiry instead of recording a failed delivery.
+    assert row["status"] == "leased" and row["code"] is None and row["attempts"] == 1
+    assert [attempt["outcome"] for attempt in await s.rows("delivery_attempts")] == [None]
+    monkeypatch.setattr(s.subscriptions.bindings, "resolve", original)
+    await s.due()
+    assert (await s.dispatch()).delivered == 1
+    assert len(s.remote.requests) == 1
+    assert {attempt["outcome"] for attempt in await s.rows("delivery_attempts")} == {"ACK_UNKNOWN", "ACK"}