From b117e35e4e26a22eb969bf61d192b338102770a1 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Thu, 8 Oct 2026 10:42:09 -0700 Subject: [PATCH 1/5] Share the capacity rejection check with background consumers --- src/firefly_weave/connectors/execution.py | 17 ++++------------- src/firefly_weave/definitions/models.py | 8 ++++++++ 2 files changed, 12 insertions(+), 13 deletions(-) diff --git a/src/firefly_weave/connectors/execution.py b/src/firefly_weave/connectors/execution.py index 4a134763..089402ed 100644 --- a/src/firefly_weave/connectors/execution.py +++ b/src/firefly_weave/connectors/execution.py @@ -37,6 +37,7 @@ TaskError, TaskLease, ) +from firefly_weave.definitions.models import capacity_rejected from firefly_weave.email.service import EmailService from firefly_weave.operations.execution import request_execution from firefly_weave.persistence.uow import UnitOfWork @@ -152,16 +153,6 @@ async def email_submit(payload: JsonObject) -> JsonObject: active = False -# Admission rejections: the operation ran nothing, so the identical call may be sent again. -CAPACITY_CODES = frozenset({"WV-OPERATION-CAPACITY", "WV-REQUEST-CAPACITY"}) - - -def _capacity_rejected(error: BaseException) -> bool: - from firefly_weave.definitions.models import CatalogError - - return isinstance(error, CatalogError) and error.status == 429 and error.code in CAPACITY_CODES - - class ServiceTransport: """The in-process worker transport, with the remote transport's replay policy (sdk/transport.py).""" @@ -182,7 +173,7 @@ async def claim(self, limit: int) -> list[TaskLease]: if error.status == 503 and error.code == "WV-COMPATIBILITY": return [] # A busy database turned the claim away before it ran: claim nothing and poll again. - if _capacity_rejected(error): + if capacity_rejected(error): return [] raise @@ -195,7 +186,7 @@ async def _replay(self, name: str, *args: Any, settlement: bool = False) -> Any: try: return await self.service.operation(name, self.scope, self.principal_id, *args) except Exception as error: - if not _capacity_rejected(error): + if not capacity_rejected(error): raise rejected = error attempts, seconds = (48, 10) if settlement else (3, 1) @@ -206,7 +197,7 @@ async def _replay(self, name: str, *args: Any, settlement: bool = False) -> Any: try: return await self.service.operation(name, self.scope, self.principal_id, *args) except Exception as error: - if not _capacity_rejected(error): + if not capacity_rejected(error): raise rejected = error except TimeoutError: diff --git a/src/firefly_weave/definitions/models.py b/src/firefly_weave/definitions/models.py index bfaaf767..78eedaa9 100644 --- a/src/firefly_weave/definitions/models.py +++ b/src/firefly_weave/definitions/models.py @@ -36,3 +36,11 @@ def __init__( self.result = result # Whole seconds a client should wait before sending a refused request again. self.retry_after = retry_after + + +# Admission rejections: the operation ran nothing, so the identical call may be sent again. +CAPACITY_CODES = frozenset({"WV-OPERATION-CAPACITY", "WV-REQUEST-CAPACITY"}) + + +def capacity_rejected(error: BaseException) -> bool: + return isinstance(error, CatalogError) and error.status == 429 and error.code in CAPACITY_CODES From 7c3454c34cb9c23566fd3d3ce4bbb1b260280ec5 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Thu, 8 Oct 2026 10:42:10 -0700 Subject: [PATCH 2/5] Leave an outbox attempt recoverable when its database work is refused for capacity --- docs/reference/integration-events.md | 6 + .../operations/event_delivery.py | 11 +- tests/integration/test_outbox.py | 65 +++++++ tests/unit/operations/test_outbox_capacity.py | 171 ++++++++++++++++++ 4 files changed, 249 insertions(+), 4 deletions(-) create mode 100644 tests/unit/operations/test_outbox_capacity.py diff --git a/docs/reference/integration-events.md b/docs/reference/integration-events.md index 373b9ba7..5b3719e1 100644 --- a/docs/reference/integration-events.md +++ b/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. diff --git a/src/firefly_weave/operations/event_delivery.py b/src/firefly_weave/operations/event_delivery.py index a614341c..89d59a71 100644 --- a/src/firefly_weave/operations/event_delivery.py +++ b/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: diff --git a/tests/integration/test_outbox.py b/tests/integration/test_outbox.py index c12cc848..e5817248 100644 --- a/tests/integration/test_outbox.py +++ b/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"} diff --git a/tests/unit/operations/test_outbox_capacity.py b/tests/unit/operations/test_outbox_capacity.py new file mode 100644 index 00000000..bc166a50 --- /dev/null +++ b/tests/unit/operations/test_outbox_capacity.py @@ -0,0 +1,171 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + +"""A capacity refusal during an outbox attempt leaves the fenced lease recoverable instead of settling it.""" + +from contextlib import asynccontextmanager +from types import SimpleNamespace +from uuid import uuid4 + +import pytest + +from firefly_weave.connectors.http import HttpPolicy +from firefly_weave.contracts.access import Scope +from firefly_weave.definitions.models import CatalogError +from firefly_weave.operations.event_delivery import OutboxDispatcher + + +def _refused(code: str = "WV-OPERATION-CAPACITY") -> CatalogError: + return CatalogError(429, code, "Operation capacity unavailable") + + +class _Harness: + """Real attempt; the store, authority check, secret resolution, receiver, and settlement are fakes.""" + + def __init__(self, monkeypatch, *, opens=(), settles=(), resolve_error=None): + self.scope = Scope(tenant_id=uuid4(), project_id=uuid4(), environment_id=uuid4()) + self.identifier, self.token = uuid4(), uuid4() + # Errors raised by the next mutation transactions at the outer UnitOfWork.open boundary. + self.opens = list(opens) + self.settles = list(settles) + self.settled: list[tuple[str, bool, bool]] = [] + self.sent: list[str] = [] + harness = self + + @asynccontextmanager + async def open(scope, *, mutation=True): + if mutation and harness.opens: + error = harness.opens.pop(0) + if error is not None: + raise error + yield SimpleNamespace(scope=scope) + + async def resolve(scope, binding_id, check): + if resolve_error is not None: + raise resolve_error + revision = SimpleNamespace( + config={"baseUrl": "https://receiver.example/"}, allowed_destinations=("receiver.example",) + ) + return revision, {"signing": SimpleNamespace(value="s" * 32, provider_version="one")} + + class Client: + async def request_bounded(self, method, url, **options): + harness.sent.append(url) + return SimpleNamespace(status_code=204) + + class Repository: + def __init__(self, tx): + pass + + async def rows(self, sql, **params): + return [{"payload": {"ok": True}}] + + async def execute(self, sql, **params): + return None + + monkeypatch.setattr("firefly_weave.operations.event_delivery.RuntimeRepository", Repository) + registry = SimpleNamespace(require_operational=lambda: None) + subscriptions = SimpleNamespace( + uow=SimpleNamespace(open=open), + bindings=SimpleNamespace(connections=SimpleNamespace(registry=registry), resolve=resolve), + ) + self.dispatcher = OutboxDispatcher(SimpleNamespace(), subscriptions, Client(), HttpPolicy()) + subscription = SimpleNamespace(binding_id=uuid4(), signing_slot="signing", path="events", statuses=(204,)) + + async def checked(tx, identifier, token): + return {"event_id": uuid4()}, subscription + + async def settle(scope, identifier, token, code, accepted, *, terminal=False): + if harness.settles: + raise harness.settles.pop(0) + harness.settled.append((code, accepted, terminal)) + return "incident" if terminal else ("delivered" if accepted else "retry") + + monkeypatch.setattr(self.dispatcher, "_checked", checked) + monkeypatch.setattr(self.dispatcher, "_settle", settle) + + async def attempt(self) -> str: + return await self.dispatcher._attempt(self.scope, self.identifier, self.token) + + +async def test_unrefused_attempt_delivers_and_settles_the_acknowledgment(monkeypatch): + harness = _Harness(monkeypatch) + + assert await harness.attempt() == "delivered" + + assert harness.sent == ["https://receiver.example/events"] + assert harness.settled == [("ACK", True, False)] + + +@pytest.mark.parametrize("code", ["WV-OPERATION-CAPACITY", "WV-REQUEST-CAPACITY"]) +async def test_refusal_before_send_leaves_the_attempt_recoverable(monkeypatch, code): + # The transaction that records the provider version before the request is turned away; + # the recheck that follows would be admitted. + harness = _Harness(monkeypatch, opens=[_refused(code)]) + + assert await harness.attempt() == "retry" + + assert harness.sent == [] + assert harness.settled == [] + + +async def test_refused_recheck_does_not_revoke_the_subscription(monkeypatch): + # Secret resolution fails for an unrelated reason; the recheck that would prove revocation is refused. + harness = _Harness(monkeypatch, resolve_error=RuntimeError("secret store unavailable"), opens=[_refused()]) + + assert await harness.attempt() == "retry" + + assert harness.sent == [] + assert harness.settled == [] + + +async def test_refused_settlement_after_send_records_no_failure(monkeypatch): + harness = _Harness(monkeypatch, settles=[_refused()]) + + assert await harness.attempt() == "retry" + + # The receiver saw the request; lease expiry records ACK_UNKNOWN and redelivers the same event ID. + assert harness.sent == ["https://receiver.example/events"] + assert harness.settled == [] + + +async def test_other_failures_before_send_still_settle_delivery_failed(monkeypatch): + harness = _Harness(monkeypatch, resolve_error=RuntimeError("secret store unavailable")) + + assert await harness.attempt() == "retry" + + assert harness.sent == [] + assert harness.settled == [("DELIVERY_FAILED", False, False)] + + +async def test_other_catalog_errors_on_recheck_still_revoke(monkeypatch): + missing = CatalogError(404, "WV-NOT-FOUND", "Subscription not found") + # The first check passes; the recheck after the failed resolution finds the subscription gone. + harness = _Harness(monkeypatch, resolve_error=RuntimeError("secret store unavailable")) + passed = harness.dispatcher._checked + calls = [] + + async def checked(tx, identifier, token): + calls.append(identifier) + if len(calls) > 1: + raise missing + return await passed(tx, identifier, token) + + monkeypatch.setattr(harness.dispatcher, "_checked", checked) + + assert await harness.attempt() == "incident" + + assert harness.settled == [("AUTHORITY_REVOKED", False, True)] From 70d889e10891303be56213fba1bd65bbe9bae1aa Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Thu, 8 Oct 2026 10:42:10 -0700 Subject: [PATCH 3/5] Retry refused platform calls of a native handler instead of failing its task --- docs/reference/http-and-webhooks.md | 10 + src/firefly_weave/connectors/execution.py | 92 +++++- .../test_native_handler_capacity.py | 262 ++++++++++++++++++ 3 files changed, 350 insertions(+), 14 deletions(-) create mode 100644 tests/unit/connectors/test_native_handler_capacity.py diff --git a/docs/reference/http-and-webhooks.md b/docs/reference/http-and-webhooks.md index 5260e003..2b65aa50 100644 --- a/docs/reference/http-and-webhooks.md +++ b/docs/reference/http-and-webhooks.md @@ -356,6 +356,16 @@ 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.** A capacity rejection (`WV-OPERATION-CAPACITY` or +`WV-REQUEST-CAPACITY`) no longer fails a claimed task at once. If one turns away +the execution slot or 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 diff --git a/src/firefly_weave/connectors/execution.py b/src/firefly_weave/connectors/execution.py index 089402ed..4de1689b 100644 --- a/src/firefly_weave/connectors/execution.py +++ b/src/firefly_weave/connectors/execution.py @@ -18,6 +18,10 @@ import asyncio import json +import math +from collections.abc import Awaitable, Callable +from contextlib import AsyncExitStack +from functools import partial from typing import TYPE_CHECKING, Any, cast from uuid import NAMESPACE_URL, UUID, uuid5 @@ -42,12 +46,51 @@ from firefly_weave.operations.execution import 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__( @@ -70,38 +113,59 @@ async def operation(self, name: str, scope: Scope, principal_id: UUID, *args: An async def execute(self, scope: Scope, principal_id: UUID, lease: TaskLease) -> JsonValue: self.registry.require_operational() - async with request_execution(): + async with AsyncExitStack() as stack: + await _admitted(lease.proof, lambda: stack.enter_async_context(request_execution()), waits=True) 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: diff --git a/tests/unit/connectors/test_native_handler_capacity.py b/tests/unit/connectors/test_native_handler_capacity.py new file mode 100644 index 00000000..33586c1c --- /dev/null +++ b/tests/unit/connectors/test_native_handler_capacity.py @@ -0,0 +1,262 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + +"""A native handler waits through capacity refusals of calls that ran nothing instead of failing its task.""" + +import asyncio +from contextlib import AsyncExitStack, asynccontextmanager +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from uuid import uuid4 + +import pytest + +from firefly_weave.access.authorization import AccessDenied +from firefly_weave.connectors.execution import ConnectorExecutionService, ServiceTransport +from firefly_weave.contracts.access import Scope +from firefly_weave.contracts.workers import LeaseProof, TaskLease +from firefly_weave.definitions.models import CatalogError +from firefly_weave.operations.execution import WORK_SLOTS, request_execution +from firefly_weave.sdk.worker import Worker + + +def busy(code: str = "WV-OPERATION-CAPACITY") -> CatalogError: + return CatalogError(429, code, "Operation capacity unavailable") + + +def lease() -> TaskLease: + now = datetime.now(UTC) + return TaskLease( + proof=LeaseProof(task_id=uuid4(), generation=1, owner=uuid4(), token="opaque"), + input={"value": 1}, + operation_key="stable", + deadline=now + timedelta(seconds=10), + expires_at=now + timedelta(seconds=3), + capability="call@1.0.0", + worker_release_id=uuid4(), + ) + + +class Tasks: + """The task service seen by native execution; queued errors refuse the next calls (None admits one).""" + + def __init__(self, errors: dict[str, list[Exception | None]]) -> None: + self.errors = {name: list(queue) for name, queue in errors.items()} + self.calls: list[str] = [] + self.settled: list[tuple[str, object]] = [] + self.invocation_value = SimpleNamespace(connection=SimpleNamespace(id=uuid4(), secret_refs={"token": "h"})) + + def admit(self, name: str) -> None: + self.calls.append(name) + queue = self.errors.get(name) + if queue: + error = queue.pop(0) + if error is not None: + raise error + + async def invocation(self, tx, proof, *, actor, scope, context): + self.admit("invocation") + return "adapter", self.invocation_value + + async def heartbeat(self, tx, proof, *, actor, scope, context): + now = datetime.now(UTC) + return TaskLease( + proof=proof, + input=None, + operation_key="stable", + deadline=now + timedelta(seconds=10), + expires_at=now + timedelta(seconds=3), + capability="call@1.0.0", + worker_release_id=uuid4(), + ) + + async def complete(self, tx, proof, completion_id, output, *, actor, scope, context): + self.settled.append(("complete", output)) + + async def fail(self, tx, proof, error, *, actor, scope, context): + self.settled.append(("fail", error.code)) + + async def credentials(self, request, *, actor, scope, context): + self.admit("credentials") + return SimpleNamespace(slot=request.slot) + + async def credential_authority(self, request, *, actor, scope, context): + self.admit("credential_authority") + + +class Harness: + """Real native execution, in-process transport, and worker; storage and the connector are fakes.""" + + def __init__(self, errors=None, body=None) -> None: + self.tasks = Tasks(errors or {}) + self.ran: list[object] = [] + harness = self + + @asynccontextmanager + async def open(scope, *, mutation=True): + yield SimpleNamespace(scope=scope) + + async def load_principal(principal_id, tx=None): + return SimpleNamespace(id=principal_id) + + class Adapter: + async def execute(self, values, context): + harness.ran.append(values) + if body is not None: + return await body(context) + return {"ok": True} + + registry = SimpleNamespace(require_operational=lambda: None, get=lambda adapter: Adapter()) + self.service = ConnectorExecutionService( + self.tasks, # type: ignore[arg-type] + SimpleNamespace(load_principal=load_principal), # type: ignore[arg-type] + SimpleNamespace(open=open), # type: ignore[arg-type] + registry, # type: ignore[arg-type] + SimpleNamespace(), # type: ignore[arg-type] + ) + scope, principal = Scope(tenant_id=uuid4(), project_id=uuid4(), environment_id=uuid4()), uuid4() + + async def handler(task: TaskLease): + return await self.service.execute(scope, principal, task) + + self.worker = Worker(ServiceTransport(self.service, scope, principal, uuid4()), {"call@1.0.0": handler}, 1) + + async def run(self) -> None: + await self.worker._execute(lease()) + + +async def test_unrefused_task_completes(): + harness = Harness() + + await harness.run() + + assert harness.tasks.settled == [("complete", {"ok": True})] + assert harness.ran == [{"value": 1}] + + +async def test_task_waits_for_a_work_slot_instead_of_failing(): + harness = Harness() + held, release = asyncio.Event(), asyncio.Event() + + async def occupy(): + # Other requests own every pure-execution work slot when the task arrives. + async with AsyncExitStack() as stack: + for _ in range(WORK_SLOTS): + await stack.enter_async_context(request_execution()) + held.set() + await release.wait() + + holder = asyncio.create_task(occupy()) + try: + await held.wait() + execution = asyncio.create_task(harness.run()) + await asyncio.sleep(0.2) + assert not execution.done() and harness.ran == [] + finally: + release.set() + await holder + await asyncio.wait_for(execution, 5) + + assert harness.tasks.settled == [("complete", {"ok": True})] + assert harness.ran == [{"value": 1}] + + +@pytest.mark.parametrize("code", ["WV-OPERATION-CAPACITY", "WV-REQUEST-CAPACITY"]) +async def test_refused_invocation_check_waits_past_the_direct_window(code): + # Six refusals outlast the one-second window of a direct call; the owning handler keeps its lease window. + harness = Harness({"invocation": [busy(code)] * 6}) + + await harness.run() + + assert harness.tasks.settled == [("complete", {"ok": True})] + assert harness.tasks.calls.count("invocation") == 7 + assert harness.ran == [{"value": 1}] + + +async def test_refused_authorization_inside_the_handler_is_sent_again(): + async def body(context): + await context.authorize() + return {"authorized": True} + + harness = Harness({"invocation": [None, busy()], "credential_authority": [busy()]}, body) + + await harness.run() + + assert harness.tasks.settled == [("complete", {"authorized": True})] + assert harness.tasks.calls == ["invocation"] * 3 + ["credential_authority"] * 2 + + +async def test_refused_credentials_are_sent_again(): + async def body(context): + return {"slot": (await context.credentials("token")).slot} + + harness = Harness({"credentials": [busy(), busy()]}, body) + + await harness.run() + + assert harness.tasks.settled == [("complete", {"slot": "token"})] + assert harness.tasks.calls.count("credentials") == 3 + + +@pytest.mark.parametrize( + "error", [AccessDenied(), CatalogError(409, "WV-OPERATION-UNAVAILABLE", "x"), CatalogError(429, "WV-PAGE", "x")] +) +async def test_other_refusals_still_fail_the_task(error): + harness = Harness({"invocation": [error]}) + + await harness.run() + + assert harness.tasks.settled == [("fail", "HANDLER_FAILED")] + assert harness.tasks.calls == ["invocation"] + assert harness.ran == [] + + +async def test_checks_while_the_connector_runs_keep_the_direct_window(): + outcome: list[str] = [] + + async def body(context): + # The connector may hold its own resources here, so a refusal is not waited out. + try: + await context.authorize() + except CatalogError as error: + outcome.append(error.code) + return {"checked": True} + + harness = Harness({"credential_authority": [busy()] * 10}, body) + + await harness.run() + + assert outcome == ["WV-OPERATION-CAPACITY"] + assert harness.tasks.calls.count("credential_authority") == 3 + assert harness.tasks.settled == [("complete", {"checked": True})] + + +async def test_a_retry_cannot_outlive_the_handler(): + leftovers: list[asyncio.Task] = [] + + async def body(context): + # A connector's background task asks for a credential and is refused; the handler returns. + leftovers.append(asyncio.create_task(context.credentials("token"))) + await asyncio.sleep(0) + return {"done": True} + + harness = Harness({"credentials": [busy()]}, body) + + await harness.run() + + with pytest.raises(ValueError, match="Credential unavailable"): + await leftovers[0] + assert harness.tasks.calls.count("credentials") == 1 From 08bce190f6d9ea0c7ac19749e7c30ad6c5bab456 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Thu, 8 Oct 2026 12:03:48 -0700 Subject: [PATCH 4/5] Describe native executor capacity without the request execution slot --- docs/reference/http-and-webhooks.md | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/docs/reference/http-and-webhooks.md b/docs/reference/http-and-webhooks.md index 2b65aa50..cc9bf958 100644 --- a/docs/reference/http-and-webhooks.md +++ b/docs/reference/http-and-webhooks.md @@ -356,11 +356,13 @@ 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.** A capacity rejection (`WV-OPERATION-CAPACITY` or -`WV-REQUEST-CAPACITY`) no longer fails a claimed task at once. If one turns away -the execution slot or 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 +**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. From ff0569c029027cba0266fce072aaa58a7878823e Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Thu, 8 Oct 2026 12:03:48 -0700 Subject: [PATCH 5/5] List capacity refusal and native executor capacity changes in the changelog --- CHANGELOG.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 526781fb..623a156a 100644 --- a/CHANGELOG.md +++ b/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. ## 0.1.0a14