Keep compatibility rescans from refusing work they confirm

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) |

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

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

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:

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

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)

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

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

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)

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: