diff --git a/docs/operations/troubleshooting.md b/docs/operations/troubleshooting.md index 68efe76a..f7c19d3f 100644 --- a/docs/operations/troubleshooting.md +++ b/docs/operations/troubleshooting.md @@ -136,6 +136,7 @@ publish a full environment dump to explain either case. | Native OpenAPI or remote client import fails | The base compiler intentionally omits runtime dependencies | Install the `openapi` or `client` extra from the locked framework provenance | | The server rejects the schema | Startup never migrates or repairs an incompatible database | Install the packaged migrations and run them with explicit migration authority; see [upgrades](upgrades.md) | | Readiness stays restricted | Scheduler catalog credentials, the compatibility report, the policy fingerprint, or retained requirements block it; disabling the scheduler does not bypass the startup inventory | Follow [start and inspect compatibility](upgrades.md#start-and-inspect-compatibility) | +| Readiness drops for about a minute, then returns | A periodic compatibility rescan could not finish and withdrew readiness until the next rescan | Read the `Compatibility rescan withdrew readiness` warning for the finding codes and error class; see [start and inspect compatibility](upgrades.md#start-and-inspect-compatibility) | | Work admission reaches capacity | Admission limits are separate from physical disk space | Inspect the authorized usage and capabilities response, retained records, and configured limits; read [retention](retention.md) before maintenance | | The collector receives no telemetry | Export is disabled by default and needs an exact signal URL | Check `WEAVE_TELEMETRY`, the URL, collector TLS and authentication, and the network namespace; see [observability](observability.md) | | A remote command prints `WV-CLI-CONFIG` | No saved platform is selected and no `--base-url` was given | Run `weave auth setup SERVER`; see [configuration](configuration.md#client-environment-variables) | diff --git a/docs/operations/upgrades.md b/docs/operations/upgrades.md index d518abd8..a9962e2d 100644 --- a/docs/operations/upgrades.md +++ b/docs/operations/upgrades.md @@ -194,9 +194,19 @@ truncated or incomplete report is not evidence of full compatibility. **The server also rescans by itself.** Each process requests a compatibility rescan every 60 seconds. Automatic and explicit scans share one reservation, so -an automatic scan that would overlap is skipped instead of queued. A failed scan -leaves the process restricted until a later scan succeeds, and catalog cleanup -must succeed before readiness returns. +an automatic scan that would overlap is skipped instead of queued. A scan +classifies retained items in its own reserved execution slot, so a burst of API +requests cannot make it fail. A failed scan leaves the process restricted until +a later scan succeeds, and catalog cleanup must succeed before readiness +returns. A rescan that confirms a ready process keeps it ready while it releases +its catalog connection; if that cleanup fails, readiness is withdrawn. While the +process is restricted, refused changes answer HTTP 503 `WV-COMPATIBILITY` with +`Retry-After` set to the seconds until the next automatic rescan. + +When a rescan withdraws readiness, the server logs the warning `Compatibility +rescan withdrew readiness` with the finding kinds and codes and the class of +any error the scan caught, such as `CatalogError WV-OPERATION-CAPACITY`. The +warning never includes error messages or connection details. The capabilities response also reports the effective policy, fixed server ceilings, the supported worker convention, and a small readiness projection. An diff --git a/docs/reference/api.md b/docs/reference/api.md index b5d0b125..59c850a1 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -304,10 +304,15 @@ accepts it. The rules for using a key: different body fails with HTTP 409. - Clients never retry a change on their own after a timeout or another outcome they cannot know. If you retry, send the exact same body and key. The - one exception is a capacity rejection: HTTP 429 with `WV-OPERATION-CAPACITY` - or `WV-REQUEST-CAPACITY` means the platform did not admit the request, so - Studio sends a read, or a change that carries a key, again with the same key, - honoring `Retry-After`, for at most four attempts in all. + exceptions are rejections that mean the platform did not admit the request: + HTTP 429 with `WV-OPERATION-CAPACITY` or `WV-REQUEST-CAPACITY`, and HTTP 503 + with `WV-COMPATIBILITY`. For these, Studio sends a read, or a change that + carries a key, again with the same key, honoring `Retry-After`, for at most + four attempts in all. Studio sends a `WV-COMPATIBILITY` refusal again only + when its `Retry-After` is 5 seconds or less. +- A `WV-COMPATIBILITY` refusal carries `Retry-After` with the whole seconds + until the server's next automatic compatibility rescan. An explicit + compatibility check can lift a restricted verdict sooner. - Creating a connection revision accepts an optional key since 0.1.0a7: with one, a retry returns the revision already created; without one, every request creates a new revision. Debug commands diff --git a/src/firefly_weave/api/errors.py b/src/firefly_weave/api/errors.py index b2bec54e..d7493a70 100644 --- a/src/firefly_weave/api/errors.py +++ b/src/firefly_weave/api/errors.py @@ -34,9 +34,10 @@ async def access_denied(self, error: AccessDenied) -> JSONResponse: @exception_handler(CatalogError) async def catalog_error(self, error: CatalogError) -> JSONResponse: body = {"code": error.code, "message": error.message} + headers = None if error.retry_after is None else {"Retry-After": str(error.retry_after)} if error.result is not None: - return JSONResponse({**body, "result": error.result}, status_code=error.status) - return JSONResponse(body, status_code=error.status) + return JSONResponse({**body, "result": error.result}, status_code=error.status, headers=headers) + return JSONResponse(body, status_code=error.status, headers=headers) @exception_handler(ValidationError) async def invalid_contract(self, error: ValidationError) -> JSONResponse: diff --git a/src/firefly_weave/app.py b/src/firefly_weave/app.py index 3dcf01c1..6961e5d0 100644 --- a/src/firefly_weave/app.py +++ b/src/firefly_weave/app.py @@ -320,7 +320,7 @@ async def lifespan(app: Starlette) -> AsyncIterator[None]: compatibility = pyfly.context.get_bean(CompatibilityService) app.state.compatibility = compatibility app.state.telemetry_service = pyfly.context.get_bean(TelemetryService) - registry.set_operational_guard(lambda: compatibility.ready) + registry.set_operational_guard(lambda: compatibility.ready, retry_after=compatibility.retry_after_seconds) async def open_control_loop(*, required: bool = False) -> None: nonlocal recovery_loop diff --git a/src/firefly_weave/connections/registry.py b/src/firefly_weave/connections/registry.py index 33d8e8c7..fe9a5724 100644 --- a/src/firefly_weave/connections/registry.py +++ b/src/firefly_weave/connections/registry.py @@ -40,6 +40,7 @@ class ConnectorRegistry: def __init__(self, allowlist: tuple[str, ...] = ()) -> None: self._operational_guard: Callable[[], bool] | None = None + self._operational_retry: Callable[[], int] | None = None self._adapters: dict[str, ConnectorAdapter] = {} self._retired: set[str] = set() self._descriptors: dict[str, ConnectorDescriptor] = {} @@ -139,12 +140,18 @@ def register_trusted(self, reference: str, adapter: ConnectorAdapter) -> None: raise ValueError("Invalid connector adapter") self._adapters[reference] = adapter - def set_operational_guard(self, guard: Callable[[], bool]) -> None: + def set_operational_guard(self, guard: Callable[[], bool], *, retry_after: Callable[[], int] | None = None) -> None: self._operational_guard = guard + self._operational_retry = retry_after def require_operational(self) -> None: if self._operational_guard is not None and not self._operational_guard(): - raise CatalogError(503, "WV-COMPATIBILITY", "Execution compatibility unavailable") + raise CatalogError( + 503, + "WV-COMPATIBILITY", + "Execution compatibility unavailable", + retry_after=None if self._operational_retry is None else self._operational_retry(), + ) def retire(self, reference: str) -> None: self._retired.add(reference) diff --git a/src/firefly_weave/definitions/models.py b/src/firefly_weave/definitions/models.py index 57dcf8e0..bfaaf767 100644 --- a/src/firefly_weave/definitions/models.py +++ b/src/firefly_weave/definitions/models.py @@ -20,9 +20,19 @@ class CatalogError(Exception): - def __init__(self, status: int, code: str, message: str, *, result: dict[str, Any] | None = None) -> None: + def __init__( + self, + status: int, + code: str, + message: str, + *, + result: dict[str, Any] | None = None, + retry_after: int | None = None, + ) -> None: super().__init__(message) self.status = status self.code = code self.message = message self.result = result + # Whole seconds a client should wait before sending a refused request again. + self.retry_after = retry_after diff --git a/src/firefly_weave/operations/compatibility.py b/src/firefly_weave/operations/compatibility.py index 2b580f54..257ce135 100644 --- a/src/firefly_weave/operations/compatibility.py +++ b/src/firefly_weave/operations/compatibility.py @@ -19,6 +19,7 @@ import asyncio import json import logging +import math import time from collections.abc import AsyncIterator, Awaitable, Callable from contextlib import asynccontextmanager, suppress @@ -45,7 +46,7 @@ from firefly_weave.contracts.providers import ProviderSource, ProviderSourceRequest from firefly_weave.definitions.models import CatalogError from firefly_weave.operations.compatibility_catalog import classify_requirement -from firefly_weave.operations.execution import execute_pure +from firefly_weave.operations.execution import execute_pure, inventory_execution from firefly_weave.operations.telemetry import TelemetryService from firefly_weave.persistence.migrations import check_schema from firefly_weave.persistence.uow import UnitOfWork @@ -131,6 +132,15 @@ def classify_inventory_row( return classify_requirement(kind, payload, registry) +def log_withdrawal(findings: list[CompatibilityFinding], error: BaseException | None) -> None: + """Name why a rescan withdrew readiness without logging values or exception messages.""" + causes = ",".join(sorted({f"{finding.kind}:{finding.code}" for finding in findings})) or "none" + cause = "none" if error is None else type(error).__name__ + if isinstance(error, CatalogError): + cause = f"{cause} {error.code}" + logging.getLogger(__name__).warning("Compatibility rescan withdrew readiness: findings=%s error=%s", causes, cause) + + class FindingAccumulator: """Admission sees every finding even when presentation is truncated.""" @@ -175,6 +185,7 @@ def __init__( self._freshness_task: asyncio.Task[None] | None = None self._opened_at: float | None = None self._completed_at: float | None = None + self._next_scan = math.inf self._catalog_engine: AsyncEngine | None = None self.on_ready: Callable[[], Awaitable[None]] | None = None self.on_restricted: Callable[[], Awaitable[None]] | None = None @@ -194,18 +205,18 @@ def refresh_inventory_age(self) -> None: async def open(self) -> None: if self._freshness_task is None: self._opened_at = time.monotonic() + self._next_scan = self._opened_at + INVENTORY_RESCAN_SECONDS async def refresh() -> None: - next_scan = time.monotonic() + INVENTORY_RESCAN_SECONDS while True: await asyncio.sleep(INVENTORY_AGE_SECONDS) self.refresh_inventory_age() - if time.monotonic() >= next_scan and not self._scan_lock.locked(): + if time.monotonic() >= self._next_scan and not self._scan_lock.locked(): try: await self.scan() except Exception: logging.getLogger(__name__).error("Compatibility refresh failed") - next_scan = time.monotonic() + INVENTORY_RESCAN_SECONDS + self._next_scan = time.monotonic() + INVENTORY_RESCAN_SECONDS self._freshness_task = asyncio.create_task(refresh(), name="weave-inventory-freshness") @@ -265,6 +276,12 @@ def ready(self) -> bool: report = self.report return report.mode == "ready" and report.complete + def retry_after_seconds(self) -> int: + """Whole seconds until the next periodic rescan can change a restricted verdict.""" + # The refresh loop notices a due rescan on its next age tick. + wait = min(self._next_scan - time.monotonic(), INVENTORY_RESCAN_SECONDS) + INVENTORY_AGE_SECONDS + return max(1, math.ceil(wait)) + @asynccontextmanager async def _completed_scan(self) -> AsyncIterator[None]: try: @@ -280,9 +297,10 @@ async def scan(self) -> CompatibilityReport: async with self._scan_lock, self._completed_scan(): findings = FindingAccumulator() inspected = 0 - complete = False + complete = traversed = False engine = None failure: BaseException | None = None + swallowed: Exception | None = None inventory: dict[str, int] = {} record = findings.record @@ -309,6 +327,7 @@ async def scan(self) -> CompatibilityReport: ) self._catalog_engine = engine async with ( + inventory_execution(), asyncio.timeout(CATALOG_SCAN_SECONDS), engine.connect() as connection, connection.begin(), @@ -377,18 +396,26 @@ async def scan(self) -> CompatibilityReport: ) after_kind, after_id = rows[-1]["kind"], rows[-1]["id"] after_tenant = tenants[-1] + traversed = True except asyncio.CancelledError as error: failure = error record(CompatibilityFinding(kind="inventory", code="inventory_incomplete")) - except InventoryAuthorityError: + except InventoryAuthorityError as error: + swallowed = error record(CompatibilityFinding(kind="inventory", code="authority_missing")) - except Exception: + except Exception as error: + swallowed = error record(CompatibilityFinding(kind="inventory", code="inventory_incomplete")) finally: - # Readiness must be withdrawn before fallible or stalled cleanup. - self.report = CompatibilityReport( - checked_at=datetime.now(UTC), findings=findings.items, findings_truncated=findings.truncated - ) + # Readiness must be withdrawn before fallible or stalled cleanup, unless this + # rescan confirms the ready verdict in force: its effect owners are already + # initialized and cleanup is bounded, so only a failed cleanup withdraws it. + was_ready = self.ready + confirmed = was_ready and traversed and complete and not findings.incomplete and not findings.blocking + if not confirmed: + self.report = CompatibilityReport( + checked_at=datetime.now(UTC), findings=findings.items, findings_truncated=findings.truncated + ) try: await self._dispose_catalog() except BaseException as error: @@ -404,11 +431,20 @@ async def scan(self) -> CompatibilityReport: findings=findings.items, findings_truncated=findings.truncated, ) + confirmed = confirmed and candidate.mode == "ready" # A compatible inventory alone does not establish initialized effect owners. - self.report = candidate.model_copy(update={"mode": "restricted"}) + withdrawn = candidate.model_copy(update={"mode": "restricted"}) + if not confirmed: + self.report = withdrawn + if was_ready and candidate.mode == "restricted": + log_withdrawal(findings.items, swallowed or failure) callback = self.on_ready if candidate.mode == "ready" else self.on_restricted if callback is not None: - await callback() + try: + await callback() + except BaseException: + self.report = withdrawn + raise if failure is not None: raise failure self.report = candidate diff --git a/src/firefly_weave/operations/execution.py b/src/firefly_weave/operations/execution.py index e25bd56c..2cf8fe87 100644 --- a/src/firefly_weave/operations/execution.py +++ b/src/firefly_weave/operations/execution.py @@ -29,9 +29,11 @@ WORK_SLOTS = 2 CONTROL_SLOTS = 4 DEBUG_EXECUTION_SLOTS = 1 +INVENTORY_SLOTS = 1 _work_slots = threading.BoundedSemaphore(WORK_SLOTS) _control_slots = threading.BoundedSemaphore(CONTROL_SLOTS) _debug_slots = threading.BoundedSemaphore(DEBUG_EXECUTION_SLOTS) +_inventory_slots = threading.BoundedSemaphore(INVENTORY_SLOTS) class _Lease: @@ -104,6 +106,23 @@ async def request_execution(*, control: bool = False) -> AsyncIterator[None]: lease.close() +@asynccontextmanager +async def inventory_execution() -> AsyncIterator[None]: + """Own the compatibility inventory's reserved slot for one sequential scan. + + A request burst can hold every control slot; the inventory must not read that + refusal as an incompatible retained item. A classification still running from a + cancelled scan keeps the slot, so the next scan is refused instead of overlapping. + """ + lease = _Lease(True, slots=_inventory_slots) + token = _request_lease.set(lease) + try: + yield + finally: + _request_lease.reset(token) + lease.close() + + async def execute_pure[Result](call: Callable[[], Result], *, control: bool = False) -> Result: inherited = _request_lease.get() lease = inherited or _Lease(control) diff --git a/src/firefly_weave/operations/transport.py b/src/firefly_weave/operations/transport.py index 021b101c..db38b189 100644 --- a/src/firefly_weave/operations/transport.py +++ b/src/firefly_weave/operations/transport.py @@ -164,7 +164,9 @@ def __init__(self, app: ASGIApp, policy: TransportPolicy | None = None) -> None: for path in {operation.path, operation.canonical_path} ) - async def problem(self, send: Send, status: int, code: str, request_id: str) -> None: + async def problem( + self, send: Send, status: int, code: str, request_id: str, *, retry_after: int | None = None + ) -> None: body = json.dumps( { "status": status, @@ -176,8 +178,10 @@ async def problem(self, send: Send, status: int, code: str, request_id: str) -> ).encode() headers = [(b"content-type", b"application/problem+json"), (b"content-length", str(len(body)).encode())] headers.extend([(b"x-weave-wire-version", b"weave/api-v1"), (b"x-weave-request-id", request_id.encode())]) - if status == 429: - headers.append((b"retry-after", b"1")) + if retry_after is None and status == 429: + retry_after = 1 + if retry_after is not None: + headers.append((b"retry-after", str(retry_after).encode())) await send({"type": "http.response.start", "status": status, "headers": headers}) await send({"type": "http.response.body", "body": body}) @@ -237,7 +241,9 @@ async def _serve(self, scope: Scope, receive: Receive, send: Send) -> None: or "/operations/compatibility" in path ) if compatibility is not None and not compatibility.ready and not restricted_safe: - await self.problem(send, 503, "WV-COMPATIBILITY", request_id) + await self.problem( + send, 503, "WV-COMPATIBILITY", request_id, retry_after=compatibility.retry_after_seconds() + ) return headers = scope.get("headers", []) if sum(len(k) + len(v) for k, v in headers) > 16_384: @@ -314,7 +320,7 @@ async def buffered_receive() -> Message: except CatalogError as error: if started: raise - await self.problem(send, error.status, error.code, request_id) + await self.problem(send, error.status, error.code, request_id, retry_after=error.retry_after) except Exception: logging.getLogger("weave.operations").error("Request failed", extra={"error_code": "WV-INTERNAL"}) if not started: diff --git a/studio/src/app/api.ts b/studio/src/app/api.ts index a3c2f3e1..93558e93 100644 --- a/studio/src/app/api.ts +++ b/studio/src/app/api.ts @@ -69,13 +69,25 @@ const capacityCodes = new Set([ "WV-STUDIO-BUSY", ]); const capacityAttempts = 4; -/** `Retry-After` in seconds, kept between a quarter second and five seconds. */ +/** The longest pause, in seconds, before a rejected request is sent again. */ +const longestPause = 5; +/** `Retry-After` in seconds, kept between a quarter second and the longest pause. */ const retryDelay = (header: string | null) => { const seconds = Number(header); return Number.isFinite(seconds) && header !== null && header.trim() !== "" - ? Math.min(Math.max(seconds, 0.25), 5) * 1000 + ? Math.min(Math.max(seconds, 0.25), longestPause) * 1000 : 1000; }; +/** + * The platform refused a change before it ran: for capacity, or while it confirms + * compatibility. A compatibility refusal is sent again only when the platform's next + * check falls within one pause; a longer restriction fails at once. + */ +const replayable = (status: number, code: string, retryAfter: string | null) => + (status === 429 && capacityCodes.has(code)) || + (status === 503 && + code === "WV-COMPATIBILITY" && + !(Number(retryAfter) > longestPause)); export class StudioApi { session: Session = { paired: false, @@ -105,9 +117,9 @@ export class StudioApi { if (body !== undefined) h["Content-Type"] = "application/json"; if (method !== "GET" && this.session.csrfToken) h["X-Weave-CSRF"] = this.session.csrfToken; - // A capacity rejection means the host or platform did not admit the request, so a - // read, or a change that carries an idempotency key, can be sent again. - const replayable = method === "GET" || "Idempotency-Key" in h; + // A capacity or compatibility rejection means the host or platform did not admit the + // request, so a read, or a change that carries an idempotency key, can be sent again. + const safe = method === "GET" || "Idempotency-Key" in h; for (let attempt = 1; ; attempt++) { const response = await fetch(path, { method, @@ -122,10 +134,13 @@ export class StudioApi { if (response.ok) return value as T; const error = new ApiError(response.status, value); if ( - replayable && + safe && attempt < capacityAttempts && - response.status === 429 && - capacityCodes.has(error.code) + replayable( + response.status, + error.code, + response.headers.get("Retry-After"), + ) ) { await this.wait(retryDelay(response.headers.get("Retry-After"))); continue; diff --git a/studio/tests/api.test.ts b/studio/tests/api.test.ts index 5c57558c..705384c8 100644 --- a/studio/tests/api.test.ts +++ b/studio/tests/api.test.ts @@ -196,6 +196,58 @@ describe("capacity rejections", () => { }); expect(fetch).toHaveBeenCalledTimes(1); }); + it("replays a keyed change the platform refused while confirming compatibility", async () => { + const restricted = (retryAfter: string) => + new Response( + JSON.stringify({ + code: "WV-COMPATIBILITY", + message: "Request unavailable", + }), + { + status: 503, + headers: { + "Content-Type": "application/json", + "Retry-After": retryAfter, + }, + }, + ); + const fetch = vi + .fn() + .mockResolvedValueOnce(restricted("2")) + .mockResolvedValueOnce(ok()); + vi.stubGlobal("fetch", fetch); + const client = api(); + await expect( + client.mutate("/studio/api/runs", "POST", {}, undefined, "run-key"), + ).resolves.toEqual({ items: [] }); + expect(fetch).toHaveBeenCalledTimes(2); + expect(client.wait).toHaveBeenCalledWith(2000); + const unkeyed = vi.fn().mockResolvedValue(restricted("2")); + vi.stubGlobal("fetch", unkeyed); + await expect( + api().request("/studio/api/runs", "POST", {}), + ).rejects.toMatchObject({ status: 503, code: "WV-COMPATIBILITY" }); + expect(unkeyed).toHaveBeenCalledTimes(1); + // A restriction that outlasts the replay window fails at once. + const lasting = vi.fn().mockResolvedValue(restricted("45")); + vi.stubGlobal("fetch", lasting); + const waiting = api(); + await expect( + waiting.mutate("/studio/api/runs", "POST", {}, undefined, "run-key"), + ).rejects.toMatchObject({ status: 503, code: "WV-COMPATIBILITY" }); + expect(lasting).toHaveBeenCalledTimes(1); + expect(waiting.wait).not.toHaveBeenCalled(); + const other = vi + .fn() + .mockResolvedValue( + new Response(JSON.stringify({ code: "WV-INTERNAL" }), { status: 503 }), + ); + vi.stubGlobal("fetch", other); + await expect(api().request("/studio/api/runs")).rejects.toMatchObject({ + status: 503, + }); + expect(other).toHaveBeenCalledTimes(1); + }); it("does not replay other 429 answers", async () => { const fetch = vi.fn().mockResolvedValue(busy("WV-PAGE-LIMIT")); vi.stubGlobal("fetch", fetch); diff --git a/tests/integration/test_operations.py b/tests/integration/test_operations.py index 24776dc9..7ecf7825 100644 --- a/tests/integration/test_operations.py +++ b/tests/integration/test_operations.py @@ -348,7 +348,9 @@ async def test_native_inventory_restricts_effects_but_keeps_terminal_control( assert (await client.get("/health/live")).status_code == 200 assert (await client.get("/health/ready")).status_code == 503 assert (await client.get(env_url + "/runs", headers=headers)).status_code == 200 - assert (await client.post(env_url + "/runs", headers=headers, json={})).status_code == 503 + refused = await client.post(env_url + "/runs", headers=headers, json={}) + assert refused.status_code == 503 and refused.json()["code"] == "WV-COMPATIBILITY" + assert 1 <= int(refused.headers["retry-after"]) <= 65 cancelled = await client.post( f"{env_url}/runs/{waiting_run}/cancel", headers=headers, json={"reason": "restricted owned trial"} ) @@ -356,6 +358,80 @@ async def test_native_inventory_restricts_effects_but_keeps_terminal_control( assert cancelled.json()["status"] == "cancelled" and cancelled.json()["unavailable"] +async def test_native_rescan_stays_ready_while_requests_hold_every_control_slot( + waiting_run, client, headers, env_url, monkeypatch +): + import asyncio + + from firefly_weave.operations.execution import CONTROL_SLOTS + + app = client._transport.app + compatibility = app.state.compatibility + assert compatibility.ready + held, release = asyncio.Event(), asyncio.Event() + holding = 0 + database_ready = app.state.resources.is_ready + + async def slow_probe(): + nonlocal holding + holding += 1 + if holding == CONTROL_SLOTS: + held.set() + await release.wait() + return await database_ready() + + # Each in-flight read keeps its control slot until it answers. + monkeypatch.setattr(app.state.resources, "is_ready", slow_probe) + probes = [asyncio.create_task(client.get("/health/ready")) for _ in range(CONTROL_SLOTS)] + try: + async with asyncio.timeout(10): + await held.wait() + assert (await client.get(env_url + "/runs", headers=headers)).status_code == 429 + report = await compatibility.scan() + assert report.mode == "ready" and report.complete and report.inspected > 0, report.findings + assert compatibility.ready + finally: + release.set() + answered = await asyncio.gather(*probes) + assert [probe.status_code for probe in answered] == [200] * CONTROL_SLOTS + + +async def test_native_run_start_is_admitted_while_a_confirming_rescan_cleans_up( + waiting_run, author, headers, env_url, monkeypatch +): + import asyncio + + compatibility = author[0]._transport.app.state.compatibility + assert compatibility.ready + run = await author[0].get(f"{env_url}/runs/{waiting_run}", headers=headers) + activation = run.json()["activation"]["id"] + cleaning, release = asyncio.Event(), asyncio.Event() + dispose = compatibility._dispose_catalog + + async def held_cleanup(): + if compatibility._catalog_engine is not None: + cleaning.set() + await release.wait() + await dispose() + + monkeypatch.setattr(compatibility, "_dispose_catalog", held_cleanup) + scan = asyncio.create_task(compatibility.scan()) + try: + async with asyncio.timeout(10): + await cleaning.wait() + assert compatibility.ready + started = await author[0].post( + env_url + "/runs", + headers={**headers, "Idempotency-Key": "during-rescan-cleanup"}, + json={"activation_id": activation, "input": {}}, + ) + assert started.status_code == 201, started.text + finally: + release.set() + report = await scan + assert report.mode == "ready" and compatibility.ready + + async def test_native_committed_transition_and_request_emit_safe_telemetry( waiting_run, client, headers, env_url, monkeypatch ): diff --git a/tests/unit/operations/test_compatibility_rescan.py b/tests/unit/operations/test_compatibility_rescan.py new file mode 100644 index 00000000..d08ad532 --- /dev/null +++ b/tests/unit/operations/test_compatibility_rescan.py @@ -0,0 +1,294 @@ +# 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 periodic rescan neither competes with request admission nor withdraws a verdict it confirms.""" + +import asyncio +import logging +from contextlib import asynccontextmanager +from types import SimpleNamespace +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest + +import firefly_weave.operations.compatibility as module +from firefly_weave.definitions.models import CatalogError +from firefly_weave.operations.execution import CONTROL_SLOTS, request_execution +from firefly_weave.settings import Settings + + +class Catalog: + """One tenant with one retained row; disposal can be held or made to fail.""" + + def __init__(self, settings: Settings) -> None: + self.settings = settings + self.disposing = asyncio.Event() + self.release = asyncio.Event() + self.release.set() + self.dispose_error: BaseException | None = None + self.commit_error: BaseException | None = None + self.tenant = uuid4() + + @asynccontextmanager + async def connect(self): + yield self + + @asynccontextmanager + async def begin(self): + yield self + if self.commit_error is not None: + raise self.commit_error + + async def execute(self, stmt, params=None): + sql = str(stmt) + if "weave_compatibility_tenants" in sql: + tenants = [] if params["after"] else [self.tenant] + return SimpleNamespace(scalars=lambda: tenants) + if "weave_compatibility_page" in sql: + rows = ( + [] + if params["kind"] + else [ + { + "kind": "workflow", + "id": uuid4(), + "project_id": uuid4(), + "environment_id": None, + "payload": {}, + "facts_valid": True, + } + ] + ) + return SimpleNamespace(mappings=lambda: SimpleNamespace(all=lambda: rows)) + return SimpleNamespace() + + async def scalar(self, stmt): + return self.settings.operations.fingerprint if "policy" in str(stmt) else {} + + def hold(self) -> None: + self.disposing.clear() + self.release.clear() + + async def dispose(self): + self.disposing.set() + await self.release.wait() + if self.dispose_error is not None: + raise self.dispose_error + + +@pytest.fixture +async def rescan(monkeypatch): + settings = Settings( + database_url="postgresql+asyncpg://app@localhost/test", + scheduler_database_url="postgresql+asyncpg://catalog@localhost/test", + scheduler_enabled=False, + ) + catalog = Catalog(settings) + compatible = [] + monkeypatch.setattr(module, "check_inventory_authority", AsyncMock()) + monkeypatch.setattr(module, "create_async_engine", lambda *a, **k: catalog) + monkeypatch.setattr(module, "classify_inventory_row", lambda *a: compatible.append(a) and None) + service = module.CompatibilityService( + settings, None, None, None, SimpleNamespace(record=lambda *a, **k: None, inventory=lambda *a, **k: None) + ) + service.on_ready = AsyncMock() + service.on_restricted = AsyncMock() + await service.open() + try: + yield service, catalog, compatible + finally: + catalog.dispose_error = None + catalog.release.set() + await service.close() + + +@asynccontextmanager +async def busy_requests(count: int = CONTROL_SLOTS): + """Hold every control slot the way in-flight GET requests do.""" + entered, release = asyncio.Event(), asyncio.Event() + holding = 0 + + async def request(): + nonlocal holding + async with request_execution(control=True): + holding += 1 + if holding == count: + entered.set() + await release.wait() + + tasks = [asyncio.create_task(request()) for _ in range(count)] + try: + async with asyncio.timeout(1): + await entered.wait() + yield + finally: + release.set() + await asyncio.gather(*tasks) + + +async def test_request_burst_cannot_turn_a_compatible_inventory_restricted(rescan): + service, _, compatible = rescan + assert (await service.scan()).mode == "ready" + async with busy_requests(): + report = await service.scan() + with pytest.raises(CatalogError, match="capacity"): + async with request_execution(control=True): + raise AssertionError("a fifth control request was admitted") + assert report.mode == "ready" and report.complete and not report.findings + assert service.ready and len(compatible) == 2 + service.on_restricted.assert_not_awaited() + + +async def test_confirmed_ready_verdict_stays_in_force_through_catalog_cleanup(rescan): + service, catalog, _ = rescan + await service.scan() + assert service.ready + observed = [] + + async def on_ready(): + observed.append(service.ready) + + service.on_ready = on_ready + catalog.hold() + scan = asyncio.create_task(service.scan()) + await catalog.disposing.wait() + assert service.ready, "a rescan that confirms readiness withdrew it during catalog cleanup" + catalog.release.set() + report = await scan + assert report.mode == "ready" and service.ready and observed == [True] + + +async def test_confirmed_ready_verdict_is_withdrawn_when_cleanup_fails(rescan): + service, catalog, _ = rescan + await service.scan() + catalog.hold() + catalog.dispose_error = RuntimeError("fixture disposal failure") + scan = asyncio.create_task(service.scan()) + await catalog.disposing.wait() + assert service.ready + catalog.release.set() + with pytest.raises(RuntimeError, match="fixture disposal failure"): + await scan + assert not service.ready + assert any(f.code == "inventory_incomplete" for f in service.report.findings) + service.on_restricted.assert_awaited_once() + + +async def test_unexpected_interruption_after_traversal_withdraws_readiness(rescan): + class Interrupted(BaseException): + pass + + service, catalog, _ = rescan + await service.scan() + catalog.commit_error = Interrupted() + with pytest.raises(Interrupted): + await service.scan() + assert not service.ready + + +async def test_confirmed_ready_verdict_is_withdrawn_when_the_ready_callback_fails(rescan): + service, _, _ = rescan + await service.scan() + service.on_ready = AsyncMock(side_effect=RuntimeError("fixture effect failure")) + with pytest.raises(RuntimeError, match="fixture effect failure"): + await service.scan() + assert not service.ready + + +async def test_transition_into_ready_stays_withdrawn_until_effects_are_initialized(rescan): + service, catalog, _ = rescan + assert not service.ready + observed = [] + + async def on_ready(): + observed.append(service.ready) + + service.on_ready = on_ready + catalog.hold() + scan = asyncio.create_task(service.scan()) + await catalog.disposing.wait() + assert not service.ready + catalog.release.set() + await scan + assert observed == [False] and service.ready + + +async def test_restricting_rescan_withdraws_before_cleanup_and_names_its_cause(rescan, monkeypatch, caplog): + service, catalog, _ = rescan + await service.scan() + + def refuse(*args): + raise CatalogError(429, "WV-OPERATION-CAPACITY", "Pure execution capacity unavailable") + + monkeypatch.setattr(module, "classify_inventory_row", refuse) + catalog.hold() + with caplog.at_level(logging.WARNING, logger=module.__name__): + scan = asyncio.create_task(service.scan()) + await catalog.disposing.wait() + assert not service.ready + catalog.release.set() + report = await scan + assert report.mode == "restricted" + service.on_restricted.assert_awaited_once() + [message] = [r.getMessage() for r in caplog.records if r.name == module.__name__] + assert "withdrew readiness" in message + assert "inventory:inventory_incomplete" in message + assert "CatalogError WV-OPERATION-CAPACITY" in message + assert "capacity unavailable" not in message + + +async def test_restricted_rescan_that_stays_restricted_logs_nothing(rescan, caplog): + service, _, _ = rescan + service.settings = service.settings.model_copy(update={"scheduler_database_url": None}) + with caplog.at_level(logging.WARNING, logger=module.__name__): + await service.scan() + await service.scan() + assert not service.ready + assert not [r for r in caplog.records if r.name == module.__name__] + + +async def test_retry_after_counts_down_to_the_next_scheduled_rescan(monkeypatch): + clock = [100.0] + monkeypatch.setattr(module, "time", SimpleNamespace(monotonic=lambda: clock[0])) + service = module.CompatibilityService(None, None, None, None, None) + rescan_after = module.INVENTORY_RESCAN_SECONDS + module.INVENTORY_AGE_SECONDS + assert service.retry_after_seconds() == int(rescan_after) + await service.open() + try: + assert service.retry_after_seconds() == int(rescan_after) + clock[0] += 50.5 + assert service.retry_after_seconds() == int(rescan_after - 50) + clock[0] += 3600 + assert service.retry_after_seconds() == 1 + finally: + await service.close() + + +async def test_operational_refusal_carries_retry_after_to_the_error_response(): + from firefly_weave.api.errors import ErrorAdvice + from firefly_weave.connections.registry import ConnectorRegistry + + registry = ConnectorRegistry() + registry.set_operational_guard(lambda: False, retry_after=lambda: 23) + with pytest.raises(CatalogError) as refused: + registry.require_operational() + response = await ErrorAdvice().catalog_error(refused.value) + assert (response.status_code, response.headers["retry-after"]) == (503, "23") + registry.set_operational_guard(lambda: True) + registry.require_operational() + other = await ErrorAdvice().catalog_error(CatalogError(404, "WV-NOT-FOUND", "Not found")) + assert "retry-after" not in other.headers diff --git a/tests/unit/operations/test_execution.py b/tests/unit/operations/test_execution.py index 2042389c..66ff2001 100644 --- a/tests/unit/operations/test_execution.py +++ b/tests/unit/operations/test_execution.py @@ -133,3 +133,48 @@ async def run(): assert await asyncio.to_thread(exited.wait, 2) async with debug_execution(): assert await execute_pure(lambda: "debug-recovered") == "debug-recovered" + + +async def test_inventory_slot_is_reserved_apart_from_request_admission(): + from firefly_weave.operations.execution import CONTROL_SLOTS, inventory_execution, request_execution + + entered, release, exited = threading.Event(), threading.Event(), threading.Event() + + def blocked(): + entered.set() + try: + release.wait(5) + finally: + exited.set() + + async def classify(): + async with inventory_execution(): + await execute_pure(blocked, control=True) + + requests = [request_execution() for _ in range(2)] + [request_execution(control=True) for _ in range(CONTROL_SLOTS)] + for request in requests: + await request.__aenter__() + try: + async with inventory_execution(): + assert await execute_pure(lambda: "classified", control=True) == "classified" + assert await execute_pure(lambda: "classified again", control=True) == "classified again" + with pytest.raises(CatalogError, match="capacity"): + async with inventory_execution(): + raise AssertionError("a second inventory owner was admitted") + finally: + for request in reversed(requests): + await request.__aexit__(None, None, None) + task = asyncio.create_task(classify()) + try: + assert await asyncio.to_thread(entered.wait, 2) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + with pytest.raises(CatalogError, match="capacity"): + async with inventory_execution(): + raise AssertionError("a cancelled classification was still running") + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + assert await asyncio.to_thread(exited.wait, 2) + async with inventory_execution(): + assert await execute_pure(lambda: "recovered", control=True) == "recovered" diff --git a/tests/unit/operations/test_transport.py b/tests/unit/operations/test_transport.py index e7f68f58..c3982b32 100644 --- a/tests/unit/operations/test_transport.py +++ b/tests/unit/operations/test_transport.py @@ -245,3 +245,38 @@ async def send(message): await asyncio.gather(*held, return_exceptions=True) assert reservations.requests == reservations.bytes == reservations.debug == 0 assert (await invoke(app, [b"{}"], path="/compiler/compile"))[0]["status"] == 200 + + +async def test_compatibility_refusals_tell_clients_when_a_rescan_can_lift_them(): + from types import SimpleNamespace + + from firefly_weave.definitions.models import CatalogError + from firefly_weave.operations.transport import BodyBoundary + + compatibility = SimpleNamespace(ready=False, retry_after_seconds=lambda: 17) + + async def refused(scope, receive, send): + await receive() + raise CatalogError(503, "WV-COMPATIBILITY", "Execution compatibility unavailable", retry_after=9) + + async def call(method): + sent = [] + + async def receive(): + return {"type": "http.request", "body": b"{}", "more_body": False} + + async def send(message): + sent.append(message) + + scope = {"type": "http", "method": method, "path": "/runs", "headers": []} + scope["app"] = SimpleNamespace(state=SimpleNamespace(compatibility=compatibility)) + await BodyBoundary(refused)(scope, receive, send) + return sent[0]["status"], dict(sent[0]["headers"]), json.loads(sent[-1]["body"])["code"] + + # Restricted: the boundary refuses before the body and names the next rescan. + status, headers, code = await call("POST") + assert (status, code, headers[b"retry-after"]) == (503, "WV-COMPATIBILITY", b"17") + # Ready at admission, refused inside the request: the refusal keeps its own hint. + compatibility.ready = True + status, headers, code = await call("POST") + assert (status, code, headers[b"retry-after"]) == (503, "WV-COMPATIBILITY", b"9")