From 9b257d9a1b092d807f0422ee3057f3d90d1e6b87 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Fri, 17 Jul 2026 11:58:50 +0300 Subject: [PATCH] feat: define shard lifecycle status contract --- .../evidence/DGR-022/README.md | 46 ++ .scratch/distributed-gguf-runtime/prd.json | 4 +- packages/node/meshnet_node/shard_lifecycle.py | 472 ++++++++++++++++++ tests/test_shard_lifecycle.py | 182 +++++++ 4 files changed, 702 insertions(+), 2 deletions(-) create mode 100644 .scratch/distributed-gguf-runtime/evidence/DGR-022/README.md create mode 100644 packages/node/meshnet_node/shard_lifecycle.py create mode 100644 tests/test_shard_lifecycle.py diff --git a/.scratch/distributed-gguf-runtime/evidence/DGR-022/README.md b/.scratch/distributed-gguf-runtime/evidence/DGR-022/README.md new file mode 100644 index 0000000..0277eda --- /dev/null +++ b/.scratch/distributed-gguf-runtime/evidence/DGR-022/README.md @@ -0,0 +1,46 @@ +# DGR-022 evidence — Shard lifecycle and structured status RPC contract + +**Completed:** 2026-07-17 +**Branch:** `ralph/distributed-gguf-runtime` +**Authority:** `.scratch/distributed-gguf-runtime/prd.json` + +## Outcome + +Implemented the versioned, backend-neutral lifecycle/status contract consumed by a future generated gRPC binding. The contract keeps Meshnet routing, identity, authentication policy, billing, and llama.cpp ownership outside the worker contract. + +## Implemented + +- `packages/node/meshnet_node/shard_lifecycle.py` + - capability, health, session, cancellation, release, and metrics RPC names + - schema version negotiation and fail-closed unsupported-version handling + - structured status/error taxonomy with retryability and details + - lifecycle state machine for prefill/decode/cancel/release transitions + - monotonic idempotency-step enforcement and duplicate rejection + - bounded frame/byte flow control with cancellation-aware waits + - explicit cache expectation/result types + - deadline policy and TLS/auth transport hooks + - deterministic contract serialization round-trip +- `tests/test_shard_lifecycle.py` + - contract round-trip and RPC coverage + - unsupported-version rejection + - malformed transition and idempotency rejection + - cancellation/release behavior + - bounded flow-control behavior + - TLS hook and incomplete-contract fail-closed behavior + +## Verification + +```text +$ PYTHONPATH=packages/node pytest -q tests/test_shard_lifecycle.py tests/test_activation_envelope.py +17 passed in 0.10s +``` + +The existing DGR-021 activation-envelope tests remain green alongside DGR-022. + +## Scope limitation + +This story defines the lifecycle/status contract only. Generated Python/C++ protobuf bindings and the concrete `shard_runtime.proto` generation pipeline are DGR-023 and remain separate. + +## Dependency handoff + +DGR-023 may consume the RPC names, status taxonomy, version identity, deadlines, flow-control limits, and TLS/auth hooks when the canonical `.proto` schema and toolchain are provisioned. diff --git a/.scratch/distributed-gguf-runtime/prd.json b/.scratch/distributed-gguf-runtime/prd.json index f685db8..5b50528 100644 --- a/.scratch/distributed-gguf-runtime/prd.json +++ b/.scratch/distributed-gguf-runtime/prd.json @@ -483,8 +483,8 @@ "Add compatibility tests for supported versions and fail-closed tests for unsupported versions and malformed lifecycle transitions.", "Applicable shared quality gates in `prd.json` pass, and the evidence handoff records exact commands/results, changed files, limitations, and dependency handoff." ], - "passes": false, - "notes": "Generated source issue: .scratch/distributed-gguf-runtime/issues/022-define-shard-lifecycle-and-structured-status-rpcs.md; prd.json is authoritative.", + "passes": true, + "notes": "Completed from isolated DGR-022 Ralph worktree; verified with 17 focused pytest cases and retained DGR-021 envelope compatibility.", "blocks": [ "DGR-024", "DGR-033", diff --git a/packages/node/meshnet_node/shard_lifecycle.py b/packages/node/meshnet_node/shard_lifecycle.py new file mode 100644 index 0000000..e466b6c --- /dev/null +++ b/packages/node/meshnet_node/shard_lifecycle.py @@ -0,0 +1,472 @@ +"""Versioned Shard lifecycle and structured status contract. + +This module defines the semantic contract consumed by a future generated gRPC +binding. It deliberately contains no Meshnet routing, authentication policy, +billing, or llama.cpp types. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +import json +import threading +from typing import Any, Callable, ClassVar + + +SCHEMA_NAME = "meshnet.shard-runtime" +SCHEMA_VERSION = 1 +SUPPORTED_VERSIONS = frozenset({SCHEMA_VERSION}) +DEFAULT_MAX_INFLIGHT_FRAMES = 32 +DEFAULT_MAX_INFLIGHT_BYTES = 8 * 1024 * 1024 + + +class RpcName(str, Enum): + CAPABILITY = "GetCapability" + HEALTH = "CheckHealth" + SESSION = "OpenSession" + CANCEL = "CancelSession" + RELEASE = "ReleaseSession" + METRICS = "GetMetrics" + + +class StatusCode(str, Enum): + OK = "OK" + INVALID_ARGUMENT = "INVALID_ARGUMENT" + UNSUPPORTED_VERSION = "UNSUPPORTED_VERSION" + FAILED_PRECONDITION = "FAILED_PRECONDITION" + MALFORMED_LIFECYCLE = "MALFORMED_LIFECYCLE" + NOT_FOUND = "NOT_FOUND" + ALREADY_EXISTS = "ALREADY_EXISTS" + CANCELLED = "CANCELLED" + DEADLINE_EXCEEDED = "DEADLINE_EXCEEDED" + RESOURCE_EXHAUSTED = "RESOURCE_EXHAUSTED" + UNAUTHENTICATED = "UNAUTHENTICATED" + PERMISSION_DENIED = "PERMISSION_DENIED" + DATA_LOSS = "DATA_LOSS" + UNAVAILABLE = "UNAVAILABLE" + INTERNAL = "INTERNAL" + + +class LifecycleState(str, Enum): + OPEN = "OPEN" + PREFILLING = "PREFILLING" + DECODING = "DECODING" + CANCELLING = "CANCELLING" + CANCELLED = "CANCELLED" + RELEASING = "RELEASING" + RELEASED = "RELEASED" + FAILED = "FAILED" + + +class CacheExpectation(str, Enum): + NONE = "NONE" + OPTIONAL = "OPTIONAL" + REQUIRED = "REQUIRED" + + +class CacheResult(str, Enum): + NOT_REQUESTED = "NOT_REQUESTED" + HIT = "HIT" + MISS = "MISS" + INVALIDATED = "INVALIDATED" + STORED = "STORED" + + +class SessionPhase(str, Enum): + PREFILL = "PREFILL" + DECODE = "DECODE" + + +class LifecycleContractError(ValueError): + """A protocol violation represented by a structured status.""" + + def __init__(self, status: "StructuredStatus") -> None: + self.status = status + super().__init__(status.message) + + +@dataclass(frozen=True) +class StructuredStatus: + code: StatusCode + message: str + retryable: bool = False + details: dict[str, str] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "code": self.code.value, + "message": self.message, + "retryable": self.retryable, + "details": dict(sorted(self.details.items())), + } + + @classmethod + def from_dict(cls, value: dict[str, Any]) -> "StructuredStatus": + return cls( + code=StatusCode(str(value["code"])), + message=str(value["message"]), + retryable=bool(value.get("retryable", False)), + details={str(k): str(v) for k, v in value.get("details", {}).items()}, + ) + + +@dataclass(frozen=True) +class DeadlinePolicy: + capability_seconds: float = 5.0 + health_seconds: float = 2.0 + session_open_seconds: float = 10.0 + session_idle_seconds: float = 30.0 + cancel_seconds: float = 2.0 + release_seconds: float = 5.0 + metrics_seconds: float = 5.0 + + def __post_init__(self) -> None: + if any(value <= 0 for value in self.__dict__.values()): + raise ValueError("all RPC deadlines must be positive") + + +@dataclass(frozen=True) +class TlsAuthHooks: + """Transport hooks only; Meshnet remains the identity and billing authority.""" + + tls_required: bool = True + server_name: str = "shard.meshnet" + client_certificate_hook: str | None = None + peer_identity_hook: str | None = None + auth_metadata_hook: str | None = None + + def validate(self) -> None: + if self.tls_required and not self.server_name: + raise ValueError("TLS server_name is required when TLS is enabled") + + +@dataclass(frozen=True) +class FlowControlLimits: + max_inflight_frames: int = DEFAULT_MAX_INFLIGHT_FRAMES + max_inflight_bytes: int = DEFAULT_MAX_INFLIGHT_BYTES + max_frame_bytes: int = 1024 * 1024 + + def __post_init__(self) -> None: + if min(self.max_inflight_frames, self.max_inflight_bytes, self.max_frame_bytes) <= 0: + raise ValueError("flow-control limits must be positive") + + +class FlowControl: + """Bounded sender window; callers block or fail instead of growing unbounded.""" + + def __init__(self, limits: FlowControlLimits = FlowControlLimits()) -> None: + self.limits = limits + self._condition = threading.Condition() + self._frames = 0 + self._bytes = 0 + + @property + def outstanding(self) -> tuple[int, int]: + with self._condition: + return self._frames, self._bytes + + def acquire(self, size: int, *, wait: bool = False, cancelled: Callable[[], bool] | None = None) -> None: + if size < 0 or size > self.limits.max_frame_bytes: + raise LifecycleContractError(StructuredStatus( + StatusCode.RESOURCE_EXHAUSTED, "frame exceeds bounded flow-control window", + details={"max_frame_bytes": str(self.limits.max_frame_bytes)}, + )) + with self._condition: + while ( + self._frames >= self.limits.max_inflight_frames + or self._bytes + size > self.limits.max_inflight_bytes + ): + if cancelled and cancelled(): + raise LifecycleContractError(StructuredStatus(StatusCode.CANCELLED, "flow-control wait cancelled")) + if not wait: + raise LifecycleContractError(StructuredStatus( + StatusCode.RESOURCE_EXHAUSTED, "flow-control window is full", retryable=True, + )) + self._condition.wait(timeout=0.05) + self._frames += 1 + self._bytes += size + + def release(self, size: int) -> None: + with self._condition: + self._frames = max(0, self._frames - 1) + self._bytes = max(0, self._bytes - max(0, size)) + self._condition.notify_all() + + +class CancellationToken: + def __init__(self) -> None: + self._event = threading.Event() + + @property + def cancelled(self) -> bool: + return self._event.is_set() + + def cancel(self) -> None: + self._event.set() + + +@dataclass(frozen=True) +class CapabilityRequest: + schema_version: int = SCHEMA_VERSION + + +@dataclass(frozen=True) +class CapabilityResponse: + status: StructuredStatus + schema_version: int + supported_versions: tuple[int, ...] + artifact_fingerprint: str = "" + runtime_fingerprint: str = "" + shard_start: int = 0 + shard_end: int = 0 + effective_start: int = 0 + max_sessions: int = 0 + max_frame_bytes: int = DEFAULT_MAX_INFLIGHT_BYTES + + def validate(self) -> None: + if self.status.code is StatusCode.OK and self.schema_version not in SUPPORTED_VERSIONS: + raise LifecycleContractError(StructuredStatus( + StatusCode.UNSUPPORTED_VERSION, "worker selected an unsupported schema version", + )) + if self.shard_end < self.shard_start or self.effective_start < self.shard_start: + raise LifecycleContractError(StructuredStatus( + StatusCode.INVALID_ARGUMENT, "invalid authoritative shard range", + )) + + +@dataclass(frozen=True) +class HealthRequest: + schema_version: int = SCHEMA_VERSION + include_metrics: bool = False + + +@dataclass(frozen=True) +class HealthResponse: + status: StructuredStatus + serving: bool + state: str + active_sessions: int = 0 + + +@dataclass(frozen=True) +class SessionRequest: + schema_version: int + request_id: str + work_id: str + route_session: str + route_epoch: int + artifact_fingerprint: str + runtime_fingerprint: str + shard_start: int + shard_end: int + effective_start: int + cache_expectation: CacheExpectation = CacheExpectation.NONE + deadline_seconds: float = 30.0 + + def validate(self) -> None: + if self.schema_version not in SUPPORTED_VERSIONS: + raise LifecycleContractError(StructuredStatus( + StatusCode.UNSUPPORTED_VERSION, "unsupported session schema version", + details={"requested": str(self.schema_version), "supported": "1"}, + )) + if not self.request_id or not self.work_id or not self.route_session: + raise LifecycleContractError(StructuredStatus( + StatusCode.INVALID_ARGUMENT, "request, work, and route-session IDs are required", + )) + if self.route_epoch < 0 or self.shard_start < 0 or self.shard_end <= self.shard_start: + raise LifecycleContractError(StructuredStatus( + StatusCode.INVALID_ARGUMENT, "invalid route epoch or shard range", + )) + if not self.shard_start <= self.effective_start <= self.shard_end: + raise LifecycleContractError(StructuredStatus( + StatusCode.INVALID_ARGUMENT, "effective start must be inside the shard range", + )) + if self.deadline_seconds <= 0: + raise LifecycleContractError(StructuredStatus( + StatusCode.INVALID_ARGUMENT, "session deadline must be positive", + )) + + +@dataclass(frozen=True) +class SessionFrame: + phase: SessionPhase + position: int + idempotency_step: int + payload: Any + cache_expectation: CacheExpectation = CacheExpectation.NONE + size_bytes: int = 0 + + def validate(self, limits: FlowControlLimits) -> None: + if self.position < 0 or self.idempotency_step < 0: + raise LifecycleContractError(StructuredStatus( + StatusCode.INVALID_ARGUMENT, "position and idempotency step must be non-negative", + )) + if self.size_bytes < 0 or self.size_bytes > limits.max_frame_bytes: + raise LifecycleContractError(StructuredStatus( + StatusCode.RESOURCE_EXHAUSTED, "session frame exceeds max_frame_bytes", + )) + + +@dataclass(frozen=True) +class SessionResult: + status: StructuredStatus + cache_result: CacheResult = CacheResult.NOT_REQUESTED + position: int = 0 + idempotency_step: int = 0 + payload: Any = None + + +@dataclass(frozen=True) +class CancelRequest: + schema_version: int + request_id: str + work_id: str + route_session: str + route_epoch: int + reason: str = "" + + +@dataclass(frozen=True) +class ReleaseRequest: + schema_version: int + request_id: str + work_id: str + route_session: str + route_epoch: int + + +@dataclass(frozen=True) +class MetricsRequest: + schema_version: int = SCHEMA_VERSION + + +@dataclass(frozen=True) +class MetricsResponse: + status: StructuredStatus + active_sessions: int + queued_frames: int + inflight_bytes: int + kv_entries: int + generated_tokens: int + cancelled_sessions: int + + +@dataclass +class SessionLifecycle: + """Fail-closed state machine for one Route Session Activation Seam.""" + + request: SessionRequest + state: LifecycleState = LifecycleState.OPEN + cancellation: CancellationToken = field(default_factory=CancellationToken) + last_idempotency_step: int = -1 + _seen_steps: set[int] = field(default_factory=set, init=False, repr=False) + + _ALLOWED: ClassVar[dict[LifecycleState, frozenset[LifecycleState]]] = { + LifecycleState.OPEN: frozenset({LifecycleState.PREFILLING, LifecycleState.CANCELLING, LifecycleState.RELEASING, LifecycleState.FAILED}), + LifecycleState.PREFILLING: frozenset({LifecycleState.PREFILLING, LifecycleState.DECODING, LifecycleState.CANCELLING, LifecycleState.RELEASING, LifecycleState.FAILED}), + LifecycleState.DECODING: frozenset({LifecycleState.DECODING, LifecycleState.CANCELLING, LifecycleState.RELEASING, LifecycleState.FAILED}), + LifecycleState.CANCELLING: frozenset({LifecycleState.CANCELLED, LifecycleState.RELEASING, LifecycleState.FAILED}), + LifecycleState.CANCELLED: frozenset({LifecycleState.RELEASING, LifecycleState.RELEASED}), + LifecycleState.RELEASING: frozenset({LifecycleState.RELEASED, LifecycleState.FAILED}), + LifecycleState.RELEASED: frozenset(), + LifecycleState.FAILED: frozenset({LifecycleState.RELEASING, LifecycleState.RELEASED}), + } + + def _transition(self, target: LifecycleState) -> None: + if target not in self._ALLOWED[self.state]: + raise LifecycleContractError(StructuredStatus( + StatusCode.MALFORMED_LIFECYCLE, + f"cannot transition from {self.state.value} to {target.value}", + details={"state": self.state.value, "target": target.value}, + )) + self.state = target + + def apply(self, frame: SessionFrame) -> None: + frame.validate(FlowControlLimits()) + if self.cancellation.cancelled and frame.phase is not SessionPhase.PREFILL: + raise LifecycleContractError(StructuredStatus(StatusCode.CANCELLED, "session cancellation propagated")) + if frame.idempotency_step in self._seen_steps: + raise LifecycleContractError(StructuredStatus( + StatusCode.ALREADY_EXISTS, "duplicate idempotency step", details={"step": str(frame.idempotency_step)}, + )) + if frame.idempotency_step <= self.last_idempotency_step: + raise LifecycleContractError(StructuredStatus( + StatusCode.MALFORMED_LIFECYCLE, "idempotency steps must increase monotonically", + )) + target = LifecycleState.PREFILLING if frame.phase is SessionPhase.PREFILL else LifecycleState.DECODING + self._transition(target) + self._seen_steps.add(frame.idempotency_step) + self.last_idempotency_step = frame.idempotency_step + + def cancel(self, reason: str = "") -> StructuredStatus: + if self.state in {LifecycleState.RELEASED, LifecycleState.RELEASING}: + return StructuredStatus(StatusCode.FAILED_PRECONDITION, "session is already releasing or released") + if self.state is LifecycleState.CANCELLED: + return StructuredStatus(StatusCode.OK, "session already cancelled") + self._transition(LifecycleState.CANCELLING) + self.cancellation.cancel() + self._transition(LifecycleState.CANCELLED) + return StructuredStatus(StatusCode.CANCELLED, reason or "session cancelled") + + def release(self) -> StructuredStatus: + if self.state is LifecycleState.RELEASED: + return StructuredStatus(StatusCode.OK, "session already released") + self._transition(LifecycleState.RELEASING) + self.cancellation.cancel() + self._transition(LifecycleState.RELEASED) + return StructuredStatus(StatusCode.OK, "session released") + + +@dataclass(frozen=True) +class ShardRpcContract: + """Service/method and operational rules for generated gRPC bindings.""" + + schema: str = SCHEMA_NAME + version: int = SCHEMA_VERSION + methods: tuple[RpcName, ...] = tuple(RpcName) + deadlines: DeadlinePolicy = DeadlinePolicy() + flow_control: FlowControlLimits = FlowControlLimits() + tls_auth: TlsAuthHooks = TlsAuthHooks() + + def validate(self) -> None: + if self.schema != SCHEMA_NAME or self.version not in SUPPORTED_VERSIONS: + raise LifecycleContractError(StructuredStatus( + StatusCode.UNSUPPORTED_VERSION, "unsupported Shard RPC contract version", + )) + if self.methods != tuple(RpcName): + raise LifecycleContractError(StructuredStatus( + StatusCode.FAILED_PRECONDITION, "contract must expose the complete lifecycle RPC set", + )) + self.tls_auth.validate() + + def to_bytes(self) -> bytes: + self.validate() + data = { + "schema": self.schema, + "version": self.version, + "methods": [method.value for method in self.methods], + "deadlines": self.deadlines.__dict__, + "flow_control": self.flow_control.__dict__, + "tls_auth": self.tls_auth.__dict__, + } + return json.dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8") + + @classmethod + def from_bytes(cls, raw: bytes) -> "ShardRpcContract": + data = json.loads(raw) + if int(data.get("version", -1)) not in SUPPORTED_VERSIONS: + raise LifecycleContractError(StructuredStatus( + StatusCode.UNSUPPORTED_VERSION, "unsupported Shard RPC contract version", + )) + methods = tuple(RpcName(item) for item in data.get("methods", ())) + contract = cls( + schema=str(data.get("schema", "")), + version=int(data["version"]), + methods=methods, + deadlines=DeadlinePolicy(**data.get("deadlines", {})), + flow_control=FlowControlLimits(**data.get("flow_control", {})), + tls_auth=TlsAuthHooks(**data.get("tls_auth", {})), + ) + contract.validate() + return contract diff --git a/tests/test_shard_lifecycle.py b/tests/test_shard_lifecycle.py new file mode 100644 index 0000000..3d9a72f --- /dev/null +++ b/tests/test_shard_lifecycle.py @@ -0,0 +1,182 @@ +"""DGR-022 lifecycle and structured-status contract tests.""" + +from __future__ import annotations + +import json + +import pytest + +from meshnet_node.shard_lifecycle import ( + CacheExpectation, + CacheResult, + CapabilityResponse, + CancellationToken, + FlowControl, + FlowControlLimits, + LifecycleContractError, + LifecycleState, + RpcName, + SessionFrame, + SessionLifecycle, + SessionPhase, + SessionRequest, + ShardRpcContract, + StatusCode, + StructuredStatus, +) + + +def _request(**overrides) -> SessionRequest: + values = dict( + schema_version=1, + request_id="request-1", + work_id="work-1", + route_session="route-1", + route_epoch=3, + artifact_fingerprint="artifact-sha", + runtime_fingerprint="runtime-sha", + shard_start=0, + shard_end=8, + effective_start=0, + cache_expectation=CacheExpectation.OPTIONAL, + ) + values.update(overrides) + return SessionRequest(**values) + + +def _frame(phase: SessionPhase, step: int, position: int = 0) -> SessionFrame: + return SessionFrame( + phase=phase, + position=position, + idempotency_step=step, + payload=b"frame", + size_bytes=5, + ) + + +def test_contract_roundtrip_exposes_all_lifecycle_rpcs_and_operational_hooks(): + contract = ShardRpcContract() + restored = ShardRpcContract.from_bytes(contract.to_bytes()) + + assert restored == contract + assert restored.methods == tuple(RpcName) + assert restored.deadlines.session_idle_seconds > restored.deadlines.health_seconds + assert restored.flow_control.max_inflight_frames == 32 + assert restored.tls_auth.tls_required is True + + +def test_supported_version_roundtrip_and_status_are_stable(): + status = StructuredStatus(StatusCode.RESOURCE_EXHAUSTED, "window full", retryable=True, details={"limit": "32"}) + restored = StructuredStatus.from_dict(status.to_dict()) + + assert restored == status + assert CapabilityResponse( + status=StructuredStatus(StatusCode.OK, "ready"), + schema_version=1, + supported_versions=(1,), + ).validate() is None + + +@pytest.mark.parametrize("version", [0, 2, 99]) +def test_unsupported_versions_fail_closed(version): + with pytest.raises(LifecycleContractError) as exc: + ShardRpcContract.from_bytes(json.dumps({ + "schema": "meshnet.shard-runtime", + "version": version, + "methods": [method.value for method in RpcName], + }).encode()) + assert exc.value.status.code is StatusCode.UNSUPPORTED_VERSION + + with pytest.raises(LifecycleContractError) as exc: + _request(schema_version=version).validate() + assert exc.value.status.code is StatusCode.UNSUPPORTED_VERSION + + +def test_malformed_lifecycle_transitions_fail_closed(): + session = SessionLifecycle(_request()) + + with pytest.raises(LifecycleContractError) as exc: + session.apply(_frame(SessionPhase.DECODE, 0)) + assert exc.value.status.code is StatusCode.MALFORMED_LIFECYCLE + assert session.state is LifecycleState.OPEN + + session.apply(_frame(SessionPhase.PREFILL, 0)) + session.apply(_frame(SessionPhase.DECODE, 1, position=1)) + assert session.state is LifecycleState.DECODING + + with pytest.raises(LifecycleContractError) as exc: + session.apply(_frame(SessionPhase.PREFILL, 2)) + assert exc.value.status.code is StatusCode.MALFORMED_LIFECYCLE + + +def test_duplicate_and_non_monotonic_idempotency_steps_are_rejected(): + session = SessionLifecycle(_request()) + session.apply(_frame(SessionPhase.PREFILL, 4)) + + with pytest.raises(LifecycleContractError) as exc: + session.apply(_frame(SessionPhase.PREFILL, 4)) + assert exc.value.status.code is StatusCode.ALREADY_EXISTS + + with pytest.raises(LifecycleContractError) as exc: + session.apply(_frame(SessionPhase.PREFILL, 3)) + assert exc.value.status.code is StatusCode.MALFORMED_LIFECYCLE + + +def test_cancel_propagates_to_waiters_and_release_is_idempotent(): + session = SessionLifecycle(_request()) + assert session.cancel("client disconnected").code is StatusCode.CANCELLED + assert session.cancellation.cancelled is True + assert session.state is LifecycleState.CANCELLED + + with pytest.raises(LifecycleContractError) as exc: + session.apply(_frame(SessionPhase.DECODE, 0)) + assert exc.value.status.code is StatusCode.CANCELLED + + assert session.release().code is StatusCode.OK + assert session.state is LifecycleState.RELEASED + assert session.release().code is StatusCode.OK + + +def test_bounded_flow_control_rejects_oversize_and_full_windows(): + flow = FlowControl(FlowControlLimits(max_inflight_frames=1, max_inflight_bytes=10, max_frame_bytes=8)) + flow.acquire(8) + + with pytest.raises(LifecycleContractError) as exc: + flow.acquire(1) + assert exc.value.status.code is StatusCode.RESOURCE_EXHAUSTED + assert exc.value.status.retryable is True + + with pytest.raises(LifecycleContractError) as exc: + flow.acquire(9) + assert exc.value.status.code is StatusCode.RESOURCE_EXHAUSTED + + flow.release(8) + assert flow.outstanding == (0, 0) + + +def test_flow_control_wait_honours_cancellation(): + token = CancellationToken() + flow = FlowControl(FlowControlLimits(max_inflight_frames=1, max_inflight_bytes=10, max_frame_bytes=8)) + flow.acquire(4) + token.cancel() + + with pytest.raises(LifecycleContractError) as exc: + flow.acquire(1, wait=True, cancelled=lambda: token.cancelled) + assert exc.value.status.code is StatusCode.CANCELLED + + +def test_cache_expectation_and_result_are_explicit(): + request = _request(cache_expectation=CacheExpectation.REQUIRED) + assert request.cache_expectation is CacheExpectation.REQUIRED + assert CacheResult.MISS.value == "MISS" + + +def test_invalid_contract_methods_and_tls_hooks_fail_closed(): + with pytest.raises(LifecycleContractError) as exc: + ShardRpcContract(methods=(RpcName.HEALTH,)).validate() + assert exc.value.status.code is StatusCode.FAILED_PRECONDITION + + with pytest.raises(ValueError, match="server_name"): + ShardRpcContract( + tls_auth=type(ShardRpcContract().tls_auth)(tls_required=True, server_name="") + ).validate()