diff --git a/.github/actions/package/verify-installed/verify.py b/.github/actions/package/verify-installed/verify.py index 2b58966..cda3afd 100644 --- a/.github/actions/package/verify-installed/verify.py +++ b/.github/actions/package/verify-installed/verify.py @@ -47,6 +47,7 @@ def main() -> None: raise SystemExit("Installed Provider version is incorrect") required_api = { + "HealthStatus", "ManifestError", "Platform", "ProtocolVersion", diff --git a/.github/actions/provider/assert-health/action.yml b/.github/actions/provider/assert-health/action.yml new file mode 100644 index 0000000..6dd5d1e --- /dev/null +++ b/.github/actions/provider/assert-health/action.yml @@ -0,0 +1,40 @@ +name: Assert standard Provider Health behavior +description: Exercise Check, ordered Watch changes, and shutdown on the installed endpoint. + +inputs: + container-name: + description: Fedora container running the Agent and Provider. + required: true + unit-name: + description: Transient systemd unit running the Provider fixture. + required: true + socket-path: + description: Generation-specific Provider Unix socket. + required: true + evidence-directory: + description: Host directory receiving bounded health evidence. + required: true + +runs: + using: composite + steps: + - name: Exercise standard Check, Watch, and shutdown behavior + shell: bash + env: + ACTION_PATH: ${{ github.action_path }} + CONTAINER_NAME: ${{ inputs.container-name }} + EVIDENCE_DIRECTORY: ${{ inputs.evidence-directory }} + SOCKET_PATH: ${{ inputs.socket-path }} + UNIT_NAME: ${{ inputs.unit-name }} + run: | + set -euo pipefail + mkdir -p "${EVIDENCE_DIRECTORY}" + verifier_directory="/opt/stateforge-provider-tests/health" + verifier="${verifier_directory}/assert_health.py" + docker exec "${CONTAINER_NAME}" mkdir -p "${verifier_directory}" + docker cp "${ACTION_PATH}/assert_health.py" \ + "${CONTAINER_NAME}:${verifier}" + docker exec "${CONTAINER_NAME}" /usr/bin/python3 "${verifier}" \ + --socket-path "${SOCKET_PATH}" \ + --unit-name "${UNIT_NAME}" \ + > "${EVIDENCE_DIRECTORY}/health.json" diff --git a/.github/actions/provider/assert-health/assert_health.py b/.github/actions/provider/assert-health/assert_health.py new file mode 100644 index 0000000..e54e4b4 --- /dev/null +++ b/.github/actions/provider/assert-health/assert_health.py @@ -0,0 +1,181 @@ +"""Verify standard Provider Health behavior against an installed runtime.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import re +import subprocess +import sys + +import grpc +from grpc_health.v1 import health_pb2, health_pb2_grpc + +_BOOTSTRAP_SERVICE = "stateforge.sfp.bootstrap.v1.BootstrapService" +_MANIFEST_SERVICE = "stateforge.sfp.manifest.v1.ManifestService" + + +async def _read_status( + watch: grpc.aio.UnaryStreamCall[ + health_pb2.HealthCheckRequest, + health_pb2.HealthCheckResponse, + ], +) -> str: + """Read one bounded Watch response and return its enum name.""" + response = await asyncio.wait_for(watch.read(), timeout=10) + if response is grpc.aio.EOF: + raise RuntimeError("Health Watch ended before the expected transition") + return health_pb2.HealthCheckResponse.ServingStatus.Name(response.status) + + +def _signal(unit_name: str, signal_name: str) -> None: + """Deliver one deterministic fixture transition through systemd.""" + if not re.fullmatch( + r"stateforge-test-[a-z][a-z0-9-]{0,20}\.service", + unit_name, + ): + raise ValueError("unsafe Provider fixture unit name") + if signal_name not in {"SIGHUP", "SIGUSR1", "SIGUSR2"}: + raise ValueError("unsupported Provider fixture health signal") + # Both arguments are allow-listed above and the executable is absolute. + subprocess.run( # noqa: S603 + ["/usr/bin/systemctl", "kill", "--signal", signal_name, unit_name], + check=True, + timeout=10, + ) + + +def _stop(unit_name: str) -> None: + """Stop the fixture while its overall Health Watch remains active.""" + if not re.fullmatch( + r"stateforge-test-[a-z][a-z0-9-]{0,20}\.service", + unit_name, + ): + raise ValueError("unsafe Provider fixture unit name") + subprocess.run( # noqa: S603 + ["/usr/bin/systemctl", "stop", unit_name], + check=True, + timeout=20, + ) + + +async def _assert_shutdown_transition( + watch: grpc.aio.UnaryStreamCall[ + health_pb2.HealthCheckRequest, + health_pb2.HealthCheckResponse, + ], + unit_name: str, +) -> str: + """Stop the Provider and require fail-closed Watch delivery.""" + stop = asyncio.create_task(asyncio.to_thread(_stop, unit_name)) + try: + observed = await _read_status(watch) + finally: + await stop + if observed != "NOT_SERVING": + raise RuntimeError( + f"Provider shutdown produced {observed}, expected NOT_SERVING" + ) + return observed + + +async def _verify(socket_path: str, unit_name: str) -> dict[str, object]: + """Exercise Check, Watch, author transitions, and hook failure.""" + channel = grpc.aio.insecure_channel(f"unix:{socket_path}") + await asyncio.wait_for(channel.channel_ready(), timeout=10) + stub = health_pb2_grpc.HealthStub(channel) + checks: dict[str, str] = {} + transitions: list[str] = [] + watch = stub.Watch(health_pb2.HealthCheckRequest(service="")) + unknown_watch = None + try: + for service in ("", _BOOTSTRAP_SERVICE, _MANIFEST_SERVICE): + response = await stub.Check( + health_pb2.HealthCheckRequest(service=service), + timeout=10, + ) + name = health_pb2.HealthCheckResponse.ServingStatus.Name( + response.status + ) + if name != "SERVING": + raise RuntimeError(f"{service!r} reported {name}") + checks[service or ""] = name + + unknown_service = "stateforge.sfp.unknown.v1.UnknownService" + try: + await stub.Check( + health_pb2.HealthCheckRequest(service=unknown_service), + timeout=10, + ) + except grpc.aio.AioRpcError as error: + if error.code() is not grpc.StatusCode.NOT_FOUND: + raise + checks[unknown_service] = error.code().name + else: + raise RuntimeError("unknown Health Check did not return NOT_FOUND") + + unknown_watch = stub.Watch( + health_pb2.HealthCheckRequest(service=unknown_service) + ) + unknown_initial = await _read_status(unknown_watch) + if unknown_initial != "SERVICE_UNKNOWN": + raise RuntimeError( + f"unknown Health Watch reported {unknown_initial}" + ) + + transitions.append(await _read_status(watch)) + for signal_name, expected in ( + ("SIGUSR1", "NOT_SERVING"), + ("SIGUSR2", "SERVING"), + ("SIGHUP", "NOT_SERVING"), + ("SIGUSR2", "SERVING"), + ): + _signal(unit_name, signal_name) + observed = await _read_status(watch) + if observed != expected: + raise RuntimeError( + f"{signal_name} produced {observed}, expected {expected}" + ) + transitions.append(observed) + + transitions.append(await _assert_shutdown_transition(watch, unit_name)) + finally: + watch.cancel() + if unknown_watch is not None: + unknown_watch.cancel() + await channel.close() + + expected_transitions = [ + "SERVING", + "NOT_SERVING", + "SERVING", + "NOT_SERVING", + "SERVING", + "NOT_SERVING", + ] + if transitions != expected_transitions: + raise RuntimeError(f"unexpected Health transitions: {transitions}") + return { + "checks": checks, + "unknownWatchInitial": unknown_initial, + "watchTransitions": transitions, + } + + +def _arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--socket-path", required=True) + parser.add_argument("--unit-name", required=True) + return parser.parse_args() + + +def main() -> None: + """Run the installed-runtime assertion and emit bounded JSON evidence.""" + arguments = _arguments() + evidence = asyncio.run(_verify(arguments.socket_path, arguments.unit_name)) + sys.stdout.write(json.dumps(evidence, sort_keys=True) + "\n") + + +if __name__ == "__main__": + main() diff --git a/.github/actions/provider/create-fixture/fixture.py b/.github/actions/provider/create-fixture/fixture.py index bf77fae..ce075b9 100644 --- a/.github/actions/provider/create-fixture/fixture.py +++ b/.github/actions/provider/create-fixture/fixture.py @@ -3,11 +3,15 @@ from __future__ import annotations import argparse +import asyncio +import contextlib +import signal from pathlib import Path from pydantic import SecretStr from stateforge_provider import ( + HealthStatus, Platform, Provider, ProviderModel, @@ -72,15 +76,49 @@ def __init__( state_directory=state_directory, ) self._event_log = event_log + self._health_status = HealthStatus.healthy() + self._fail_health = False + self._health_tasks: set[asyncio.Task[None]] = set() async def on_startup(self) -> None: """Record that author resources were initialized.""" + loop = asyncio.get_running_loop() + loop.add_signal_handler( + signal.SIGUSR1, + self._schedule_health, + HealthStatus.maintenance(), + False, + ) + loop.add_signal_handler( + signal.SIGUSR2, + self._schedule_health, + HealthStatus.healthy(), + False, + ) + loop.add_signal_handler( + signal.SIGHUP, + self._schedule_health, + HealthStatus.not_serving(), + True, + ) self._record("startup") async def on_shutdown(self) -> None: """Record that author resources were released.""" + loop = asyncio.get_running_loop() + for signal_number in (signal.SIGUSR1, signal.SIGUSR2, signal.SIGHUP): + with contextlib.suppress(NotImplementedError, ValueError): + loop.remove_signal_handler(signal_number) + if self._health_tasks: + await asyncio.gather(*self._health_tasks, return_exceptions=True) self._record("shutdown") + async def health(self) -> HealthStatus: + """Report deterministic health controlled by scenario signals.""" + if self._fail_health: + raise RuntimeError("fixture health hook failed") + return self._health_status + @operation("stateforge.test/echo@1") async def echo(self, request: EchoInput) -> EchoResult: """Declare one typed operation without invoking it in R11.""" @@ -91,6 +129,29 @@ def _record(self, event: str) -> None: with self._event_log.open("a", encoding="utf-8") as stream: stream.write(f"{event}\n") + def _schedule_health( + self, + status: HealthStatus, + fail_health: bool, + ) -> None: + """Schedule one event-driven public health refresh.""" + task = asyncio.create_task( + self._publish_health(status, fail_health), + ) + self._health_tasks.add(task) + task.add_done_callback(self._health_tasks.discard) + + async def _publish_health( + self, + status: HealthStatus, + fail_health: bool, + ) -> None: + """Update fixture evidence and invoke the public refresh API.""" + self._health_status = status + self._fail_health = fail_health + accepted = await self.refresh_health() + self._record(f"health:{accepted.value}") + def _arguments() -> argparse.Namespace: parser = argparse.ArgumentParser() diff --git a/.github/actions/provider/stop/action.yml b/.github/actions/provider/stop/action.yml index e0a8e61..5650b8b 100644 --- a/.github/actions/provider/stop/action.yml +++ b/.github/actions/provider/stop/action.yml @@ -41,7 +41,11 @@ runs: run: | set -euo pipefail mkdir -p "${EVIDENCE_DIRECTORY}" - docker exec "${CONTAINER_NAME}" systemctl stop "${UNIT_NAME}" + state="$(docker exec "${CONTAINER_NAME}" \ + systemctl is-active "${UNIT_NAME}" 2>/dev/null || true)" + if [[ "${state}" != "inactive" && "${state}" != "failed" ]]; then + docker exec "${CONTAINER_NAME}" systemctl stop "${UNIT_NAME}" + fi deadline=$((SECONDS + TIMEOUT_SECONDS)) while (( SECONDS < deadline )); do diff --git a/.github/actions/scenarios/health/standard-health/action.yml b/.github/actions/scenarios/health/standard-health/action.yml new file mode 100644 index 0000000..ed3a22b --- /dev/null +++ b/.github/actions/scenarios/health/standard-health/action.yml @@ -0,0 +1,94 @@ +name: Test standard Provider Health lifecycle +description: Verify Health Check and Watch through the installed Agent system. + +inputs: + agent-artifact-directory: + description: Directory containing the exact Agent RPM build. + required: true + agent-test-image-directory: + description: Bundled exact Agent Fedora test-image definition. + required: true + provider-artifact-directory: + description: Directory containing the exact Provider build. + required: true + protocol-artifact-directory: + description: Directory containing the exact Protocol build. + required: true + runtime-directory: + description: Scenario-owned directory under the runner temp directory. + required: true + evidence-directory: + description: Directory receiving scenario evidence. + required: true + +runs: + using: composite + steps: + - name: Install exact runtime artifacts + id: system + uses: ./.github/actions/agent/install + with: + agent-artifact-directory: ${{ inputs.agent-artifact-directory }} + agent-test-image-directory: ${{ inputs.agent-test-image-directory }} + provider-artifact-directory: ${{ inputs.provider-artifact-directory }} + protocol-artifact-directory: ${{ inputs.protocol-artifact-directory }} + evidence-directory: ${{ inputs.evidence-directory }}/installation + + - name: Start installed Agent + uses: ./.github/actions/agent/start + with: + container-name: ${{ steps.system.outputs.container-name }} + + - name: Create health-capable Provider fixture + id: fixture + uses: ./.github/actions/provider/create-fixture + with: + runtime-directory: ${{ inputs.runtime-directory }} + instance-name: health-provider + identity: stateforge.test.health + + - name: Start Provider fixture + id: provider + uses: ./.github/actions/provider/start + with: + container-name: ${{ steps.system.outputs.container-name }} + fixture-path: ${{ steps.fixture.outputs.fixture-path }} + instance-name: ${{ steps.fixture.outputs.instance-name }} + state-name: ${{ steps.fixture.outputs.state-name }} + identity: stateforge.test.health + evidence-directory: ${{ inputs.evidence-directory }}/provider + + - name: Assert Agent-accepted session + uses: ./.github/actions/agent/assert-session + with: + container-name: ${{ steps.system.outputs.container-name }} + provider-instance-id: ${{ steps.provider.outputs.provider-instance-id }} + endpoint-generation: ${{ steps.provider.outputs.endpoint-generation }} + evidence-directory: ${{ inputs.evidence-directory }}/agent + + - name: Assert standard Provider Health behavior + uses: ./.github/actions/provider/assert-health + with: + container-name: ${{ steps.system.outputs.container-name }} + unit-name: ${{ steps.provider.outputs.unit-name }} + socket-path: ${{ steps.provider.outputs.socket-path }} + evidence-directory: ${{ inputs.evidence-directory }}/health + + - name: Verify Provider shutdown cleanup + uses: ./.github/actions/provider/stop + with: + container-name: ${{ steps.system.outputs.container-name }} + unit-name: ${{ steps.provider.outputs.unit-name }} + descriptor-path: ${{ steps.provider.outputs.descriptor-path }} + socket-path: ${{ steps.provider.outputs.socket-path }} + event-log: ${{ steps.provider.outputs.event-log }} + evidence-directory: ${{ inputs.evidence-directory }}/provider + + - name: Preserve bounded evidence and clean up + if: ${{ always() }} + uses: ./.github/actions/evidence/collect + with: + stage: standard-health + container-name: ${{ steps.system.outputs.container-name }} + runtime-directory: ${{ inputs.runtime-directory }} + evidence-directory: ${{ inputs.evidence-directory }} diff --git a/.github/workflows/provider-runtime-scenarios.yml b/.github/workflows/provider-runtime-scenarios.yml index d15d256..ce1b998 100644 --- a/.github/workflows/provider-runtime-scenarios.yml +++ b/.github/workflows/provider-runtime-scenarios.yml @@ -62,6 +62,41 @@ jobs: if-no-files-found: warn retention-days: ${{ inputs.retention-days }} + health: + name: Standard Health / installed Agent RPM + runs-on: ubuntu-24.04 + timeout-minutes: 35 + steps: + - name: Check out scenario automation + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Download exact runtime bundle + uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8.0.2 + with: + name: ${{ inputs.artifact-name }} + path: ${{ runner.temp }}/stateforge-provider-bundle + + - name: Run standard-health scenario + uses: ./.github/actions/scenarios/health/standard-health + with: + agent-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/agent + agent-test-image-directory: ${{ runner.temp }}/stateforge-provider-bundle/agent/test-image + provider-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/provider + protocol-artifact-directory: ${{ runner.temp }}/stateforge-provider-bundle/protocol + runtime-directory: ${{ runner.temp }}/sfp-health + evidence-directory: evidence/runtime/standard-health + + - name: Upload runtime evidence + if: ${{ always() }} + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 + with: + name: ${{ inputs.evidence-prefix }}-standard-health + path: evidence/runtime/standard-health + if-no-files-found: warn + retention-days: ${{ inputs.retention-days }} + full: name: ${{ matrix.scenario }} / installed Agent RPM if: ${{ inputs.scenario-tier == 'full' }} diff --git a/README.md b/README.md index d4d20c4..5c811b3 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,9 @@ The runtime validates an immutable schema-1 manifest and serves it after handshake on the existing endpoint. See the [authoring guide](docs/manifest-authoring.md) and [complete example](examples/reference_provider.py). +The standard gRPC Health service is hosted on that same endpoint; authors +return a typed `HealthStatus` and never handle health protobuf messages. See +the [health guide](docs/health-reporting.md). Install the Provider framework from PyPI: @@ -23,7 +26,13 @@ python -m pip install stateforge-provider Import the public authoring API from `stateforge_provider`: ```python -from stateforge_provider import Platform, Provider, ProviderModel, operation +from stateforge_provider import ( + HealthStatus, + Platform, + Provider, + ProviderModel, + operation, +) ``` For repository development, install the project and its development tools diff --git a/docs/ci-automation.md b/docs/ci-automation.md index 1c7f734..38dc814 100644 --- a/docs/ci-automation.md +++ b/docs/ci-automation.md @@ -134,6 +134,8 @@ The runtime suite currently covers: - normal startup, Agent-observed Bootstrap and Manifest acceptance, graceful shutdown, and cleanup; +- standard Health Check and Watch on the same Provider socket, including + initial replay, maintenance, recovery, unknown service, and hook failure; - two distinct Provider installations running concurrently; - same-installation replacement with a stable identity and newer generation; - rejection of malformed Provider discovery data followed by valid recovery; diff --git a/docs/endpoint-lifecycle.md b/docs/endpoint-lifecycle.md index f4975a7..09d6ccc 100644 --- a/docs/endpoint-lifecycle.md +++ b/docs/endpoint-lifecycle.md @@ -6,18 +6,22 @@ For each Provider process generation, the runtime follows this order: 2. Load the stable Provider instance ID and allocate a new endpoint generation. 3. Call the Provider author's `on_startup()` hook. 4. Build and validate the complete immutable manifest and revision. -5. Compose the Bootstrap and Manifest procedure registrations on one gRPC - server. The endpoint derives its transport ceilings from those registrations - while each RPC retains its exact procedure limit. -6. Create the generation-specific Unix socket and start gRPC. -7. Secure the socket ownership and permissions. -8. Atomically publish the descriptor, with both services ready. -9. Serve SFP, retaining negotiated sessions for manifest validation. -10. Withdraw the owned descriptor, invalidate sessions, and stop gRPC with +5. Evaluate the typed Provider health hook. Failure remains local and leaves + the process-wide state fail-closed as `NOT_SERVING`. +6. Compose Bootstrap, Manifest, and the standard gRPC Health registration on + one gRPC server. Health registers the exact hosted SFP service names. The + endpoint derives its transport ceilings from those registrations while + each RPC retains its exact procedure limit. +7. Create the generation-specific Unix socket and start gRPC. +8. Secure the socket ownership and permissions. +9. Atomically publish the descriptor, with every service ready. +10. Serve SFP, retaining negotiated sessions for manifest validation. +11. Set Provider-wide and named health to `NOT_SERVING`, then withdraw the + owned descriptor, invalidate sessions, and stop gRPC with bounded grace so no new work can reach Provider-owned resources. -11. Call the Provider author's `on_shutdown()` hook within the configured +12. Call the Provider author's `on_shutdown()` hook within the configured shutdown timeout. -12. Verify and remove any remaining generation-specific socket path, then reset +13. Verify and remove any remaining generation-specific socket path, then reset the runtime state. The descriptor is published only after gRPC is ready. This prevents the Agent @@ -44,5 +48,8 @@ generation's descriptor. Manifest declarations are static for a generation; see [Manifest declarations](manifest-authoring.md) for session lifetime, limits, revision stability, and schema safety rules. +Health is event-driven rather than polled; see +[Health and readiness](health-reporting.md) for the public hook and Watch +semantics. See [Runtime package ownership](runtime-architecture.md) for the boundary between the shared endpoint and procedure-specific services. diff --git a/docs/health-reporting.md b/docs/health-reporting.md new file mode 100644 index 0000000..00a3357 --- /dev/null +++ b/docs/health-reporting.md @@ -0,0 +1,71 @@ +# Health and readiness + +Every Provider hosts the standard `grpc.health.v1.Health` service on the same +Unix-domain gRPC endpoint as Bootstrap and Manifest. The runtime owns the +protobuf and streaming adaptation; Provider authors return a typed +[`HealthStatus`][stateforge_provider.health.HealthStatus]. + +```python +from stateforge_provider import HealthStatus, Platform, Provider + + +class GitProvider(Provider): + identity = "stateforge.git" + display_name = "Git Provider" + version = "1.0.0" + platforms = (Platform("linux", "x86_64"),) + + async def health(self) -> HealthStatus: + if self._maintenance: + return HealthStatus.maintenance() + if not self._client.ready: + return HealthStatus.degraded() + return HealthStatus.healthy() +``` + +The runtime evaluates `health()` after `on_startup()` and before publishing the +endpoint descriptor. It begins fail-closed as `NOT_SERVING` and reports +`SERVING` only when startup, manifest construction, service registration, and +the author hook are ready. + +## Publishing a changed state + +The runtime does not poll third-party systems. When Provider-owned evidence +changes, call `await self.refresh_health()` from the Provider's own event or +monitoring path. This re-evaluates `health()` and publishes an ordered standard +Health Watch transition: + +```python +async def enter_maintenance(self) -> None: + self._maintenance = True + await self.refresh_health() +``` + +`DEGRADED`, `MAINTENANCE`, and `NOT_SERVING` all map to standard gRPC +`NOT_SERVING`. Their distinction is local author intent; it is not sent as a +diagnostic string. Unexpected hook exceptions and invalid return values are +logged locally and also fail closed as `NOT_SERVING`. + +## Wire behavior + +- The empty service name (`""`) is the authoritative Provider-wide state. +- The exact hosted Bootstrap and Manifest protobuf service names are also + registered for diagnostics. +- `Check` returns the current state and rejects unknown names with `NOT_FOUND`. +- `Watch` immediately replays the current state and then sends every ordered + transition. +- A Watch for an unknown name begins with `SERVICE_UNKNOWN` and stays open in + accordance with the standard health contract. +- Observer count, per-observer buffering, and delivery time are bounded. A + slow observer is terminated with `RESOURCE_EXHAUSTED` without blocking the + Provider. + +Calling `request_shutdown()` immediately changes all registered health keys to +`NOT_SERVING` before descriptor withdrawal, session invalidation, transport +shutdown, or author-resource cleanup. Health changes gate new work only; they +do not cancel or rewrite already admitted operations. + +Provider-wide `SERVING` does not prove that the Agent accepted the manifest, +that a target supports a particular operation, or that an operation is +authorized. The Agent derives operational eligibility from health together +with current transport, handshake, generation, session, and manifest state. diff --git a/docs/index.md b/docs/index.md index 0e0312c..7eb4ffa 100644 --- a/docs/index.md +++ b/docs/index.md @@ -24,7 +24,7 @@ generation, starts the protected local endpoint, and negotiates SFP with the Agent without any of that appearing in the author-facing API: ```python -from stateforge_provider import Platform, Provider +from stateforge_provider import HealthStatus, Platform, Provider class GitProvider(Provider): @@ -39,6 +39,10 @@ class GitProvider(Provider): async def on_shutdown(self) -> None: """Release Provider resources.""" + async def health(self) -> HealthStatus: + """Report whether this process can accept new SFP work.""" + return HealthStatus.healthy() + if __name__ == "__main__": GitProvider().run() @@ -46,6 +50,8 @@ if __name__ == "__main__": See [Manifest declarations](manifest-authoring.md) for typed operation models, triggers, configuration schemas, requirements, and explicit compliance roles. +See [Health and readiness](health-reporting.md) for fail-closed lifecycle +states and event-driven health updates. An application that already owns an event loop calls `await GitProvider().serve()` instead. See `.claude/PROVIDER_AUTHORING_VISION.md` for the complete intended author diff --git a/docs/reference/provider.md b/docs/reference/provider.md index 2853dbf..dc0c341 100644 --- a/docs/reference/provider.md +++ b/docs/reference/provider.md @@ -2,4 +2,6 @@ ::: stateforge_provider.provider.Provider +::: stateforge_provider.health.HealthStatus + ::: stateforge_provider.ProviderServerError diff --git a/docs/runtime-architecture.md b/docs/runtime-architecture.md index f8860dc..e815232 100644 --- a/docs/runtime-architecture.md +++ b/docs/runtime-architecture.md @@ -9,6 +9,7 @@ stateforge_provider/ _runtime/ # shared endpoint and process infrastructure _services/ bootstrap/ # Handshake implementation and local message limits + health/ # Standard gRPC Health adaptation and bounded Watch manifest/ # Manifest declaration, validation, and RPC service ``` @@ -41,6 +42,10 @@ Versioned protobuf messages, generated bindings, and contract constants remain in `stateforge_protocol.sfp..v1`. The Provider library imports those contracts; it does not copy or regenerate them. +Health is the intentional exception to that generated-module path: it uses the +official `grpc.health.v1` bindings from `grpcio-health-checking`, registers the +actual hosted SFP service names, and adds no StateForge health protobuf. + To add a future procedure: 1. create one `_services/` package; diff --git a/examples/reference_provider.py b/examples/reference_provider.py index 9ff87e1..1d54a72 100644 --- a/examples/reference_provider.py +++ b/examples/reference_provider.py @@ -3,6 +3,7 @@ from pydantic import SecretStr from stateforge_provider import ( + HealthStatus, Platform, Provider, ProviderModel, @@ -50,6 +51,10 @@ class ReferenceProvider(Provider): ) triggers = (Trigger("example.reference/changed@1", ReferenceChanged),) + async def health(self) -> HealthStatus: + """Report whether this Provider can accept new SFP work.""" + return HealthStatus.healthy() + @operation("example.reference/normalize@1") async def normalize(self, request: NormalizeInput) -> NormalizeResult: """Normalize a reference; R1 advertises but does not dispatch it.""" diff --git a/mkdocs.yml b/mkdocs.yml index 7c7f753..d3fc158 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -19,6 +19,7 @@ nav: - Endpoint lifecycle: endpoint-lifecycle.md - Runtime architecture: runtime-architecture.md - Manifest declarations: manifest-authoring.md + - Health and readiness: health-reporting.md - Package and runtime CI/CD: ci-automation.md - Python API: - Provider: reference/provider.md diff --git a/pyproject.toml b/pyproject.toml index cf436e6..a518f03 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,7 @@ classifiers = [ dependencies = [ "grpcio>=1.83", + "grpcio-health-checking>=1.83", "jsonschema>=4.25,<5", "pydantic>=2.12,<3", "referencing>=0.36,<1", @@ -52,6 +53,7 @@ dev = [ "pytest-cov>=6.2", "ruff>=0.12", "types-grpcio", + "types-grpcio-health-checking", "types-jsonschema", "types-protobuf", ] diff --git a/src/stateforge_provider/__init__.py b/src/stateforge_provider/__init__.py index ce8e188..bc0d248 100644 --- a/src/stateforge_provider/__init__.py +++ b/src/stateforge_provider/__init__.py @@ -9,10 +9,12 @@ Trigger, operation, ) +from stateforge_provider.health import HealthStatus from stateforge_provider.provider import Provider from stateforge_provider.version import ProtocolVersion __all__ = [ + "HealthStatus", "ManifestError", "Platform", "ProtocolVersion", diff --git a/src/stateforge_provider/_runtime/configuration.py b/src/stateforge_provider/_runtime/configuration.py index 5afd231..508fc2e 100644 --- a/src/stateforge_provider/_runtime/configuration.py +++ b/src/stateforge_provider/_runtime/configuration.py @@ -14,10 +14,18 @@ class ProviderRuntimeLimits: max_sessions: int = 1_024 max_json_nodes: int = 65_536 max_schema_nodes: int = 4_096 + max_health_watchers: int = 128 + health_watch_queue_capacity: int = 16 def __post_init__(self) -> None: """Reject unusable operational limits.""" - for name in ("max_sessions", "max_json_nodes", "max_schema_nodes"): + for name in ( + "max_sessions", + "max_json_nodes", + "max_schema_nodes", + "max_health_watchers", + "health_watch_queue_capacity", + ): value = getattr(self, name) if ( isinstance(value, bool) @@ -37,6 +45,7 @@ class ProviderRuntimeConfiguration: state_directory: Path discovery_directory: Path shutdown_timeout_seconds: float = 5.0 + health_watch_delivery_timeout_seconds: float = 1.0 limits: ProviderRuntimeLimits = field( default_factory=ProviderRuntimeLimits ) @@ -69,13 +78,17 @@ def __post_init__(self) -> None: value = getattr(self, name) if not isinstance(value, Path) or not value.is_absolute(): raise ValueError(f"{name} must be absolute") - timeout = self.shutdown_timeout_seconds - if ( - isinstance(timeout, bool) - or not isinstance(timeout, (int, float)) - or not math.isfinite(timeout) - or timeout <= 0 + for name in ( + "shutdown_timeout_seconds", + "health_watch_delivery_timeout_seconds", ): - raise ValueError("shutdown_timeout_seconds must be positive") + timeout = getattr(self, name) + if ( + isinstance(timeout, bool) + or not isinstance(timeout, (int, float)) + or not math.isfinite(timeout) + or timeout <= 0 + ): + raise ValueError(f"{name} must be positive") if not isinstance(self.limits, ProviderRuntimeLimits): raise ValueError("limits must be ProviderRuntimeLimits") diff --git a/src/stateforge_provider/_runtime/server.py b/src/stateforge_provider/_runtime/server.py index 5e3ec08..5c10bf0 100644 --- a/src/stateforge_provider/_runtime/server.py +++ b/src/stateforge_provider/_runtime/server.py @@ -48,6 +48,10 @@ [ProviderEndpointIdentity, SessionRegistry], Sequence[ProcedureRegistration], ] +SupplementalProcedureFactory = Callable[ + [Sequence[ProcedureRegistration]], + Sequence[ProcedureRegistration], +] @dataclass(frozen=True, slots=True) @@ -125,6 +129,9 @@ async def start( *, on_identity_allocated: Callable[[], Awaitable[None]] | None = None, procedure_factory: ProcedureFactory | None = None, + supplemental_procedure_factory: ( + SupplementalProcedureFactory | None + ) = None, ) -> None: """Start gRPC and atomically advertise the ready endpoint. @@ -167,6 +174,10 @@ async def start( registrations.extend( procedure_factory(self._allocated_identity, sessions) ) + if supplemental_procedure_factory is not None: + registrations.extend( + supplemental_procedure_factory(tuple(registrations)) + ) server = create_server(registrations) self._sessions = sessions self._server = server diff --git a/src/stateforge_provider/_services/health/__init__.py b/src/stateforge_provider/_services/health/__init__.py new file mode 100644 index 0000000..932b167 --- /dev/null +++ b/src/stateforge_provider/_services/health/__init__.py @@ -0,0 +1 @@ +"""Adapt Provider readiness to the standard gRPC Health service.""" diff --git a/src/stateforge_provider/_services/health/limits.py b/src/stateforge_provider/_services/health/limits.py new file mode 100644 index 0000000..89e84fa --- /dev/null +++ b/src/stateforge_provider/_services/health/limits.py @@ -0,0 +1,4 @@ +"""Bound standard gRPC Health messages on the shared Provider endpoint.""" + +MAX_REQUEST_BYTES = 4 * 1_024 +MAX_RESPONSE_BYTES = 64 diff --git a/src/stateforge_provider/_services/health/service.py b/src/stateforge_provider/_services/health/service.py new file mode 100644 index 0000000..256227c --- /dev/null +++ b/src/stateforge_provider/_services/health/service.py @@ -0,0 +1,234 @@ +"""Serve bounded standard gRPC Health Check and Watch requests.""" + +from __future__ import annotations + +import asyncio +from collections import defaultdict +from collections.abc import Iterable +from dataclasses import dataclass +from typing import TypeAlias + +import grpc +from grpc_health.v1 import health_pb2, health_pb2_grpc + +from stateforge_provider._runtime.grpc import ProcedureRegistration +from stateforge_provider._services.health.limits import ( + MAX_REQUEST_BYTES, + MAX_RESPONSE_BYTES, +) +from stateforge_provider.health import HealthStatus + +_HEALTH_SERVICE_NAME = "grpc.health.v1.Health" +_OVERALL_SERVICE_NAME = "" +_Status: TypeAlias = health_pb2.HealthCheckResponse.ServingStatus + + +@dataclass(eq=False, slots=True) +class _Observer: + """Own one bounded Watch queue and its overflow signal.""" + + queue: asyncio.Queue[_Status] + overflowed: asyncio.Event + + +class ProviderHealthService(health_pb2_grpc.HealthServicer): + """Publish Provider-wide health with bounded ordered Watch delivery.""" + + def __init__( + self, + *, + max_watchers: int, + watch_queue_capacity: int, + watch_delivery_timeout_seconds: float, + ) -> None: + self._max_watchers = max_watchers + self._watch_queue_capacity = watch_queue_capacity + self._watch_delivery_timeout_seconds = watch_delivery_timeout_seconds + self._statuses: dict[str, _Status] = { + _OVERALL_SERVICE_NAME: health_pb2.HealthCheckResponse.NOT_SERVING + } + self._observers: defaultdict[str, set[_Observer]] = defaultdict(set) + self._observer_count = 0 + self._stopping = False + + def register_services(self, service_names: Iterable[str]) -> None: + """Register hosted SFP services as initially not serving.""" + overall_status = self._statuses[_OVERALL_SERVICE_NAME] + for service_name in service_names: + if service_name and service_name != _HEALTH_SERVICE_NAME: + self._set_wire_status(service_name, overall_status) + + async def Check( # noqa: N802 + self, + request: health_pb2.HealthCheckRequest, + context: grpc.aio.ServicerContext[ + health_pb2.HealthCheckRequest, + health_pb2.HealthCheckResponse, + ], + ) -> health_pb2.HealthCheckResponse: + """Return the current status for a registered service.""" + await self._validate_request_size(request, context) + status = self._statuses.get(request.service) + if status is None: + await context.abort( + grpc.StatusCode.NOT_FOUND, + "Health service is not registered", + ) + return health_pb2.HealthCheckResponse(status=status) + + async def Watch( # noqa: N802 + self, + request: health_pb2.HealthCheckRequest, + context: grpc.aio.ServicerContext[ + health_pb2.HealthCheckRequest, + health_pb2.HealthCheckResponse, + ], + ) -> None: + """Stream the current status and every ordered status change.""" + await self._validate_request_size(request, context) + if self._observer_count >= self._max_watchers: + await context.abort( + grpc.StatusCode.RESOURCE_EXHAUSTED, + "Health Watch observer capacity exhausted", + ) + + observer = _Observer( + asyncio.Queue(maxsize=self._watch_queue_capacity), + asyncio.Event(), + ) + service = request.service + self._observers[service].add(observer) + self._observer_count += 1 + observer.queue.put_nowait( + self._statuses.get( + service, + health_pb2.HealthCheckResponse.SERVICE_UNKNOWN, + ) + ) + try: + while True: + status = await self._next_status(observer, context) + response = health_pb2.HealthCheckResponse(status=status) + async with asyncio.timeout( + self._watch_delivery_timeout_seconds + ): + await context.write(response) + except TimeoutError: + await context.abort( + grpc.StatusCode.RESOURCE_EXHAUSTED, + "Health Watch delivery deadline exceeded", + ) + finally: + observers = self._observers.get(service) + if observers is not None: + observers.discard(observer) + if not observers: + self._observers.pop(service, None) + self._observer_count -= 1 + + def publish(self, status: HealthStatus) -> None: + """Publish one author-facing state to all hosted service keys.""" + if self._stopping: + return + wire_status = ( + health_pb2.HealthCheckResponse.SERVING + if status.is_serving + else health_pb2.HealthCheckResponse.NOT_SERVING + ) + for service in tuple(self._statuses): + self._set_wire_status(service, wire_status) + + def begin_shutdown(self) -> None: + """Fail closed and permanently reject later readiness updates.""" + if self._stopping: + return + self._stopping = True + for service in tuple(self._statuses): + self._set_wire_status( + service, + health_pb2.HealthCheckResponse.NOT_SERVING, + ) + + def close(self) -> None: + """Wake any runtime-owned observers after transport shutdown.""" + self._stopping = True + for observers in tuple(self._observers.values()): + for observer in tuple(observers): + observer.overflowed.set() + + def registration( + self, + hosted_service_names: Iterable[str], + ) -> ProcedureRegistration: + """Build the standard Health registration for this runtime.""" + self.register_services(hosted_service_names) + + def install(server: grpc.aio.Server) -> None: + health_pb2_grpc.add_HealthServicer_to_server(self, server) + + return ProcedureRegistration( + name=_HEALTH_SERVICE_NAME, + max_request_bytes=MAX_REQUEST_BYTES, + max_response_bytes=MAX_RESPONSE_BYTES, + install=install, + ) + + async def _validate_request_size( + self, + request: health_pb2.HealthCheckRequest, + context: grpc.aio.ServicerContext[ + health_pb2.HealthCheckRequest, + health_pb2.HealthCheckResponse, + ], + ) -> None: + """Reject a service key outside the local Health request bound.""" + if request.ByteSize() > MAX_REQUEST_BYTES: + await context.abort( + grpc.StatusCode.RESOURCE_EXHAUSTED, + "Health request limit exceeded", + ) + + async def _next_status( + self, + observer: _Observer, + context: grpc.aio.ServicerContext[ + health_pb2.HealthCheckRequest, + health_pb2.HealthCheckResponse, + ], + ) -> _Status: + """Wait for a queued status while detecting observer overflow.""" + queued = asyncio.create_task(observer.queue.get()) + overflowed = asyncio.create_task(observer.overflowed.wait()) + tasks = (queued, overflowed) + try: + done, pending = await asyncio.wait( + tasks, + return_when=asyncio.FIRST_COMPLETED, + ) + except BaseException: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + if overflowed in done and overflowed.result(): + queued.cancel() + await asyncio.gather(queued, return_exceptions=True) + await context.abort( + grpc.StatusCode.RESOURCE_EXHAUSTED, + "Health Watch observer exceeded delivery capacity", + ) + return queued.result() + + def _set_wire_status(self, service: str, status: _Status) -> None: + """Record and fan out one changed status without blocking.""" + if self._statuses.get(service) == status: + return + self._statuses[service] = status + for observer in tuple(self._observers.get(service, ())): + try: + observer.queue.put_nowait(status) + except asyncio.QueueFull: + observer.overflowed.set() diff --git a/src/stateforge_provider/health.py b/src/stateforge_provider/health.py new file mode 100644 index 0000000..73eb93f --- /dev/null +++ b/src/stateforge_provider/health.py @@ -0,0 +1,45 @@ +"""Expose the typed Provider health states available to authors.""" + +from __future__ import annotations + +from enum import StrEnum +from typing import Self + + +class HealthStatus(StrEnum): + """Describe whether a Provider process can accept new SFP work. + + The Provider runtime maps every non-serving state to standard gRPC + ``NOT_SERVING``. The additional states let author code communicate intent + without exposing protobuf values or vendor diagnostics. + """ + + SERVING = "serving" + DEGRADED = "degraded" + MAINTENANCE = "maintenance" + NOT_SERVING = "not_serving" + + @property + def is_serving(self) -> bool: + """Return whether the status permits new SFP work.""" + return self is HealthStatus.SERVING + + @classmethod + def healthy(cls) -> Self: + """Return the normal ready-to-serve state.""" + return cls.SERVING + + @classmethod + def degraded(cls) -> Self: + """Return a fail-closed degraded state.""" + return cls.DEGRADED + + @classmethod + def maintenance(cls) -> Self: + """Return a fail-closed maintenance state.""" + return cls.MAINTENANCE + + @classmethod + def not_serving(cls) -> Self: + """Return the generic fail-closed state.""" + return cls.NOT_SERVING diff --git a/src/stateforge_provider/provider.py b/src/stateforge_provider/provider.py index 2c52d7f..c567705 100644 --- a/src/stateforge_provider/provider.py +++ b/src/stateforge_provider/provider.py @@ -4,7 +4,9 @@ import asyncio import contextlib +import logging import signal +from collections.abc import Sequence from pathlib import Path from typing import ClassVar, final @@ -16,6 +18,7 @@ from stateforge_provider._runtime.identity import ProviderEndpointIdentity from stateforge_provider._runtime.server import ProviderServer from stateforge_provider._runtime.sessions import SessionRegistry +from stateforge_provider._services.health.service import ProviderHealthService from stateforge_provider._services.manifest.declaration import ( ManifestDeclaration, ) @@ -28,6 +31,9 @@ RuntimeRequirement, Trigger, ) +from stateforge_provider.health import HealthStatus + +_LOGGER = logging.getLogger(__name__) class Provider: @@ -81,6 +87,8 @@ def __init__( ) self._server = ProviderServer(self._configuration) self._shutdown_requested = asyncio.Event() + self._health_service: ProviderHealthService | None = None + self._health_refresh_lock = asyncio.Lock() async def on_startup(self) -> None: """Override to initialize Provider resources before publication.""" @@ -94,6 +102,33 @@ async def on_shutdown(self) -> None: Must tolerate partially initialized resources and no ready endpoint. """ + async def health(self) -> HealthStatus: + """Report whether this Provider can accept new SFP work. + + Override this hook when readiness depends on Provider-owned resources. + Return only a typed :class:`HealthStatus`; raw probe details and + exceptions remain local to the Provider process. Call + :meth:`refresh_health` when the underlying readiness evidence changes. + """ + return HealthStatus.healthy() + + @final + async def refresh_health(self) -> HealthStatus: + """Re-evaluate the health hook and publish a fail-closed result. + + Returns: + The accepted author-facing health state. Hook failures and invalid + return values are logged locally and become ``NOT_SERVING``. + + Raises: + ProviderServerError: The Provider runtime is not active. + """ + service = self._health_service + if service is None: + raise ProviderServerError("Provider health runtime is not active") + async with self._health_refresh_lock: + return await self._refresh_health(service) + @final def run(self) -> None: """Run to completion with an owned event loop and signal handlers.""" @@ -119,12 +154,23 @@ async def serve(self) -> None: type(self), limits=self._configuration.limits, ) + health_service = ProviderHealthService( + max_watchers=self._configuration.limits.max_health_watchers, + watch_queue_capacity=( + self._configuration.limits.health_watch_queue_capacity + ), + watch_delivery_timeout_seconds=( + self._configuration.health_watch_delivery_timeout_seconds + ), + ) + self._health_service = health_service startup_entered = False async def startup() -> None: nonlocal startup_entered startup_entered = True await self.on_startup() + await self.refresh_health() def procedures( identity: ProviderEndpointIdentity, @@ -133,18 +179,30 @@ def procedures( snapshot = declaration.snapshot(identity) return (manifest_registration(snapshot, sessions),) + def health_registration( + registrations: Sequence[ProcedureRegistration], + ) -> tuple[ProcedureRegistration, ...]: + return ( + health_service.registration( + registration.name for registration in registrations + ), + ) + try: await self._server.start( on_identity_allocated=startup, procedure_factory=procedures, + supplemental_procedure_factory=health_registration, ) await self._shutdown_requested.wait() finally: + health_service.begin_shutdown() shutdown_called = False async def shutdown_author_resources() -> None: nonlocal shutdown_called shutdown_called = True + health_service.close() if startup_entered: async with asyncio.timeout( self._configuration.shutdown_timeout_seconds @@ -166,6 +224,7 @@ async def shutdown_author_resources() -> None: except BaseException as error: shutdown_failure = error + self._health_service = None if endpoint_failure is not None: if shutdown_failure is not None: endpoint_failure.add_note( @@ -179,8 +238,28 @@ async def shutdown_author_resources() -> None: @final def request_shutdown(self) -> None: """Signal a running Provider to begin graceful shutdown.""" + health_service = self._health_service + if health_service is not None: + health_service.begin_shutdown() self._shutdown_requested.set() + async def _refresh_health( + self, + service: ProviderHealthService, + ) -> HealthStatus: + """Evaluate one hook invocation without leaking its failure.""" + try: + status = await self.health() + if not isinstance(status, HealthStatus): + raise TypeError("health() must return HealthStatus") + except Exception: + _LOGGER.exception( + "Provider health hook failed; reporting NOT_SERVING" + ) + status = HealthStatus.not_serving() + service.publish(status) + return status + def _install_signal_handlers(self) -> None: """Request graceful shutdown on SIGINT/SIGTERM where supported.""" loop = asyncio.get_running_loop() diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index f8df5db..316593c 100644 --- a/tests/test_ci_automation.py +++ b/tests/test_ci_automation.py @@ -7,7 +7,7 @@ ACTIONS = ROOT / ".github" / "actions" SHA_LENGTH = 40 SUPPORTED_PYTHON_COUNT = 3 -SCENARIO_COUNT = 6 +SCENARIO_COUNT = 7 DOWNSTREAM_BUNDLE_CONSUMER_COUNT = 3 DEPENDENCY_CHECKOUT_COUNT = 2 @@ -28,12 +28,14 @@ def test_action_library_has_focused_domain_boundaries() -> None: "evidence/collect", "package/build", "package/verify-installed", + "provider/assert-health", "provider/crash", "provider/create-fixture", "provider/start", "provider/stop", "scenarios/adversarial/invalid-peer", "scenarios/destructive/uninstall-reinstall", + "scenarios/health/standard-health", "scenarios/lifecycle/concurrent-providers", "scenarios/lifecycle/same-instance-replacement", "scenarios/resilience/crash-restart", @@ -230,6 +232,7 @@ def test_runtime_catalog_covers_required_behavior() -> None: "invalid-peer", "crash-restart", "uninstall-reinstall", + "standard-health", } for scenario in scenarios: assert scenario in workflow @@ -243,6 +246,38 @@ def test_runtime_catalog_covers_required_behavior() -> None: assert "/agent/test-image" in workflow +def test_health_scenario_uses_standard_wire_service_and_public_hook() -> None: + """Keep R2 coverage on the real endpoint and installed runtime bundle.""" + scenario = read( + ACTIONS / "scenarios" / "health" / "standard-health" / "action.yml" + ) + action = read(ACTIONS / "provider" / "assert-health" / "action.yml") + stop = read(ACTIONS / "provider" / "stop" / "action.yml") + verifier = read( + ACTIONS / "provider" / "assert-health" / "assert_health.py" + ) + fixture = read(ACTIONS / "provider" / "create-fixture" / "fixture.py") + workflow = read(WORKFLOWS / "provider-runtime-scenarios.yml") + + assert "./.github/actions/agent/install" in scenario + assert "./.github/actions/agent/start" in scenario + assert "./.github/actions/agent/assert-session" in scenario + assert "./.github/actions/provider/assert-health" in scenario + assert "grpc_health.v1" in verifier + assert "HealthStub" in verifier + assert "SERVICE_UNKNOWN" in verifier + assert "SIGUSR1" in verifier + assert '"stop"' in verifier + assert "_assert_shutdown_transition" in verifier + assert "HealthStatus" in fixture + assert "refresh_health" in fixture + assert "docker exec" in action + assert "/opt/stateforge-provider-tests/health" in action + assert 'state}" != "inactive"' in stop + assert 'state}" != "failed"' in stop + assert "health:" in workflow + + def test_runtime_uses_real_agent_instead_of_synthetic_sfp_client() -> None: """Exercise Provider behavior through the installed Agent daemon.""" install = read(ACTIONS / "agent" / "install" / "action.yml") diff --git a/tests/test_health_service.py b/tests/test_health_service.py new file mode 100644 index 0000000..a427ef2 --- /dev/null +++ b/tests/test_health_service.py @@ -0,0 +1,312 @@ +"""Behavioral tests for Provider health publication and gRPC adaptation.""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import shutil +import tempfile +from collections.abc import AsyncIterator, Iterator +from pathlib import Path + +import grpc +import pytest +from grpc_health.v1 import health_pb2, health_pb2_grpc + +from stateforge_provider import HealthStatus, Platform, Provider +from stateforge_provider._runtime.configuration import ( + ProviderRuntimeConfiguration, + ProviderRuntimeLimits, +) +from stateforge_provider._services.health.service import ProviderHealthService + +_BOOTSTRAP_SERVICE = "stateforge.sfp.bootstrap.v1.BootstrapService" +_MANIFEST_SERVICE = "stateforge.sfp.manifest.v1.ManifestService" + + +@pytest.fixture +def endpoint_directory() -> Iterator[Path]: + directory = Path(tempfile.mkdtemp(prefix="sfp-health-", dir="/tmp")) + try: + yield directory + finally: + shutil.rmtree(directory, ignore_errors=True) + + +class _HealthProvider(Provider): + identity = "stateforge.test_health" + display_name = "Health test Provider" + version = "1.0.0" + platforms = (Platform("linux", "x86_64"),) + + def __init__(self, directory: Path) -> None: + super().__init__( + discovery_directory=directory, + state_directory=directory / "state", + ) + self.status = HealthStatus.healthy() + self.fail_health = False + self.started = asyncio.Event() + + async def on_startup(self) -> None: + self.started.set() + + async def health(self) -> HealthStatus: + if self.fail_health: + raise RuntimeError("local sensitive probe detail") + return self.status + + +@contextlib.asynccontextmanager +async def _running( + provider: _HealthProvider, +) -> AsyncIterator[ + tuple[ + health_pb2_grpc.HealthStub, + grpc.aio.Channel, + asyncio.Task[None], + ] +]: + task = asyncio.create_task(provider.serve()) + await asyncio.wait_for(provider.started.wait(), timeout=2) + channel = grpc.aio.insecure_channel(f"unix:{provider._server.socket_path}") + try: + await asyncio.wait_for(channel.channel_ready(), timeout=2) + yield health_pb2_grpc.HealthStub(channel), channel, task + finally: + provider.request_shutdown() + await channel.close() + await asyncio.wait_for(task, timeout=2) + + +def _request(service: str = "") -> health_pb2.HealthCheckRequest: + return health_pb2.HealthCheckRequest(service=service) + + +@pytest.mark.asyncio +async def test_check_uses_standard_health_on_the_shared_endpoint( + endpoint_directory: Path, +) -> None: + provider = _HealthProvider(endpoint_directory) + + async with _running(provider) as (stub, _, _): + for service in ("", _BOOTSTRAP_SERVICE, _MANIFEST_SERVICE): + response = await stub.Check(_request(service), timeout=2) + assert response.status == health_pb2.HealthCheckResponse.SERVING + + for service in ("stateforge.sfp.unknown.v1.Service", "stateforge.*"): + with pytest.raises(grpc.aio.AioRpcError) as captured: + await stub.Check(_request(service), timeout=2) + assert captured.value.code() is grpc.StatusCode.NOT_FOUND + assert "sensitive" not in captured.value.details() + + +@pytest.mark.asyncio +async def test_watch_replays_and_preserves_ordered_author_transitions( + endpoint_directory: Path, +) -> None: + provider = _HealthProvider(endpoint_directory) + + async with _running(provider) as (stub, _, _): + watch = stub.Watch(_request()) + assert (await watch.read()).status == ( + health_pb2.HealthCheckResponse.SERVING + ) + + provider.status = HealthStatus.maintenance() + assert await provider.refresh_health() is HealthStatus.MAINTENANCE + assert (await watch.read()).status == ( + health_pb2.HealthCheckResponse.NOT_SERVING + ) + + provider.status = HealthStatus.healthy() + assert await provider.refresh_health() is HealthStatus.SERVING + assert (await watch.read()).status == ( + health_pb2.HealthCheckResponse.SERVING + ) + watch.cancel() + + +@pytest.mark.asyncio +async def test_unknown_watch_remains_open_until_service_registration( + endpoint_directory: Path, +) -> None: + provider = _HealthProvider(endpoint_directory) + + async with _running(provider) as (stub, _, _): + service_name = "stateforge.sfp.future.v1.FutureService" + watch = stub.Watch(_request(service_name)) + assert (await watch.read()).status == ( + health_pb2.HealthCheckResponse.SERVICE_UNKNOWN + ) + + assert provider._health_service is not None + provider._health_service.register_services((service_name,)) + assert (await watch.read()).status == ( + health_pb2.HealthCheckResponse.SERVING + ) + watch.cancel() + + +@pytest.mark.asyncio +async def test_hook_failure_is_local_and_fails_closed( + endpoint_directory: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + provider = _HealthProvider(endpoint_directory) + + async with _running(provider) as (stub, _, _): + watch = stub.Watch(_request()) + assert (await watch.read()).status == ( + health_pb2.HealthCheckResponse.SERVING + ) + provider.fail_health = True + with caplog.at_level(logging.ERROR): + status = await provider.refresh_health() + + assert status is HealthStatus.NOT_SERVING + assert (await watch.read()).status == ( + health_pb2.HealthCheckResponse.NOT_SERVING + ) + assert "local sensitive probe detail" in caplog.text + response = await stub.Check(_request(), timeout=2) + assert response.status == health_pb2.HealthCheckResponse.NOT_SERVING + watch.cancel() + + +@pytest.mark.asyncio +async def test_shutdown_publishes_not_serving_before_transport_closes( + endpoint_directory: Path, +) -> None: + provider = _HealthProvider(endpoint_directory) + provider._configuration = ProviderRuntimeConfiguration( + state_directory=endpoint_directory / "state", + discovery_directory=endpoint_directory, + shutdown_timeout_seconds=0.25, + ) + task = asyncio.create_task(provider.serve()) + await asyncio.wait_for(provider.started.wait(), timeout=2) + channel = grpc.aio.insecure_channel(f"unix:{provider._server.socket_path}") + await asyncio.wait_for(channel.channel_ready(), timeout=2) + watch = health_pb2_grpc.HealthStub(channel).Watch(_request()) + assert ( + await watch.read() + ).status == health_pb2.HealthCheckResponse.SERVING + + provider.request_shutdown() + assert (await watch.read()).status == ( + health_pb2.HealthCheckResponse.NOT_SERVING + ) + watch.cancel() + await channel.close() + await asyncio.wait_for(task, timeout=2) + + assert not provider._server.descriptor_path.exists() + assert not provider._server.socket_path.exists() + + +@pytest.mark.asyncio +async def test_restart_has_a_new_generation_and_fresh_serving_state( + endpoint_directory: Path, +) -> None: + first = _HealthProvider(endpoint_directory) + async with _running(first) as (stub, _, _): + first_generation = first._server.endpoint_generation + assert (await stub.Check(_request(), timeout=2)).status == ( + health_pb2.HealthCheckResponse.SERVING + ) + + second = _HealthProvider(endpoint_directory) + async with _running(second) as (stub, _, _): + assert second._server.endpoint_generation == first_generation + 1 + assert (await stub.Check(_request(), timeout=2)).status == ( + health_pb2.HealthCheckResponse.SERVING + ) + + +class _AbortError(Exception): + def __init__(self, code: grpc.StatusCode, details: str) -> None: + self.code = code + self.details = details + + +class _BlockingContext: + def __init__(self) -> None: + self.first_write = asyncio.Event() + self.release_writes = asyncio.Event() + self.responses: list[health_pb2.HealthCheckResponse] = [] + + async def write(self, response: health_pb2.HealthCheckResponse) -> None: + self.responses.append(response) + self.first_write.set() + await self.release_writes.wait() + + async def abort(self, code: grpc.StatusCode, details: str = "") -> None: + raise _AbortError(code, details) + + +@pytest.mark.asyncio +async def test_slow_watch_is_bounded_without_blocking_the_publisher() -> None: + service = ProviderHealthService( + max_watchers=1, + watch_queue_capacity=1, + watch_delivery_timeout_seconds=1, + ) + service.register_services((_BOOTSTRAP_SERVICE,)) + context = _BlockingContext() + watch = asyncio.create_task(service.Watch(_request(), context)) + await asyncio.wait_for(context.first_write.wait(), timeout=1) + + service.publish(HealthStatus.healthy()) + service.publish(HealthStatus.maintenance()) + service.publish(HealthStatus.healthy()) + context.release_writes.set() + + with pytest.raises(_AbortError) as captured: + await asyncio.wait_for(watch, timeout=1) + assert captured.value.code is grpc.StatusCode.RESOURCE_EXHAUSTED + assert service._observer_count == 0 + + +@pytest.mark.asyncio +async def test_watch_capacity_and_delivery_timeout_are_bounded() -> None: + service = ProviderHealthService( + max_watchers=1, + watch_queue_capacity=1, + watch_delivery_timeout_seconds=0.01, + ) + first_context = _BlockingContext() + first = asyncio.create_task(service.Watch(_request(), first_context)) + await asyncio.wait_for(first_context.first_write.wait(), timeout=1) + + with pytest.raises(_AbortError) as captured: + await service.Watch(_request(), _BlockingContext()) + assert captured.value.code is grpc.StatusCode.RESOURCE_EXHAUSTED + + with pytest.raises(_AbortError) as timed_out: + await first + assert timed_out.value.code is grpc.StatusCode.RESOURCE_EXHAUSTED + + +def test_health_status_and_runtime_limits_are_typed_and_validated( + endpoint_directory: Path, +) -> None: + assert HealthStatus.healthy().is_serving + assert not HealthStatus.degraded().is_serving + assert not HealthStatus.maintenance().is_serving + assert not HealthStatus.not_serving().is_serving + + for field in ("max_health_watchers", "health_watch_queue_capacity"): + with pytest.raises(ValueError, match="positive integer"): + ProviderRuntimeLimits(**{field: 0}) + + with pytest.raises( + ValueError, + match="health_watch_delivery_timeout_seconds must be positive", + ): + ProviderRuntimeConfiguration( + state_directory=endpoint_directory / "state", + discovery_directory=endpoint_directory, + health_watch_delivery_timeout_seconds=0, + )