diff --git a/projects/egress-gate/README.md b/projects/egress-gate/README.md index 84f37e35..6853d668 100644 --- a/projects/egress-gate/README.md +++ b/projects/egress-gate/README.md @@ -92,12 +92,29 @@ written by a harness to disk. from egress_gate.gates import create_builtin_registry from egress_gate.service import EgressGateServer -server = EgressGateServer(create_builtin_registry()) +server = EgressGateServer( + create_builtin_registry(), + timeout_middleware_processing=10, +) server.serve_sync("127.0.0.1:50051") ``` -The service creates one `Timeout` per evaluation and passes that deadline -through slot acquisition, policy preparation, and `RequestProcessor.process`. +In this example, `timeout_middleware_processing` gives each evaluation 10 +seconds. Omitting it uses the one-second service default. The value is expressed +in seconds, must be at least 10 milliseconds, and must resolve to whole +milliseconds. The service passes one resulting `Timeout` through slot +acquisition, policy preparation, and `RequestProcessor.process`. + +`Describe` leaves the optional binding RPC timeout empty, so OpenShell applies +the timeout configured on the gateway registration to the complete RPC. The +registration CLI defaults to 30 seconds and accepts `--timeout` to write a +different value. +The helper remembers the gateway file and registration name. On later CLI +starts, `serve` reads the current gateway timeout and requires the processing +timeout to be lower. If no registration has been added with the CLI, `serve` +starts without this check. If the gateway timeout expires, OpenShell applies +the policy's `on_error` behavior; use `on_error: fail_closed` when middleware +timeout failures must deny. ## Documentation and examples diff --git a/projects/egress-gate/docs/architecture/request-lifecycle.md b/projects/egress-gate/docs/architecture/request-lifecycle.md index 47e42d56..ee876949 100644 --- a/projects/egress-gate/docs/architecture/request-lifecycle.md +++ b/projects/egress-gate/docs/architecture/request-lifecycle.md @@ -55,6 +55,12 @@ encoded output limits return an atomic deny with source `runtime_limit` and `egress_gate_limit_exceeded`. No partial mutations or findings are returned. Gate contract and execution failures remain gRPC failures. +The internal processing timeout can return this denial only while the RPC is +still active. The OpenShell gateway owns a separate outer RPC ceiling. If that +outer clock expires first, OpenShell applies the middleware entry's `on_error` +policy instead of receiving an Egress Gate result. Use `on_error: fail_closed` +when middleware timeout failures must deny the request. + ## 5. Serialize the result The Egress Gate service adapter maps the protobuf-free `EgressResult` to diff --git a/projects/egress-gate/docs/architecture/service-boundary.md b/projects/egress-gate/docs/architecture/service-boundary.md index 00f3ed4c..686cd493 100644 --- a/projects/egress-gate/docs/architecture/service-boundary.md +++ b/projects/egress-gate/docs/architecture/service-boundary.md @@ -19,7 +19,7 @@ supervisor applies allowed mutations to the intercepted request. | RPC | Behavior | | --- | --- | -| `Describe` | Advertise Egress Gate and the pre-credentials HTTP binding | +| `Describe` | Return Egress Gate's pre-credentials HTTP binding | | `ValidateConfig` | Validate a complete registry-backed pipeline without publishing it | | `EvaluateHttpRequest` | Adapt one request, prepare/reuse policy, execute, and serialize | @@ -29,10 +29,23 @@ encoded configuration before registry parsing. ## Shared deadline and workers -`EvaluateHttpRequest` creates one monotonic `Timeout`. That same deadline is -used for semaphore acquisition, policy preparation, replacement-lock waits, -gate execution, and final result checks. `RequestProcessor.process` accepts the -caller-owned timeout and never creates or stores one. +`EvaluateHttpRequest` converts `timeout_middleware_processing` into one monotonic +`Timeout` used for semaphore acquisition, policy preparation, replacement-lock +waits, gate execution, and final result checks. `Describe` leaves the binding's +optional RPC timeout empty. OpenShell therefore applies the separately +configured gateway registration timeout to the complete RPC. + +The middleware protocol does not report the resolved gateway timeout back to +Egress Gate, and OpenShell does not propagate it as a gRPC deadline. For the +normal CLI-managed path, `add-gateway-registration` remembers the gateway TOML +path and registration name. `serve` reads the current timeout from that entry +at startup and requires it to be greater than +`timeout_middleware_processing`. Direct Python API use and manually managed +registrations do not have this startup check. If the gateway timeout expires +first, OpenShell applies the policy's `on_error` behavior. + +`RequestProcessor.process` accepts the caller-owned timeout and never creates or +stores one. Synchronous work runs in a bounded four-slot executor. The gRPC server permits sixteen concurrent RPCs. Cancellation does not stop Python code that already diff --git a/projects/egress-gate/docs/assets/diagrams/component-architecture.svg b/projects/egress-gate/docs/assets/diagrams/component-architecture.svg index 0f5db97c..dc7ca530 100644 --- a/projects/egress-gate/docs/assets/diagrams/component-architecture.svg +++ b/projects/egress-gate/docs/assets/diagrams/component-architecture.svg @@ -57,7 +57,7 @@ PIPELINE PROCESSOR RequestProcessor - gate order · one timeout · aggregation · allow or deny + gate order · one deadline · aggregation · allow or deny REQUEST GATES diff --git a/projects/egress-gate/docs/assets/diagrams/processing-pipeline.svg b/projects/egress-gate/docs/assets/diagrams/processing-pipeline.svg index f1f26e99..901f114c 100644 --- a/projects/egress-gate/docs/assets/diagrams/processing-pipeline.svg +++ b/projects/egress-gate/docs/assets/diagrams/processing-pipeline.svg @@ -44,7 +44,7 @@ 2 Pipeline processor - One shared timeout + bounds + One shared deadline + bounds diff --git a/projects/egress-gate/docs/evaluation.md b/projects/egress-gate/docs/evaluation.md index a6650a28..bd678a61 100644 --- a/projects/egress-gate/docs/evaluation.md +++ b/projects/egress-gate/docs/evaluation.md @@ -36,7 +36,7 @@ The repository includes a regex policy and two request cases. Run them from uv run egress-gate evaluate \ --policy examples/regex-redaction/egress-gate-config.yaml \ --cases examples/regex-redaction/cases.yaml \ - --timeout-seconds 1 + --timeout 1s ``` The command prepares the policy once, runs each case with a fresh timeout, and diff --git a/projects/egress-gate/docs/operations.md b/projects/egress-gate/docs/operations.md index 4ff39f8c..82d9fbd8 100644 --- a/projects/egress-gate/docs/operations.md +++ b/projects/egress-gate/docs/operations.md @@ -13,7 +13,7 @@ environment as needed. ```bash title="Start Egress Gate" uv run egress-gate gates list uv run egress-gate gates schema -uv run egress-gate serve --listen 0.0.0.0:50051 --timeout-seconds 4 +uv run egress-gate serve --listen 0.0.0.0:50051 --timeout 4s ``` Use a reachable non-loopback address only when the supervisor is outside the @@ -28,15 +28,30 @@ registrations. ```bash title="Register Egress Gate" uv run egress-gate add-gateway-registration \ - --host-ip YOUR_HOST_IPV4 --name egress-gate --port 50051 + --host-ip YOUR_HOST_IPV4 --name egress-gate --port 50051 --timeout 30s ``` The command updates `OPENSHELL_GATEWAY_CONFIG`, then `$XDG_CONFIG_HOME/openshell/gateway.toml`, then `~/.config/openshell/gateway.toml`. Use `--config PATH` for another file. +It remembers the absolute gateway file path and registration name in +`$XDG_CONFIG_HOME/openshell-egress-gate/registration.toml`, or under +`~/.config` when `XDG_CONFIG_HOME` is unset. Start the gateways again with the same commands or service managers that you normally use. +The optional registration `--timeout` sets the gateway RPC timeout written to +the TOML file and defaults to 30 seconds. It accepts whole seconds or +milliseconds, such as `45s` or `500ms`, and must be greater than 10ms so the +internal processing budget can remain lower. Rerunning the command writes the +value passed on that invocation. Set Egress Gate's internal processing budget with +`egress-gate serve --timeout DURATION`; the Python API calls that setting +`timeout_middleware_processing`. It must be at least 10ms and resolve to whole +milliseconds. When a remembered registration exists, `serve` reads its current +gateway timeout and refuses to start unless the processing timeout is lower. A +manually managed setup with no remembered registration starts without this +check. + To remove a registration, stop any running gateways that use the configuration again. List the available names with: @@ -54,9 +69,10 @@ uv run egress-gate remove-gateway-registration --name egress-gate Start the gateways again after the command completes. -The generated OpenShell middleware timeout is five seconds. Keep the Egress -Gate `--timeout-seconds` below it so queueing, preparation, and transport have -headroom. +`serve --timeout` covers queueing, policy preparation, and every configured +gate. If the gateway timeout expires first despite the startup check, OpenShell +applies the policy's `on_error` setting: `fail_closed` denies the request, while +`fail_open` allows it to continue. If the middleware RPC returns gRPC `RESOURCE_EXHAUSTED`, capacity may remain accounted for briefly while completed RPCs are torn down. The OpenShell gateway diff --git a/projects/egress-gate/docs/reference/limits-and-failures.md b/projects/egress-gate/docs/reference/limits-and-failures.md index 2bec400a..ea5cdc0a 100644 --- a/projects/egress-gate/docs/reference/limits-and-failures.md +++ b/projects/egress-gate/docs/reference/limits-and-failures.md @@ -6,9 +6,10 @@ agent_markdown: true # Limits and failure behavior -Limits are fail-closed and content-safe. The `service/` package checks exact -encoded protobuf sizes. Domain models check scalar, aggregate, and result -limits. +Egress Gate-owned limits are fail-closed and content-safe. The `service/` +package checks exact encoded protobuf sizes. Domain models check scalar, +aggregate, and result limits. OpenShell owns the separate outer RPC ceiling and +applies its configured `on_error` behavior when that ceiling expires first. | Area | Limit | | --- | ---: | @@ -20,7 +21,9 @@ limits. | Result metadata aggregate strings | 32 KiB | | Gate traces per result | 10 | | Header mutations per gate evaluation | 64 | -| Processing timeout | 30 seconds maximum | +| Offline `--timeout` | 10 milliseconds minimum; whole milliseconds | +| `timeout_middleware_processing` | 10 milliseconds minimum; whole milliseconds | +| Gateway registration timeout | Operator-configurable; helper default 30 seconds | | Concurrent processing slots | 4 | Request context and target aggregates, headers, replacement bodies, regex @@ -34,7 +37,7 @@ rejected value. | --- | --- | | Invalid phase, envelope, policy, or input encoding | gRPC `INVALID_ARGUMENT` | | Gate contract or unexpected execution failure | gRPC `INTERNAL` | -| Deadline or pipeline processor limit | deny, source `runtime_limit`, code `egress_gate_limit_exceeded` | +| Internal processing deadline or pipeline processor limit | deny, source `runtime_limit`, code `egress_gate_limit_exceeded` | | Gate terminal deny | deny, source `gate`, gate-owned reason code | | Pipeline default deny | deny, source `pipeline_default`, code `egress_gate_default_deny` | | Pipeline default allow | allow, source `pipeline_default`, no reason code | @@ -44,6 +47,10 @@ trace details. Failed policy preparation leaves the active policy unchanged. Stable error catalogs and reason codes never include request content or arbitrary exception text. +An internal processing timeout returns the runtime-limit denial only while the +RPC remains active. If the gateway's independent outer RPC ceiling expires +first, OpenShell applies the middleware entry's `on_error` policy. + ## Finding contract The released OpenShell wire contract has five fields. The pipeline processor's diff --git a/projects/egress-gate/examples/class-based-gate/README.md b/projects/egress-gate/examples/class-based-gate/README.md index fecbe9d5..48763139 100644 --- a/projects/egress-gate/examples/class-based-gate/README.md +++ b/projects/egress-gate/examples/class-based-gate/README.md @@ -39,7 +39,7 @@ Start Egress Gate with this example registry and content-safe debug diagnostics: uv run egress-gate \ --debug \ --registry examples.class-based-gate.keyword_gate:registry \ - serve --listen 0.0.0.0:50051 --timeout-seconds 4 + serve --listen 0.0.0.0:50051 --timeout 4s ``` Before you change the gateway configuration, stop any running OpenShell diff --git a/projects/egress-gate/examples/custom-gate/README.md b/projects/egress-gate/examples/custom-gate/README.md index 38238df1..2905f8c2 100644 --- a/projects/egress-gate/examples/custom-gate/README.md +++ b/projects/egress-gate/examples/custom-gate/README.md @@ -46,7 +46,7 @@ Start Egress Gate with this example registry and content-safe debug diagnostics: uv run egress-gate \ --debug \ --registry examples.custom-gate.keyword_gate:registry \ - serve --listen 0.0.0.0:50051 --timeout-seconds 4 + serve --listen 0.0.0.0:50051 --timeout 4s ``` Before you change the gateway configuration, stop any running OpenShell diff --git a/projects/egress-gate/examples/regex-redaction/README.md b/projects/egress-gate/examples/regex-redaction/README.md index 9190b796..7df9a3d1 100644 --- a/projects/egress-gate/examples/regex-redaction/README.md +++ b/projects/egress-gate/examples/regex-redaction/README.md @@ -26,7 +26,7 @@ working directory contains the pattern catalog referenced by `policy.yaml`. ```bash uv run egress-gate --debug serve \ --listen 0.0.0.0:50051 \ - --timeout-seconds 4 + --timeout 4s ``` Before you change the gateway configuration, stop any running OpenShell diff --git a/projects/egress-gate/src/egress_gate/cli.py b/projects/egress-gate/src/egress_gate/cli.py index fa8e9d1a..e919052f 100644 --- a/projects/egress-gate/src/egress_gate/cli.py +++ b/projects/egress-gate/src/egress_gate/cli.py @@ -29,14 +29,14 @@ from egress_gate.base import StrictDomainModel from egress_gate.constants import ( - DEFAULT_TIMEOUT_SECONDS, + DEFAULT_GATEWAY_REGISTRATION_TIMEOUT, + DEFAULT_TIMEOUT_MIDDLEWARE_PROCESSING, MAX_BODY_BYTES, MAX_EVALUATION_CASE_NAME_BYTES, MAX_EVALUATION_CASES, MAX_EVALUATION_FILE_BYTES, MAX_EVALUATION_TAGS, MAX_PROTO_FINDING_GROUPS, - MAX_TIMEOUT_SECONDS, ) from egress_gate.errors import EgressGateError, GateRegistryError from egress_gate.gates.base import GateCapability @@ -52,16 +52,31 @@ GatewayConfigUpdate, GatewayMiddlewareRegistration, default_gateway_config_path, + forget_gateway_registration, list_gateway_registrations, + read_remembered_gateway_timeout, + remember_gateway_registration, remove_gateway_config, update_gateway_config, + validate_gateway_timeout, validate_middleware_name, ) -from egress_gate.logging import LoggingConfig, configure_logging +from egress_gate.logging import LoggingConfig, configure_logging, get_logger from egress_gate.request import HttpHeader, HttpRequest, HttpTarget, RequestContext from egress_gate.result import EgressResult, GateDecisionSource from egress_gate.string_validators import BoundedMetadataString -from egress_gate.timeout import Timeout, validate_timeout_seconds +from egress_gate.timeout import ( + Timeout, + parse_timeout_duration, + validate_timeout_middleware_processing, +) + +_DURATION_FORMAT_HELP = ( + "Use an integer followed by s for seconds or ms for milliseconds, such as " + "10s or 500ms." +) +_TIMEOUT_DURATION_HELP = f"{_DURATION_FORMAT_HELP} Minimum 10ms." +_LOG = get_logger(__name__) app = typer.Typer( name="egress-gate", @@ -138,31 +153,59 @@ def serve( ), ), ] = "127.0.0.1:50051", - timeout_seconds: Annotated[ - float, + timeout: Annotated[ + str, typer.Option( + "--timeout", help=( - "Total processing time available to all gates for one request. " - f"The value must be greater than 0 and at most {MAX_TIMEOUT_SECONDS:g}." + "Internal processing budget for one request. " + f"{_TIMEOUT_DURATION_HELP} The OpenShell gateway applies its " + "separately configured RPC timeout." ), ), - ] = DEFAULT_TIMEOUT_SECONDS, + ] = f"{DEFAULT_TIMEOUT_MIDDLEWARE_PROCESSING:g}s", ) -> None: """Start the Egress Gate gRPC service and run until shutdown.""" options = _command_options(context) from egress_gate.service.server import EgressGateServer try: - validated_timeout_seconds = validate_timeout_seconds(timeout_seconds) + timeout_middleware_processing = parse_timeout_duration(timeout) except ValueError as error: raise typer.BadParameter( str(error), - param_hint="--timeout-seconds", + param_hint="--timeout", ) from None + try: + remembered_timeout = read_remembered_gateway_timeout() + except GatewayConfigError as error: + _render_cli_error( + "Gateway timeout could not be validated", + code="gateway_config_error", + message=str(error), + ) + raise typer.Exit(code=1) from None + if remembered_timeout is not None: + remembered, timeout_gateway_ceiling = remembered_timeout + if timeout_middleware_processing >= timeout_gateway_ceiling: + raise typer.BadParameter( + "The middleware processing timeout must be less than the " + f"{timeout_gateway_ceiling:g}s gateway timeout configured for " + f"{remembered.middleware_name!r} in {remembered.config_path}.", + param_hint="--timeout", + ) + _LOG.info( + "Validated timeout_middleware_processing=%ss against " + "timeout_gateway_ceiling=%ss registration=%s gateway_config=%s", + timeout_middleware_processing, + timeout_gateway_ceiling, + remembered.middleware_name, + remembered.config_path, + ) try: EgressGateServer( options.registry, - timeout_seconds=validated_timeout_seconds, + timeout_middleware_processing=timeout_middleware_processing, ).serve_sync(listen) except EgressGateError as error: _render_egress_error("Egress Gate could not start", error) @@ -213,8 +256,18 @@ def add_gateway_registration( ), ), ] = 50051, + timeout: Annotated[ + str, + typer.Option( + "--timeout", + help=( + "Gateway RPC timeout to write in the registration. " + f"{_DURATION_FORMAT_HELP}" + ), + ), + ] = DEFAULT_GATEWAY_REGISTRATION_TIMEOUT, ) -> None: - """Add or update Egress Gate in an OpenShell gateway TOML file.""" + """Add or update Egress Gate with a configurable gateway RPC timeout.""" try: address = ipaddress.IPv4Address(host_ip) except ipaddress.AddressValueError: @@ -235,7 +288,13 @@ def add_gateway_registration( str(error), param_hint="--name", ) from None - + try: + validate_gateway_timeout(timeout) + except GatewayConfigError as error: + raise typer.BadParameter( + str(error), + param_hint="--timeout", + ) from None config_path = config or default_gateway_config_path() try: result = update_gateway_config( @@ -243,6 +302,11 @@ def add_gateway_registration( middleware_name=validated_name, host_ip=str(address), port=port, + timeout_gateway_ceiling=timeout, + ) + remember_gateway_registration( + config_path, + middleware_name=validated_name, ) except GatewayConfigError as error: _render_cli_error( @@ -263,6 +327,7 @@ def add_gateway_registration( config_path=config_path, name=validated_name, endpoint=f"http://{address}:{port}", + timeout_gateway_ceiling=timeout, change=change, next_step=( "Start Egress Gate, then restart the OpenShell gateway to load this " @@ -329,6 +394,10 @@ def remove_gateway_registration( config_path, middleware_name=name, ) + forget_gateway_registration( + config_path, + middleware_name=name, + ) except GatewayConfigError as error: _render_cli_error( "Gateway registration could not be removed", @@ -441,24 +510,25 @@ def evaluate( help="Path to the YAML file of saved request cases and expected results.", ), ], - timeout_seconds: Annotated[ - float, + timeout: Annotated[ + str, typer.Option( + "--timeout", help=( - "Maximum seconds for policy preparation and, separately, each case. " - f"The value must be greater than 0 and at most {MAX_TIMEOUT_SECONDS:g}." + "Timeout for policy preparation and, separately, each case. " + f"{_TIMEOUT_DURATION_HELP}" ), ), - ] = DEFAULT_TIMEOUT_SECONDS, + ] = f"{DEFAULT_TIMEOUT_MIDDLEWARE_PROCESSING:g}s", ) -> None: """Test saved requests against a policy without starting the service.""" options = _command_options(context) try: - validated_timeout_seconds = validate_timeout_seconds(timeout_seconds) + timeout_seconds = parse_timeout_duration(timeout) except ValueError as error: raise typer.BadParameter( str(error), - param_hint="--timeout-seconds", + param_hint="--timeout", ) from None try: policy_values = _load_policy(policy) @@ -485,7 +555,7 @@ def evaluate( options.registry, policy_values, corpus, - timeout_seconds=validated_timeout_seconds, + timeout_seconds=timeout_seconds, ) except _CaseExecutionError as error: if error.completed: @@ -854,7 +924,7 @@ def _run_corpus( timeout_seconds: float, ) -> _EvaluationSummary: """Prepare once, then evaluate every case with a fresh shared timeout.""" - validated_timeout = validate_timeout_seconds(timeout_seconds) + validated_timeout = validate_timeout_middleware_processing(timeout_seconds) validated_config = registry.validate_config(policy_values) processor = registry.prepare_processor( validated_config, @@ -988,6 +1058,7 @@ def _render_registration( config_path: Path, name: str, endpoint: str | None = None, + timeout_gateway_ceiling: str | None = None, change: str | None = None, next_step: str | None = None, status_style: str = "bold green", @@ -1001,6 +1072,11 @@ def _render_registration( details.add_row("Registration", Text(name)) if endpoint is not None: details.add_row("Endpoint", Text(endpoint)) + if timeout_gateway_ceiling is not None: + details.add_row( + "Gateway RPC ceiling", + Text(timeout_gateway_ceiling), + ) if change is not None: details.add_row("Change", Text(change)) _CONSOLE.print(details) @@ -1022,8 +1098,13 @@ def _render_gateway_registrations( table = Table(box=None, pad_edge=False, padding=(0, 2), header_style="bold cyan") table.add_column("Name", style="bold", no_wrap=True) table.add_column("Endpoint", overflow="fold") + table.add_column("Gateway RPC ceiling", no_wrap=True) for registration in registrations: - table.add_row(registration.name, registration.endpoint or "Not set") + table.add_row( + registration.name, + registration.endpoint or "Not set", + registration.timeout_gateway_ceiling or "Not set", + ) _CONSOLE.print(table) _CONSOLE.print( Text.assemble( diff --git a/projects/egress-gate/src/egress_gate/constants.py b/projects/egress-gate/src/egress_gate/constants.py index 06716351..d9de72ce 100644 --- a/projects/egress-gate/src/egress_gate/constants.py +++ b/projects/egress-gate/src/egress_gate/constants.py @@ -9,9 +9,10 @@ import re from importlib.metadata import version -# Configurable processing timeout. -DEFAULT_TIMEOUT_SECONDS = 1.0 -MAX_TIMEOUT_SECONDS = 30.0 +# Timeout defaults. The gateway registration value remains independently +# configurable by the operator. +DEFAULT_TIMEOUT_MIDDLEWARE_PROCESSING = 1.0 +DEFAULT_GATEWAY_REGISTRATION_TIMEOUT = "30s" # Middleware identity and stable response values. SERVICE_NAME = "egress-gate" @@ -21,12 +22,8 @@ LIMIT_REASON = ( "Egress Gate exceeded a processing safety limit. Check Egress Gate logs " "for the limit kind. Reduce the request or replacement size, simplify the " - "configured gates and rules, or increase the processing timeout with " - "--timeout-seconds or " - "EgressGateServer(timeout_seconds=...) to at most " - f"{MAX_TIMEOUT_SECONDS:g} seconds. If increasing it, give OpenShell's " - "middleware timeout additional headroom for queueing and configuration " - "preparation, then retry." + "configured gates and rules, or increase the middleware processing timeout, " + "then retry." ) LIMIT_REASON_CODE = "egress_gate_limit_exceeded" # Text input limits. diff --git a/projects/egress-gate/src/egress_gate/errors.py b/projects/egress-gate/src/egress_gate/errors.py index 3d99f5cf..547ce059 100644 --- a/projects/egress-gate/src/egress_gate/errors.py +++ b/projects/egress-gate/src/egress_gate/errors.py @@ -5,7 +5,7 @@ from dataclasses import dataclass from enum import StrEnum -from egress_gate.constants import MAX_PROTO_CONFIG_BYTES, MAX_TIMEOUT_SECONDS +from egress_gate.constants import MAX_PROTO_CONFIG_BYTES class ErrorKind(StrEnum): @@ -106,8 +106,8 @@ class TimeoutExpiredError(Exception): def __init__(self) -> None: super().__init__( "Egress Gate processing timed out. Reduce the request size or simplify " - "the configured gates and rules, or increase the processing timeout " - f"to at most {MAX_TIMEOUT_SECONDS:g} seconds, then retry." + "the configured gates and rules, or increase the middleware processing " + "timeout, then retry." ) diff --git a/projects/egress-gate/src/egress_gate/gateway_config.py b/projects/egress-gate/src/egress_gate/gateway_config.py index c6b0eef8..fdbefdd6 100644 --- a/projects/egress-gate/src/egress_gate/gateway_config.py +++ b/projects/egress-gate/src/egress_gate/gateway_config.py @@ -11,6 +11,10 @@ from dataclasses import dataclass from enum import Enum from pathlib import Path +from typing import Literal + +from egress_gate.constants import DEFAULT_GATEWAY_REGISTRATION_TIMEOUT +from egress_gate.timeout import parse_duration class GatewayConfigUpdate(Enum): @@ -39,6 +43,15 @@ class GatewayMiddlewareRegistration: name: str endpoint: str | None + timeout_gateway_ceiling: str | None + + +@dataclass(frozen=True) +class RememberedGatewayRegistration: + """The gateway registration most recently managed by Egress Gate.""" + + config_path: Path + middleware_name: str # Mirrors OpenShell's stable-identifier byte limit for external middleware @@ -59,6 +72,124 @@ def default_gateway_config_path() -> Path: return Path.home() / ".config" / "openshell" / "gateway.toml" +def default_registration_state_path() -> Path: + """Return the per-user location for the remembered gateway registration.""" + config_home = os.environ.get("XDG_CONFIG_HOME") + root = Path(config_home) if config_home else Path.home() / ".config" + return root / "openshell-egress-gate" / "registration.toml" + + +def remember_gateway_registration( + config_path: Path, + *, + middleware_name: str, +) -> None: + """Remember where the CLI most recently managed an Egress Gate registration.""" + validate_middleware_name(middleware_name) + escaped_path = ( + str(config_path.expanduser().resolve()) + .replace("\\", "\\\\") + .replace('"', '\\"') + ) + escaped_name = middleware_name.replace("\\", "\\\\").replace('"', '\\"') + _write_atomically( + default_registration_state_path(), + f'gateway_config = "{escaped_path}"\nregistration_name = "{escaped_name}"\n', + ) + + +def forget_gateway_registration( + config_path: Path, + *, + middleware_name: str, +) -> None: + """Forget the registration when it matches the CLI-managed registration.""" + remembered = load_remembered_gateway_registration() + if remembered is None or ( + remembered.config_path != config_path.expanduser().resolve() + or remembered.middleware_name != middleware_name + ): + return + state_path = default_registration_state_path() + try: + state_path.unlink(missing_ok=True) + except OSError as error: + raise GatewayConfigError( + f"Could not remove {state_path}. Check that its directory is writable." + ) from error + + +def load_remembered_gateway_registration() -> RememberedGatewayRegistration | None: + """Load the gateway registration most recently managed by the CLI.""" + state_path = default_registration_state_path() + try: + contents = state_path.read_text(encoding="utf-8") + except FileNotFoundError: + return None + except (OSError, UnicodeError) as error: + raise GatewayConfigError( + f"Could not read {state_path}. Check that it is readable UTF-8 TOML." + ) from error + try: + values = tomllib.loads(contents) + except tomllib.TOMLDecodeError as error: + raise GatewayConfigError( + f"Could not parse {state_path}. Remove it and register Egress Gate again." + ) from error + config_path = values.get("gateway_config") + middleware_name = values.get("registration_name") + if not isinstance(config_path, str) or not config_path: + raise GatewayConfigError( + f"{state_path} does not contain a valid gateway_config path. Remove it " + "and register Egress Gate again." + ) + if not isinstance(middleware_name, str): + raise GatewayConfigError( + f"{state_path} does not contain a valid registration_name. Remove it " + "and register Egress Gate again." + ) + validate_middleware_name(middleware_name) + return RememberedGatewayRegistration( + config_path=Path(config_path), + middleware_name=middleware_name, + ) + + +def read_remembered_gateway_timeout( + unit: Literal["s", "ms"] = "s", +) -> tuple[RememberedGatewayRegistration, float] | None: + """Read the current timeout for the remembered gateway registration.""" + remembered = load_remembered_gateway_registration() + if remembered is None: + return None + matches = [ + registration + for registration in list_gateway_registrations(remembered.config_path) + if registration.name == remembered.middleware_name + ] + if not matches: + raise GatewayConfigError( + f"The remembered registration {remembered.middleware_name!r} is not in " + f"{remembered.config_path}. Register Egress Gate again." + ) + duration = matches[0].timeout_gateway_ceiling + if duration is None: + raise GatewayConfigError( + f"The remembered registration {remembered.middleware_name!r} in " + f"{remembered.config_path} has no timeout. Add one or register Egress " + "Gate again." + ) + try: + timeout = parse_duration(duration, unit=unit) + except ValueError: + raise GatewayConfigError( + f"The timeout for the remembered registration " + f"{remembered.middleware_name!r} in {remembered.config_path} must use " + "whole seconds or milliseconds, such as 30s or 500ms." + ) from None + return remembered, timeout + + def list_gateway_registrations( path: Path, ) -> tuple[GatewayMiddlewareRegistration, ...]: @@ -79,6 +210,7 @@ def list_gateway_registrations( for entry in _middleware_entries(_load_gateway_config(contents, path), path): name = entry.get("name") endpoint = entry.get("grpc_endpoint") + timeout_gateway_ceiling = entry.get("timeout") if not isinstance(name, str) or not name: raise GatewayConfigError( f"{path} contains a middleware registration without a valid name." @@ -88,8 +220,19 @@ def list_gateway_registrations( f"The middleware registration {name!r} in {path} has an invalid " "grpc_endpoint." ) + if timeout_gateway_ceiling is not None and not isinstance( + timeout_gateway_ceiling, str + ): + raise GatewayConfigError( + f"The middleware registration {name!r} in {path} has an invalid " + "timeout." + ) registrations.append( - GatewayMiddlewareRegistration(name=name, endpoint=endpoint) + GatewayMiddlewareRegistration( + name=name, + endpoint=endpoint, + timeout_gateway_ceiling=timeout_gateway_ceiling, + ) ) return tuple(registrations) @@ -100,9 +243,11 @@ def update_gateway_config( middleware_name: str, host_ip: str, port: int, + timeout_gateway_ceiling: str = DEFAULT_GATEWAY_REGISTRATION_TIMEOUT, ) -> GatewayConfigUpdate: """Add or update one named Egress Gate middleware registration.""" validate_middleware_name(middleware_name) + validate_gateway_timeout(timeout_gateway_ceiling) endpoint = f"http://{host_ip}:{port}" try: original = path.read_text(encoding="utf-8") @@ -110,6 +255,7 @@ def update_gateway_config( updated = _new_gateway_config( middleware_name=middleware_name, endpoint=endpoint, + timeout_gateway_ceiling=timeout_gateway_ceiling, ) _write_atomically(path, updated) return GatewayConfigUpdate.CREATED @@ -122,6 +268,7 @@ def update_gateway_config( updated = _new_gateway_config( middleware_name=middleware_name, endpoint=endpoint, + timeout_gateway_ceiling=timeout_gateway_ceiling, ) _write_atomically(path, updated) return GatewayConfigUpdate.CREATED @@ -151,6 +298,7 @@ def update_gateway_config( replacement = _update_middleware_block( block.group(0), endpoint=endpoint, + timeout_gateway_ceiling=timeout_gateway_ceiling, ) updated = original[: block.start()] + replacement + original[block.end() :] result = GatewayConfigUpdate.UPDATED @@ -159,6 +307,7 @@ def update_gateway_config( original, middleware_name=middleware_name, endpoint=endpoint, + timeout_gateway_ceiling=timeout_gateway_ceiling, ) result = GatewayConfigUpdate.ADDED @@ -255,10 +404,31 @@ def validate_middleware_name(name: str) -> str: return name -def _new_gateway_config(*, middleware_name: str, endpoint: str) -> str: +def validate_gateway_timeout(duration: str) -> float: + """Validate a gateway timeout that can exceed the processing minimum.""" + message = ( + "The gateway timeout must be greater than 10ms and use whole seconds or " + "milliseconds, such as 30s or 500ms." + ) + try: + seconds = parse_duration(duration) + except ValueError: + raise GatewayConfigError(message) from None + if seconds <= 0.01: + raise GatewayConfigError(message) + return seconds + + +def _new_gateway_config( + *, + middleware_name: str, + endpoint: str, + timeout_gateway_ceiling: str, +) -> str: return "[openshell]\nversion = 1\n\n" + _middleware_block( middleware_name=middleware_name, endpoint=endpoint, + timeout_gateway_ceiling=timeout_gateway_ceiling, ) @@ -319,6 +489,7 @@ def _append_middleware_block( *, middleware_name: str, endpoint: str, + timeout_gateway_ceiling: str, ) -> str: return ( contents.rstrip() @@ -326,21 +497,32 @@ def _append_middleware_block( + _middleware_block( middleware_name=middleware_name, endpoint=endpoint, + timeout_gateway_ceiling=timeout_gateway_ceiling, ) ) -def _middleware_block(*, middleware_name: str, endpoint: str) -> str: +def _middleware_block( + *, + middleware_name: str, + endpoint: str, + timeout_gateway_ceiling: str, +) -> str: return ( "[[openshell.supervisor.middleware]]\n" f'name = "{middleware_name}"\n' f'grpc_endpoint = "{endpoint}"\n' "max_body_bytes = 4194304\n" - 'timeout = "5s"\n' + f'timeout = "{timeout_gateway_ceiling}"\n' ) -def _update_middleware_block(block: str, *, endpoint: str) -> str: +def _update_middleware_block( + block: str, + *, + endpoint: str, + timeout_gateway_ceiling: str, +) -> str: updated = _replace_or_append_assignment( block, key="grpc_endpoint", @@ -354,7 +536,7 @@ def _update_middleware_block(block: str, *, endpoint: str) -> str: return _replace_or_append_assignment( updated, key="timeout", - value='"5s"', + value=f'"{timeout_gateway_ceiling}"', ) @@ -429,12 +611,19 @@ def _write_atomically(path: Path, contents: str) -> None: __all__ = [ "GatewayConfigError", "GatewayMiddlewareRegistration", + "RememberedGatewayRegistration", "GatewayConfigRemoval", "GatewayConfigUpdate", "MAX_MIDDLEWARE_REGISTRATION_NAME_BYTES", "default_gateway_config_path", + "default_registration_state_path", + "forget_gateway_registration", "list_gateway_registrations", + "load_remembered_gateway_registration", + "remember_gateway_registration", + "read_remembered_gateway_timeout", "remove_gateway_config", "update_gateway_config", + "validate_gateway_timeout", "validate_middleware_name", ] diff --git a/projects/egress-gate/src/egress_gate/service/server.py b/projects/egress-gate/src/egress_gate/service/server.py index cd5857a4..6da9146c 100644 --- a/projects/egress-gate/src/egress_gate/service/server.py +++ b/projects/egress-gate/src/egress_gate/service/server.py @@ -11,7 +11,7 @@ from egress_gate.bindings import supervisor_middleware_pb2_grpc as pb2_grpc from egress_gate.constants import ( - DEFAULT_TIMEOUT_SECONDS, + DEFAULT_TIMEOUT_MIDDLEWARE_PROCESSING, MAX_CONCURRENT_RPCS, MAX_RECEIVE_MESSAGE_BYTES, ) @@ -30,11 +30,11 @@ def __init__( self, registry: GateRegistry, *, - timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + timeout_middleware_processing: float = DEFAULT_TIMEOUT_MIDDLEWARE_PROCESSING, ) -> None: self._middleware = EgressGateMiddleware( registry, - timeout_seconds=timeout_seconds, + timeout_middleware_processing=timeout_middleware_processing, ) def serve_sync(self, listen: str = DEFAULT_LISTEN_ADDRESS) -> None: @@ -53,7 +53,12 @@ async def serve_async(self, listen: str = DEFAULT_LISTEN_ADDRESS) -> None: bound_port = server.add_insecure_port(listen) if bound_port != requested_port: raise EgressGateError(ErrorCode.SERVER_BIND_FAILED) - _LOGGER.info("egress_gate_server_bound listen=%r", listen) + _LOGGER.info( + "egress_gate_server_bound listen=%r " + "timeout_middleware_processing=%s", + listen, + self._middleware.timeout_middleware_processing, + ) await server.start() except RuntimeError: raise EgressGateError(ErrorCode.SERVER_BIND_FAILED) from None diff --git a/projects/egress-gate/src/egress_gate/service/servicer.py b/projects/egress-gate/src/egress_gate/service/servicer.py index 965365be..78fdc3d9 100644 --- a/projects/egress-gate/src/egress_gate/service/servicer.py +++ b/projects/egress-gate/src/egress_gate/service/servicer.py @@ -20,7 +20,7 @@ from egress_gate.config import EgressGateConfig from egress_gate.constants import ( BLOCK_REASON, - DEFAULT_TIMEOUT_SECONDS, + DEFAULT_TIMEOUT_MIDDLEWARE_PROCESSING, LIMIT_REASON, LIMIT_REASON_CODE, MAX_BODY_BYTES, @@ -65,7 +65,11 @@ SourcedFinding, ) from egress_gate.string_validators import validate_bounded_metadata_string -from egress_gate.timeout import Timeout, validate_timeout_seconds +from egress_gate.timeout import ( + Timeout, + format_timeout_middleware_processing, + validate_timeout_middleware_processing, +) class EgressGateMiddleware(pb2_grpc.SupervisorMiddlewareServicer): @@ -75,11 +79,13 @@ def __init__( self, registry: GateRegistry, *, - timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + timeout_middleware_processing: float = DEFAULT_TIMEOUT_MIDDLEWARE_PROCESSING, ) -> None: registry.configuration_json_schema() self._registry = registry - self._timeout_seconds = validate_timeout_seconds(timeout_seconds) + self._timeout_middleware_processing_seconds = ( + validate_timeout_middleware_processing(timeout_middleware_processing) + ) self._policy = _ActivePolicy(registry) self._processing_slots = asyncio.Semaphore(MAX_CONCURRENT_PROCESSING) self._processing_executor = ThreadPoolExecutor( @@ -87,6 +93,13 @@ def __init__( thread_name_prefix="egress-gate-processing", ) + @property + def timeout_middleware_processing(self) -> str: + """Return the configured middleware processing timeout.""" + return format_timeout_middleware_processing( + self._timeout_middleware_processing_seconds + ) + async def close(self) -> None: """Wait for in-flight synchronous gates during shutdown.""" self._processing_executor.shutdown(wait=True, cancel_futures=True) @@ -97,7 +110,10 @@ async def Describe( request: object, context: grpc.aio.ServicerContext[object, pb2.MiddlewareManifest], ) -> pb2.MiddlewareManifest: - """Advertise the binding and its complete policy schema.""" + """Describe the binding and its complete policy schema.""" + # The protocol does not expose the operator-configured gateway timeout + # to this service. An empty binding timeout leaves that RPC limit under + # gateway ownership instead of replacing it with the internal budget. return pb2.MiddlewareManifest( name=SERVICE_NAME, service_version=SERVICE_VERSION, @@ -159,7 +175,7 @@ async def _evaluate_rpc( finding_count = 0 source_kind = "none" try: - timeout = Timeout.from_seconds(self._timeout_seconds) + timeout = Timeout.from_seconds(self._timeout_middleware_processing_seconds) response, source_kind = await self._evaluate_http_request( request, timeout, diff --git a/projects/egress-gate/src/egress_gate/timeout.py b/projects/egress-gate/src/egress_gate/timeout.py index cad58c55..9d108c36 100644 --- a/projects/egress-gate/src/egress_gate/timeout.py +++ b/projects/egress-gate/src/egress_gate/timeout.py @@ -3,31 +3,91 @@ from __future__ import annotations import math +import re from collections.abc import Iterator from contextlib import contextmanager from time import monotonic -from typing import Self +from typing import Literal, Self -from pydantic import Field +from pydantic import Field, ValidationError from egress_gate.base import StrictDomainModel -from egress_gate.constants import MAX_TIMEOUT_SECONDS from egress_gate.errors import TimeoutExpiredError +TIMEOUT_DURATION_PATTERN = r"^(?P[1-9][0-9]{0,8})(?Pms|s)$" +_TIMEOUT_DURATION_PATTERN = re.compile(TIMEOUT_DURATION_PATTERN) -def validate_timeout_seconds(seconds: object) -> float: - """Return a finite supported processing timeout in seconds.""" + +class _DurationValue(StrictDomainModel): + """A duration string validated with the shared OpenShell-style pattern.""" + + value: str = Field(pattern=TIMEOUT_DURATION_PATTERN) + + +def validate_timeout_middleware_processing(seconds: float) -> float: + """Return a supported internal processing timeout, in seconds.""" if ( isinstance(seconds, bool) or not isinstance(seconds, int | float) or not math.isfinite(seconds) - or seconds <= 0 - or seconds > MAX_TIMEOUT_SECONDS ): raise ValueError( - "timeout seconds must be a finite number greater than 0 and at most " - f"{MAX_TIMEOUT_SECONDS:g}" + "timeout_middleware_processing must be at least 10ms, using whole " + "milliseconds" + ) + validated_seconds = float(seconds) + milliseconds = validated_seconds * 1000 + if not math.isfinite(milliseconds): + raise ValueError( + "timeout_middleware_processing must be at least 10ms, using whole " + "milliseconds" + ) + rounded_milliseconds = round(milliseconds) + if rounded_milliseconds < 10 or not math.isclose( + milliseconds, rounded_milliseconds + ): + raise ValueError( + "timeout_middleware_processing must be at least 10ms, using whole " + "milliseconds" ) + return validated_seconds + + +def format_timeout_middleware_processing(seconds: float) -> str: + """Format a processing timeout for the OpenShell duration contract.""" + validated_seconds = validate_timeout_middleware_processing(seconds) + rounded_milliseconds = round(validated_seconds * 1000) + if rounded_milliseconds % 1000 == 0: + return f"{rounded_milliseconds // 1000}s" + return f"{rounded_milliseconds}ms" + + +def parse_duration( + duration: str, + unit: Literal["s", "ms"] = "s", +) -> float: + """Parse a validated OpenShell-style duration into the requested unit.""" + try: + validated_duration = _DurationValue(value=duration).value + except ValidationError: + raise ValueError("timeout must be an integer duration such as 10s or 500ms") + match = _TIMEOUT_DURATION_PATTERN.fullmatch(validated_duration) + if match is None: + raise AssertionError("Pydantic accepted an unmatched timeout duration") + magnitude = int(match.group("magnitude")) + milliseconds = magnitude if match.group("unit") == "ms" else magnitude * 1000 + return milliseconds / 1000 if unit == "s" else float(milliseconds) + + +def parse_timeout_duration(duration: str) -> float: + """Parse and validate an internal middleware processing duration.""" + seconds = parse_duration(duration) + try: + format_timeout_middleware_processing(seconds) + except ValueError: + raise ValueError( + "timeout must be at least 10ms, using whole milliseconds" + ) from None return float(seconds) @@ -38,8 +98,10 @@ class Timeout(StrictDomainModel): @classmethod def from_seconds(cls, seconds: float) -> Self: - """Create a timeout from a finite, positive bounded duration.""" - return cls(deadline=monotonic() + validate_timeout_seconds(seconds)) + """Create a timeout from a finite positive duration.""" + return cls( + deadline=monotonic() + validate_timeout_middleware_processing(seconds) + ) def remaining_seconds(self) -> float: """Return the positive duration remaining or raise ``TimeoutExpiredError``.""" @@ -63,4 +125,11 @@ def enforce(self) -> Iterator[None]: self.raise_if_expired() -__all__ = ["Timeout", "validate_timeout_seconds"] +__all__ = [ + "TIMEOUT_DURATION_PATTERN", + "Timeout", + "format_timeout_middleware_processing", + "parse_duration", + "parse_timeout_duration", + "validate_timeout_middleware_processing", +] diff --git a/projects/egress-gate/tests/gates/test_regex.py b/projects/egress-gate/tests/gates/test_regex.py index e002bb91..d9a19bab 100644 --- a/projects/egress-gate/tests/gates/test_regex.py +++ b/projects/egress-gate/tests/gates/test_regex.py @@ -523,7 +523,7 @@ def test_pattern_search_has_an_enforceable_timeout() -> None: with pytest.raises(TimeoutExpiredError): RegexGate(config, None).evaluate( _request((b"a" * 100_000) + b"!"), - timeout=Timeout.from_seconds(0.001), + timeout=Timeout.from_seconds(0.01), ) diff --git a/projects/egress-gate/tests/service/test_server.py b/projects/egress-gate/tests/service/test_server.py index 736ce38f..4cdea0ef 100644 --- a/projects/egress-gate/tests/service/test_server.py +++ b/projects/egress-gate/tests/service/test_server.py @@ -45,17 +45,33 @@ def test_server_rejects_a_registry_without_gates() -> None: EgressGateServer(GateRegistry()) -@pytest.mark.parametrize("seconds", [True, 0, 31, float("inf")]) -def test_server_validates_the_service_timeout(seconds: bool | int | float) -> None: - with pytest.raises(ValueError, match="finite number greater than 0 and at most 30"): - EgressGateServer(create_builtin_registry(), timeout_seconds=seconds) +@pytest.mark.parametrize( + "seconds", + [True, 0, 0.001, 1.0001, float("inf")], +) +def test_server_validates_timeout_middleware_processing( + seconds: bool | int | float, +) -> None: + with pytest.raises(ValueError, match="timeout"): + EgressGateServer( + create_builtin_registry(), + timeout_middleware_processing=seconds, + ) -def test_server_keeps_timeout_ownership_at_the_service_boundary() -> None: - server = EgressGateServer(create_builtin_registry(), timeout_seconds=4.5) +def test_server_uses_timeout_middleware_processing_for_pipeline_and_logs() -> None: + server = EgressGateServer( + create_builtin_registry(), + timeout_middleware_processing=4.5, + ) try: - assert server._middleware._timeout_seconds == 4.5 - assert not hasattr(server._middleware._policy, "_timeout_seconds") + assert server._middleware._timeout_middleware_processing_seconds == 4.5 + assert server._middleware.timeout_middleware_processing == "4500ms" + assert not hasattr(server._middleware, "_timeout_middleware_processing") + assert not hasattr( + server._middleware._policy, + "_timeout_middleware_processing_seconds", + ) finally: asyncio.run(server._middleware.close()) diff --git a/projects/egress-gate/tests/service/test_servicer.py b/projects/egress-gate/tests/service/test_servicer.py index 018c95df..44392d76 100644 --- a/projects/egress-gate/tests/service/test_servicer.py +++ b/projects/egress-gate/tests/service/test_servicer.py @@ -109,6 +109,19 @@ async def abort(self, code: grpc.StatusCode, details: str) -> Never: raise AssertionError("successful evaluation unexpectedly aborted") +def test_manifest_leaves_the_gateway_rpc_timeout_to_the_operator() -> None: + middleware = EgressGateMiddleware( + create_builtin_registry(), + timeout_middleware_processing=4.5, + ) + try: + manifest = asyncio.run(middleware.Describe(object(), Mock())) + finally: + asyncio.run(middleware.close()) + + assert manifest.bindings[0].timeout == "" + + def test_copied_proto_remains_the_current_five_field_finding_contract() -> None: evaluation = pb2.HttpRequestEvaluation() finding = pb2.Finding() @@ -425,10 +438,7 @@ def test_in_flight_processor_reference_survives_policy_replacement() -> None: async def test_cancelled_candidate_keeps_its_slot_until_worker_exits( monkeypatch: pytest.MonkeyPatch, ) -> None: - middleware = EgressGateMiddleware( - create_builtin_registry(), - timeout_seconds=5, - ) + middleware = EgressGateMiddleware(create_builtin_registry()) started = Event() release = Event() original_build = middleware._registry.prepare_processor diff --git a/projects/egress-gate/tests/test_cli.py b/projects/egress-gate/tests/test_cli.py index e25a19e5..aaeafad3 100644 --- a/projects/egress-gate/tests/test_cli.py +++ b/projects/egress-gate/tests/test_cli.py @@ -54,10 +54,85 @@ def test_cli_narrow_help_preserves_complete_option_names() -> None: assert result.exit_code == 0 assert "--host-ip" in result.stdout assert "--config" in result.stdout + assert "--timeout" in result.stdout assert "--host…" not in result.stdout assert "--conf…" not in result.stdout +def test_cli_serve_uses_one_concise_processing_timeout( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + calls: list[tuple[float, str]] = [] + + class FakeServer: + def __init__( + self, + registry: GateRegistry, + *, + timeout_middleware_processing: float, + ) -> None: + del registry + self.timeout_middleware_processing = timeout_middleware_processing + + def serve_sync(self, listen: str) -> None: + calls.append((self.timeout_middleware_processing, listen)) + + monkeypatch.setattr( + "egress_gate.service.server.EgressGateServer", + FakeServer, + ) + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + + result = CliRunner().invoke( + app, + ["serve", "--listen", "127.0.0.1:50055", "--timeout", "4500ms"], + ) + + assert result.exit_code == 0, result.output + assert calls == [(4.5, "127.0.0.1:50055")] + + help_result = CliRunner().invoke(app, ["serve", "--help"]) + serve_help = " ".join(help_result.stdout.split()) + assert "--timeout " in serve_help + assert "--timeout-seconds" not in serve_help + assert "s for seconds or ms for milliseconds" in serve_help + assert "Minimum 10ms" in serve_help + assert "RPC timeout" in serve_help + + evaluate_help = CliRunner().invoke(app, ["evaluate", "--help"]) + assert evaluate_help.exit_code == 0, evaluate_help.output + normalized_evaluate_help = " ".join(evaluate_help.stdout.split()) + assert "s for seconds or ms for milliseconds" in normalized_evaluate_help + assert "Minimum 10ms" in normalized_evaluate_help + + +def test_cli_serve_rejects_timeout_at_remembered_gateway_ceiling( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "user-config")) + config = tmp_path / "gateway.toml" + registration = CliRunner().invoke( + app, + [ + "add-gateway-registration", + "--host-ip", + "192.0.2.10", + "--config", + str(config), + ], + ) + assert registration.exit_code == 0, registration.output + config.write_text(config.read_text().replace('timeout = "30s"', 'timeout = "1s"')) + + result = CliRunner().invoke(app, ["serve", "--timeout", "1s"]) + + assert result.exit_code == 2 + assert "must be less than the 1s gateway timeout" in result.stderr + assert str(config) in result.stderr + + def test_cli_gates_describes_the_request_level_builtin() -> None: result = CliRunner().invoke(app, ["gates", "list"]) @@ -388,7 +463,11 @@ def test_cli_evaluate_names_a_failing_case_and_keeps_completed_results( assert '"/w=="' not in result.output -def test_cli_add_gateway_registration_reports_the_result(tmp_path: Path) -> None: +def test_cli_add_gateway_registration_reports_the_result( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "user-config")) config = tmp_path / "gateway.toml" result = CliRunner().invoke( app, @@ -398,6 +477,8 @@ def test_cli_add_gateway_registration_reports_the_result(tmp_path: Path) -> None "192.0.2.10", "--config", str(config), + "--timeout", + "45s", ], ) @@ -405,12 +486,91 @@ def test_cli_add_gateway_registration_reports_the_result(tmp_path: Path) -> None assert "Gateway registration is ready" in result.stdout assert "Gateway file" in result.stdout assert str(config) in "".join(result.stdout.split()) - assert "Registration egress-gate" in result.stdout - assert "Endpoint http://192.0.2.10:50051" in result.stdout + assert "Registration" in result.stdout + assert "egress-gate" in result.stdout + assert "Endpoint" in result.stdout + assert "http://192.0.2.10:50051" in result.stdout + assert "Gateway RPC ceiling" in result.stdout + assert "45s" in result.stdout + assert 'timeout = "45s"' in config.read_text() assert "Created the gateway configuration file" in result.stdout assert "Next: Start Egress Gate" in result.stdout +def test_cli_rejects_gateway_timeout_without_processing_headroom( + tmp_path: Path, +) -> None: + config = tmp_path / "gateway.toml" + + result = CliRunner().invoke( + app, + [ + "add-gateway-registration", + "--host-ip", + "192.0.2.10", + "--config", + str(config), + "--timeout", + "10ms", + ], + ) + + assert result.exit_code == 2 + assert "gateway timeout must be greater than 10ms" in result.stderr + assert not config.exists() + + +def test_cli_removal_forgets_registration_before_later_serve( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + calls: list[str] = [] + + class FakeServer: + def __init__( + self, + registry: GateRegistry, + *, + timeout_middleware_processing: float, + ) -> None: + del registry, timeout_middleware_processing + + def serve_sync(self, listen: str) -> None: + calls.append(listen) + + monkeypatch.setattr("egress_gate.service.server.EgressGateServer", FakeServer) + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "user-config")) + config = tmp_path / "gateway.toml" + add_result = CliRunner().invoke( + app, + [ + "add-gateway-registration", + "--host-ip", + "192.0.2.10", + "--config", + str(config), + ], + ) + assert add_result.exit_code == 0, add_result.output + + remove_result = CliRunner().invoke( + app, + [ + "remove-gateway-registration", + "--name", + "egress-gate", + "--config", + str(config), + ], + ) + assert remove_result.exit_code == 0, remove_result.output + + serve_result = CliRunner().invoke(app, ["serve"]) + + assert serve_result.exit_code == 0, serve_result.output + assert calls == ["127.0.0.1:50051"] + + def test_cli_lists_gateway_registration_names_for_removal(tmp_path: Path) -> None: config = tmp_path / "gateway.toml" config.write_text( @@ -418,7 +578,8 @@ def test_cli_lists_gateway_registration_names_for_removal(tmp_path: Path) -> Non "version = 1\n\n" "[[openshell.supervisor.middleware]]\n" 'name = "eg-regex"\n' - 'grpc_endpoint = "http://192.0.2.10:50051"\n\n' + 'grpc_endpoint = "http://192.0.2.10:50051"\n' + 'timeout = "30s"\n\n' "[[openshell.supervisor.middleware]]\n" 'name = "other-service"\n' 'grpc_endpoint = "http://192.0.2.20:9000"\n' @@ -433,6 +594,8 @@ def test_cli_lists_gateway_registration_names_for_removal(tmp_path: Path) -> Non assert "OpenShell middleware registrations" in result.stdout assert "eg-regex" in result.stdout assert "http://192.0.2.10:50051" in result.stdout + assert "Gateway RPC ceiling" in result.stdout + assert "30s" in result.stdout assert "other-service" in result.stdout assert "remove-gateway-registration --name NAME" in result.stdout @@ -522,13 +685,13 @@ def test_cli_evaluate_explains_an_invalid_timeout() -> None: str(project_dir / "examples/regex-redaction/egress-gate-config.yaml"), "--cases", str(project_dir / "examples/regex-redaction/cases.yaml"), - "--timeout-seconds", - "0", + "--timeout", + "9ms", ], color=True, ) assert result.exit_code == 2 error_output = Text.from_ansi(result.stderr).plain - assert "Invalid value for --timeout-seconds" in error_output - assert "greater than 0" in error_output + assert "Invalid value for --timeout" in error_output + assert "at least 10ms" in error_output diff --git a/projects/egress-gate/tests/test_gateway_config.py b/projects/egress-gate/tests/test_gateway_config.py index a064fee0..eed27f22 100644 --- a/projects/egress-gate/tests/test_gateway_config.py +++ b/projects/egress-gate/tests/test_gateway_config.py @@ -14,7 +14,11 @@ GatewayConfigUpdate, GatewayMiddlewareRegistration, default_gateway_config_path, + default_registration_state_path, list_gateway_registrations, + load_remembered_gateway_registration, + read_remembered_gateway_timeout, + remember_gateway_registration, remove_gateway_config, update_gateway_config, validate_middleware_name, @@ -56,6 +60,45 @@ def test_default_gateway_config_path_honors_openshell_override( assert default_gateway_config_path() == configured_path +def test_remembered_registration_reads_current_gateway_timeout( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "user-config")) + gateway_config = tmp_path / "gateway.toml" + gateway_config.write_text( + "[openshell]\n" + "version = 1\n\n" + "[[openshell.supervisor.middleware]]\n" + 'name = "egress-gate"\n' + 'grpc_endpoint = "http://192.0.2.10:50051"\n' + 'timeout = "2500ms"\n' + ) + + remember_gateway_registration( + gateway_config, + middleware_name="egress-gate", + ) + + remembered = load_remembered_gateway_registration() + assert remembered is not None + assert remembered.config_path == gateway_config.resolve() + assert remembered.middleware_name == "egress-gate" + assert default_registration_state_path().stat().st_mode & 0o777 == 0o600 + assert read_remembered_gateway_timeout() == (remembered, 2.5) + assert read_remembered_gateway_timeout(unit="ms") == (remembered, 2500.0) + + +def test_remembered_registration_is_optional( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + + assert load_remembered_gateway_registration() is None + assert read_remembered_gateway_timeout() is None + + def test_middleware_name_validation_matches_openshell_constraints() -> None: assert MAX_MIDDLEWARE_REGISTRATION_NAME_BYTES == 19 longest_name = "a" * MAX_MIDDLEWARE_REGISTRATION_NAME_BYTES @@ -96,7 +139,7 @@ def test_update_gateway_config_creates_minimal_default_config( "name": "egress-gate", "grpc_endpoint": "http://192.168.1.20:50051", "max_body_bytes": 4_194_304, - "timeout": "5s", + "timeout": "30s", } ] }, @@ -113,7 +156,8 @@ def test_list_gateway_registrations_returns_names_and_endpoints( "version = 1\n\n" "[[openshell.supervisor.middleware]]\n" 'name = "eg-regex"\n' - 'grpc_endpoint = "http://10.0.0.3:50051"\n\n' + 'grpc_endpoint = "http://10.0.0.3:50051"\n' + 'timeout = "30s"\n\n' "[[openshell.supervisor.middleware]]\n" 'name = "other-service"\n' ) @@ -122,8 +166,13 @@ def test_list_gateway_registrations_returns_names_and_endpoints( GatewayMiddlewareRegistration( name="eg-regex", endpoint="http://10.0.0.3:50051", + timeout_gateway_ceiling="30s", + ), + GatewayMiddlewareRegistration( + name="other-service", + endpoint=None, + timeout_gateway_ceiling=None, ), - GatewayMiddlewareRegistration(name="other-service", endpoint=None), ) @@ -182,6 +231,7 @@ def test_update_gateway_config_updates_only_the_named_registration( middleware_name="egress-gate", host_ip="10.0.0.4", port=50053, + timeout_gateway_ceiling="45s", ) assert result is GatewayConfigUpdate.UPDATED @@ -190,18 +240,58 @@ def test_update_gateway_config_updates_only_the_named_registration( assert "# Keep this registration comment." in contents assert 'grpc_endpoint = "http://10.0.0.4:50053"' in contents assert "max_body_bytes = 4194304" in contents - assert 'timeout = "5s"' in contents + assert 'timeout = "45s"' in contents repeated = update_gateway_config( path, middleware_name="egress-gate", host_ip="10.0.0.4", port=50053, + timeout_gateway_ceiling="45s", ) assert repeated is GatewayConfigUpdate.UNCHANGED +@pytest.mark.parametrize("timeout", ["1m", "10ms"]) +def test_update_gateway_config_rejects_invalid_timeout( + tmp_path: Path, + timeout: str, +) -> None: + with pytest.raises(GatewayConfigError, match="gateway timeout"): + update_gateway_config( + tmp_path / "gateway.toml", + middleware_name="egress-gate", + host_ip="10.0.0.4", + port=50053, + timeout_gateway_ceiling=timeout, + ) + + +def test_update_gateway_config_adds_the_operator_timeout_ceiling_when_missing( + tmp_path: Path, +) -> None: + path = tmp_path / "gateway.toml" + path.write_text( + "[openshell]\n" + "version = 1\n\n" + "[[openshell.supervisor.middleware]]\n" + 'name = "egress-gate"\n' + 'grpc_endpoint = "http://10.0.0.3:50051"\n' + "max_body_bytes = 4194304\n" + ) + + result = update_gateway_config( + path, + middleware_name="egress-gate", + host_ip="10.0.0.3", + port=50051, + ) + + assert result is GatewayConfigUpdate.UPDATED + assert 'timeout = "30s"' in path.read_text() + + @pytest.mark.parametrize( "contents", [ diff --git a/projects/egress-gate/tests/test_timeout.py b/projects/egress-gate/tests/test_timeout.py index b6c7957f..3c248507 100644 --- a/projects/egress-gate/tests/test_timeout.py +++ b/projects/egress-gate/tests/test_timeout.py @@ -5,20 +5,64 @@ import pytest from egress_gate.errors import TimeoutExpiredError -from egress_gate.timeout import Timeout +from egress_gate.timeout import ( + Timeout, + format_timeout_middleware_processing, + parse_duration, + parse_timeout_duration, +) -@pytest.mark.parametrize("seconds", [True, 0, -1, float("inf"), 31]) -def test_timeout_duration_is_strict_positive_and_bounded( +@pytest.mark.parametrize( + "seconds", + [True, 0, -1, 0.001, 0.0101, 1.0001, float("inf")], +) +def test_timeout_duration_has_minimum_and_whole_milliseconds( seconds: bool | int | float, ) -> None: with pytest.raises( ValueError, - match="finite number greater than 0 and at most 30", + match="at least 10ms, using whole milliseconds", ): Timeout.from_seconds(seconds) +@pytest.mark.parametrize( + ("seconds", "duration"), + [(0.01, "10ms"), (1.0, "1s"), (4.5, "4500ms"), (45.0, "45s")], +) +def test_middleware_timeout_uses_the_contract_duration_format( + seconds: float, + duration: str, +) -> None: + assert format_timeout_middleware_processing(seconds) == duration + + +@pytest.mark.parametrize( + ("duration", "seconds"), + [("10ms", 0.01), ("500ms", 0.5), ("1s", 1.0), ("45s", 45.0)], +) +def test_timeout_duration_parser_accepts_concise_cli_values( + duration: str, + seconds: float, +) -> None: + assert parse_timeout_duration(duration) == seconds + + +@pytest.mark.parametrize( + "duration", + ["", "1", "1.5s", "0s", "9ms", "1m", "1000000000s"], +) +def test_timeout_duration_parser_rejects_unsupported_values(duration: str) -> None: + with pytest.raises(ValueError, match="timeout"): + parse_timeout_duration(duration) + + +def test_shared_duration_parser_supports_gateway_values_and_units() -> None: + assert parse_duration("1ms") == 0.001 + assert parse_duration("45s", unit="ms") == 45_000.0 + + def test_expired_timeout_raises_typed_signal() -> None: timeout = Timeout(deadline=monotonic() - 1) @@ -27,8 +71,8 @@ def test_expired_timeout_raises_typed_signal() -> None: assert str(captured.value) == ( "Egress Gate processing timed out. Reduce the request size or simplify " - "the configured gates and rules, or increase the processing timeout " - "to at most 30 seconds, then retry." + "the configured gates and rules, or increase the middleware processing " + "timeout, then retry." )