Service

Our own service runs this package on our own machines and speaks the session protocol over WebSocket and HTTP, with the reference decision service as an HTTP API beside it. Open this page when you change the service or embed its application in another ASGI server; to install and run it, read Run our own service.

Everything below needs the service extra (pip install 'grounded-decision[service]'), except the configuration, which imports only the session layer.

service

Our own service: the package and its protocol over HTTP and WebSocket.

grounded-decision serve runs it (make serve locally); it runs on our own machines behind our own nginx, which proxies /api/ and /ws/ beside the static site (docs/how-to/run-the-service.md). It speaks the session protocol of :mod:grounded_decision.session with the same :class:~grounded_decision.session.protocol.Endpoint, one session per WebSocket, so a front end switches between the browser's worker and the service by one setting; it also serves the reference decision service (examples/reference_service.py) as an HTTP API. It needs the service extra (pip install 'grounded-decision[service]'); nothing else in the package imports it.

Modules and what they own:

:mod:~grounded_decision.service.config :class:~.config.ServiceConfig and :class:~.config.ServiceLimits: every setting, from a flag, the environment or a default. :mod:~grounded_decision.service.limits :class:~.limits.Refusal, :class:~.limits.WorkLimits, :class:~.limits.RateLimiter, :class:~.limits.SessionSlots: the limits, enforced as protocol errors. :mod:~grounded_decision.service.connection :class:~.connection.ProtocolConnection: one client's session; :class:~.connection.PythonThreads: where requests run, with a deadline. :mod:~grounded_decision.service.app :class:~.app.ServiceApp: the ASGI application and its routes. :mod:~grounded_decision.service.logs JSON log lines on standard output, with no visitor data. :mod:~grounded_decision.service.server :class:~.server.ServiceServer (uvicorn, configured) and :class:~.server.BackgroundService (on a thread, for tests).

ConfigError

Bases: ValueError

A setting has a value the service cannot use; the message names it.

ServiceConfig dataclass

ServiceConfig(
    host: str = "127.0.0.1",
    port: int = 8700,
    database: Path = (
        lambda: (
            Path(gettempdir())
            / "grounded-decision-service.sqlite3"
        )
    )(),
    seed_demo_data: bool = False,
    shadow: bool = False,
    threads: int = 4,
    log_level: str = "info",
    limits: ServiceLimits = ServiceLimits(),
)

Everything the service is started with.

Attributes:

Name	Type	Description
host	str	The interface to listen on (127.0.0.1: only nginx on the same machine reaches it).
port	int	The port.
database	Path	The decision API's SQLite file (its tables: state, audit, reviews, outbox, keys, the evidence tables).
seed_demo_data	bool	Load the reference service's demo fraud ring into the database at start-up.
shadow	bool	Run the decision API in shadow mode (record, execute nothing).
threads	int	Threads running Python for the requests (the event loop stays free to read and write sockets meanwhile).
log_level	str	debug, info, warning or error.
limits	ServiceLimits	What one client may ask.

add_arguments classmethod

add_arguments(parser: ArgumentParser) -> None

Add one flag per setting to parser (each defaults to unset).

Source code in src/grounded_decision/service/config.py

@classmethod
def add_arguments(cls, parser: argparse.ArgumentParser) -> None:
    """Add one flag per setting to ``parser`` (each defaults to unset)."""

    defaults = cls()
    for setting in SETTINGS:
        owner: object = defaults.limits if setting.group == "limits" else defaults
        setting.add_to(parser, getattr(owner, setting.name))

load classmethod

load(
    arguments: Namespace | None,
    environment: Mapping[str, str],
) -> ServiceConfig

The configuration from flags, then the environment, then the defaults.

Raises:

Type	Description
ConfigError	A value cannot be read or is out of range.

Source code in src/grounded_decision/service/config.py

@classmethod
def load(
    cls, arguments: argparse.Namespace | None, environment: Mapping[str, str]
) -> ServiceConfig:
    """The configuration from flags, then the environment, then the defaults.

    Raises:
        ConfigError: A value cannot be read or is out of range.
    """

    values: dict[str, dict[str, Any]] = {"service": {}, "limits": {}}
    for setting in SETTINGS:
        value = setting.read(arguments, environment)
        if value is not None:
            values[setting.group][setting.name] = value
    try:
        limits = ServiceLimits(**values["limits"])
        return cls(**values["service"], limits=limits)
    except TypeError as error:  # pragma: no cover - a setting without a field
        raise ConfigError(str(error)) from error

with_limits

with_limits(**changes: Any) -> ServiceConfig

This configuration with some limits changed (tests and the load test).

Source code in src/grounded_decision/service/config.py

def with_limits(self, **changes: Any) -> ServiceConfig:
    """This configuration with some limits changed (tests and the load test)."""

    return replace(self, limits=replace(self.limits, **changes))

summary

summary() -> dict[str, object]

The settings, for the start-up log line (no secrets: there are none).

Source code in src/grounded_decision/service/config.py

def summary(self) -> dict[str, object]:
    """The settings, for the start-up log line (no secrets: there are none)."""

    service = {
        item.name: str(getattr(self, item.name))
        if item.name == "database"
        else getattr(self, item.name)
        for item in fields(self)
        if item.name != "limits"
    }
    limits = {
        item.name: getattr(self.limits, item.name) for item in fields(self.limits)
    }
    return {**service, "limits": limits}

ServiceLimits dataclass

ServiceLimits(
    max_message_bytes: int = 1024 * 1024,
    max_sessions: int = 64,
    messages_per_second: float = 100.0,
    burst: int = 200,
    request_timeout: float = 30.0,
    idle_timeout: float = 600.0,
    max_steps: int = MAX_STEPS_PER_REQUEST,
    max_audit_moves: int = 1000,
    max_compare_games: int = 20,
    max_compare_seeds: int = 100,
)

What one client may ask of the service.

Attributes:

Name	Type	Description
max_message_bytes	int	The largest request, in bytes of UTF-8. A larger one is answered with a message error (HTTP 413).
max_sessions	int	Open WebSocket sessions and HTTP requests in flight in this process; past it a client gets a session error ("at capacity") and the socket closes with 1013.
messages_per_second	float	Each connection's sustained rate. A message over it is answered with a session error, not run.
burst	int	How many messages a connection may send at once before the rate applies.
request_timeout	float	Seconds one request may take (queueing included). Past it the client gets a session error and the connection closes, since its session is no longer known.
idle_timeout	float	Seconds a WebSocket may stay silent before it is closed.
max_steps	int	The most moves one step may ask for.
max_audit_moves	int	The most moves one audit may play.
max_compare_games	int	The most games one compare may play.
max_compare_seeds	int	The most seeds a comparison may have.

frame_cap_bytes property

frame_cap_bytes: int

The WebSocket frame size past which the server drops the connection.

Above :attr:max_message_bytes, so a request a little too big still gets its protocol error; only a far larger one is cut off.

config

What the service is told at start-up: one setting each, from a flag, the environment or a default.

Every setting has one name, from which its command-line flag (--max-sessions) and its environment variable (GD_SERVICE_MAX_SESSIONS) are derived; a flag beats the environment, which beats the default. docs/how-to/run-the-service.md lists them.

Concepts and the classes that own them:

:class:ServiceLimits What one client may ask of the service: message size, sessions, rate, time and the size of the work in one request. :class:ServiceConfig Where it listens, where it keeps the decision API's data, how many threads run Python, and its limits. :class:Setting One setting's name, kind, help, flag and environment variable.

ENVIRONMENT_PREFIX module-attribute

ENVIRONMENT_PREFIX = 'GD_SERVICE_'

Every environment variable the service reads starts with this.

SETTINGS module-attribute

SETTINGS: tuple[Setting, ...] = (
    Setting("host", "text", "interface to listen on"),
    Setting("port", "int", "port to listen on"),
    Setting(
        "database",
        "path",
        "SQLite file of the decision API",
    ),
    Setting(
        "seed_demo_data",
        "flag",
        "load the demo fraud ring into the database",
    ),
    Setting(
        "shadow",
        "flag",
        "decision API in shadow mode: record, execute nothing",
    ),
    Setting("threads", "int", "threads running Python"),
    Setting(
        "log_level", "text", "debug, info, warning or error"
    ),
    Setting(
        "max_message_bytes",
        "int",
        "largest request in bytes",
        "limits",
    ),
    Setting(
        "max_sessions",
        "int",
        "open sessions in this process",
        "limits",
    ),
    Setting(
        "messages_per_second",
        "float",
        "each connection's rate",
        "limits",
    ),
    Setting(
        "burst",
        "int",
        "messages a connection may send at once",
        "limits",
    ),
    Setting(
        "request_timeout",
        "float",
        "seconds one request may take",
        "limits",
    ),
    Setting(
        "idle_timeout",
        "float",
        "seconds a socket may stay silent",
        "limits",
    ),
    Setting(
        "max_steps",
        "int",
        "most moves one step request plays",
        "limits",
    ),
    Setting(
        "max_audit_moves",
        "int",
        "most moves one audit request plays",
        "limits",
    ),
    Setting(
        "max_compare_games",
        "int",
        "most games one compare request plays",
        "limits",
    ),
    Setting(
        "max_compare_seeds",
        "int",
        "most seeds one comparison has",
        "limits",
    ),
)

Every setting, in --help order.

ConfigError

Bases: ValueError

A setting has a value the service cannot use; the message names it.

ServiceLimits dataclass

ServiceLimits(
    max_message_bytes: int = 1024 * 1024,
    max_sessions: int = 64,
    messages_per_second: float = 100.0,
    burst: int = 200,
    request_timeout: float = 30.0,
    idle_timeout: float = 600.0,
    max_steps: int = MAX_STEPS_PER_REQUEST,
    max_audit_moves: int = 1000,
    max_compare_games: int = 20,
    max_compare_seeds: int = 100,
)

What one client may ask of the service.

Attributes:

Name	Type	Description
max_message_bytes	int	The largest request, in bytes of UTF-8. A larger one is answered with a message error (HTTP 413).
max_sessions	int	Open WebSocket sessions and HTTP requests in flight in this process; past it a client gets a session error ("at capacity") and the socket closes with 1013.
messages_per_second	float	Each connection's sustained rate. A message over it is answered with a session error, not run.
burst	int	How many messages a connection may send at once before the rate applies.
request_timeout	float	Seconds one request may take (queueing included). Past it the client gets a session error and the connection closes, since its session is no longer known.
idle_timeout	float	Seconds a WebSocket may stay silent before it is closed.
max_steps	int	The most moves one step may ask for.
max_audit_moves	int	The most moves one audit may play.
max_compare_games	int	The most games one compare may play.
max_compare_seeds	int	The most seeds a comparison may have.

frame_cap_bytes property

frame_cap_bytes: int

The WebSocket frame size past which the server drops the connection.

Above :attr:max_message_bytes, so a request a little too big still gets its protocol error; only a far larger one is cut off.

ServiceConfig dataclass

ServiceConfig(
    host: str = "127.0.0.1",
    port: int = 8700,
    database: Path = (
        lambda: (
            Path(gettempdir())
            / "grounded-decision-service.sqlite3"
        )
    )(),
    seed_demo_data: bool = False,
    shadow: bool = False,
    threads: int = 4,
    log_level: str = "info",
    limits: ServiceLimits = ServiceLimits(),
)

Everything the service is started with.

Attributes:

Name	Type	Description
host	str	The interface to listen on (127.0.0.1: only nginx on the same machine reaches it).
port	int	The port.
database	Path	The decision API's SQLite file (its tables: state, audit, reviews, outbox, keys, the evidence tables).
seed_demo_data	bool	Load the reference service's demo fraud ring into the database at start-up.
shadow	bool	Run the decision API in shadow mode (record, execute nothing).
threads	int	Threads running Python for the requests (the event loop stays free to read and write sockets meanwhile).
log_level	str	debug, info, warning or error.
limits	ServiceLimits	What one client may ask.

add_arguments classmethod

add_arguments(parser: ArgumentParser) -> None

Add one flag per setting to parser (each defaults to unset).

Source code in src/grounded_decision/service/config.py

@classmethod
def add_arguments(cls, parser: argparse.ArgumentParser) -> None:
    """Add one flag per setting to ``parser`` (each defaults to unset)."""

    defaults = cls()
    for setting in SETTINGS:
        owner: object = defaults.limits if setting.group == "limits" else defaults
        setting.add_to(parser, getattr(owner, setting.name))

load classmethod

load(
    arguments: Namespace | None,
    environment: Mapping[str, str],
) -> ServiceConfig

The configuration from flags, then the environment, then the defaults.

Raises:

Type	Description
ConfigError	A value cannot be read or is out of range.

Source code in src/grounded_decision/service/config.py

@classmethod
def load(
    cls, arguments: argparse.Namespace | None, environment: Mapping[str, str]
) -> ServiceConfig:
    """The configuration from flags, then the environment, then the defaults.

    Raises:
        ConfigError: A value cannot be read or is out of range.
    """

    values: dict[str, dict[str, Any]] = {"service": {}, "limits": {}}
    for setting in SETTINGS:
        value = setting.read(arguments, environment)
        if value is not None:
            values[setting.group][setting.name] = value
    try:
        limits = ServiceLimits(**values["limits"])
        return cls(**values["service"], limits=limits)
    except TypeError as error:  # pragma: no cover - a setting without a field
        raise ConfigError(str(error)) from error

with_limits

with_limits(**changes: Any) -> ServiceConfig

This configuration with some limits changed (tests and the load test).

Source code in src/grounded_decision/service/config.py

def with_limits(self, **changes: Any) -> ServiceConfig:
    """This configuration with some limits changed (tests and the load test)."""

    return replace(self, limits=replace(self.limits, **changes))

summary

summary() -> dict[str, object]

The settings, for the start-up log line (no secrets: there are none).

Source code in src/grounded_decision/service/config.py

def summary(self) -> dict[str, object]:
    """The settings, for the start-up log line (no secrets: there are none)."""

    service = {
        item.name: str(getattr(self, item.name))
        if item.name == "database"
        else getattr(self, item.name)
        for item in fields(self)
        if item.name != "limits"
    }
    limits = {
        item.name: getattr(self.limits, item.name) for item in fields(self.limits)
    }
    return {**service, "limits": limits}

Setting dataclass

Setting(
    name: str,
    kind: SettingKind,
    help: str,
    group: Literal["service", "limits"] = "service",
)

One setting: its field, kind and help; its flag and variable follow from its name.

Attributes:

Name	Type	Description
name	str	The field of :class:ServiceConfig or :class:ServiceLimits.
kind	SettingKind	How its text is read.
help	str	One line for --help.
group	Literal['service', 'limits']	service or limits: which class has the field.

read

read(
    arguments: Namespace | None,
    environment: Mapping[str, str],
) -> object | None

The value from the flag, else the environment, else None (the default).

Source code in src/grounded_decision/service/config.py

def read(
    self, arguments: argparse.Namespace | None, environment: Mapping[str, str]
) -> object | None:
    """The value from the flag, else the environment, else ``None`` (the default)."""

    given: object = (
        None if arguments is None else getattr(arguments, self.name, None)
    )
    if given is None:
        text = environment.get(self.variable)
        if text is None:
            return None
        return self.parse(text, self.variable)
    if isinstance(given, bool):
        return given
    return self.parse(str(given), self.flag)

limits

The service's limits, enforced as protocol errors.

The service answers every request it will not run with an error reply of the protocol (docs/reference/protocol.md), the same shape the engine itself sends, so a client handles one kind of error:

======================== =========== ================================== Limit Code Owner ======================== =========== ================================== Message too large message :class:Refusal (too_large) Too much work in one message :class:WorkLimits request (count, or spec moves, games, seeds) Too many messages session :class:RateLimiter Too many sessions session :class:SessionSlots Request took too long session :class:Refusal (timeout) ======================== =========== ==================================

session is the protocol's "not now": the request may succeed later.

Refusal dataclass

Refusal(
    limit: str,
    code: ErrorCode,
    message: str,
    status: HTTPStatus = OK,
)

A request the service will not run, and why.

Attributes:

Name	Type	Description
limit	str	Which limit (too_large, work, rate, capacity, timeout, binary), for the logs.
code	ErrorCode	The protocol error code.
message	str	One sentence for the person, naming the limit.
status	HTTPStatus	The HTTP status it is sent with over POST /api/ (a request asking for too much work is answered like any engine error, with 200).

reply

reply(identifier: int | None) -> str

The protocol error reply, as JSON text.

Source code in src/grounded_decision/service/limits.py

def reply(self, identifier: int | None) -> str:
    """The protocol ``error`` reply, as JSON text."""

    encoded = _ENCODER.encode(ErrorReply(identifier, self.code, self.message))
    return reply_text(encoded)

WorkLimits dataclass

WorkLimits(limits: ServiceLimits)

How much work one request may ask for, below the protocol's own maxima.

Checked on the decoded message before the engine sees it; a value that is not a whole number is left for the engine's own validation, so its message stays the engine's.

Attributes:

Name	Type	Description
limits	ServiceLimits	The service's limits.

check

check(message: JsonValue) -> Refusal | None

The refusal for a request that asks for too much, or None.

Source code in src/grounded_decision/service/limits.py

def check(self, message: JsonValue) -> Refusal | None:
    """The refusal for a request that asks for too much, or ``None``."""

    if not isinstance(message, dict):
        return None
    limits = self.limits
    kind = message.get("type")
    if kind == "step":
        return self._most(message.get("count"), limits.max_steps, "count")
    if kind == "audit":
        return self._most(message.get("moves"), limits.max_audit_moves, "moves")
    if kind == "compare":
        games = message.get("games")
        if isinstance(games, list) and len(games) > limits.max_compare_games:
            return Refusal(
                "work",
                "message",
                f"at most {limits.max_compare_games} games per request on this service",
            )
        spec = message.get("spec")
        seeds = spec.get("seeds") if isinstance(spec, dict) else None
        if _whole(seeds) and seeds > limits.max_compare_seeds:
            return Refusal(
                "work",
                "spec",
                f"--seeds must be 1 to {limits.max_compare_seeds} on this service",
            )
    return None

RateLimiter dataclass

RateLimiter(
    rate: float,
    burst: int,
    clock: Callable[[], float] = monotonic,
)

A token bucket: burst messages at once, then rate a second.

Attributes:

Name	Type	Description
rate	float	Tokens added per second.
burst	int	The bucket's size (and how full it starts).
clock	Callable[[], float]	Seconds, monotonic (injected by tests).

allow

allow() -> bool

Takes a token if there is one; False means the message is over the rate.

Source code in src/grounded_decision/service/limits.py

def allow(self) -> bool:
    """Takes a token if there is one; ``False`` means the message is over the rate."""

    now = self.clock()
    self._tokens = min(
        float(self.burst), self._tokens + (now - self._updated) * self.rate
    )
    self._updated = now
    if self._tokens >= 1.0:
        self._tokens -= 1.0
        return True
    return False

SessionSlots dataclass

SessionSlots(capacity: int, _used: int = 0)

How many sessions (WebSockets, HTTP requests in flight) the process holds.

Used only from the event loop, so it needs no lock.

Attributes:

Name	Type	Description
capacity	int	The most at once.

acquire

acquire() -> bool

Takes a slot; False when the process is at capacity.

Source code in src/grounded_decision/service/limits.py

def acquire(self) -> bool:
    """Takes a slot; ``False`` when the process is at capacity."""

    if self._used >= self.capacity:
        return False
    self._used += 1
    return True

reply_text

reply_text(reply: JsonValue) -> str

A reply as the protocol's compact JSON text (what Endpoint.handle_text writes).

Source code in src/grounded_decision/service/limits.py

def reply_text(reply: JsonValue) -> str:
    """A reply as the protocol's compact JSON text (what ``Endpoint.handle_text`` writes)."""

    return json.dumps(reply, separators=(",", ":"), ensure_ascii=False, allow_nan=False)

request_id

request_id(message: JsonValue) -> int | None

The request's id, as the engine reads it (an integer, never a boolean).

Source code in src/grounded_decision/service/limits.py

def request_id(message: JsonValue) -> int | None:
    """The request's ``id``, as the engine reads it (an integer, never a boolean)."""

    identifier = message.get("id") if isinstance(message, dict) else None
    if isinstance(identifier, int) and not isinstance(identifier, bool):
        return identifier
    return None

connection

One client's session over the service, and the threads that run its Python.

A WebSocket is one :class:ProtocolConnection: its own :class:~grounded_decision.session.protocol.Endpoint (so its own game, comparison and audit), its own rate, and the service's limits, checked before the engine sees a message. An HTTP POST /api/ is a connection for one message. Requests run on :class:PythonThreads, so the event loop keeps reading and writing every other socket meanwhile, and each has a deadline.

The reply to a request the service runs is exactly what Endpoint.handle_text returns for the same text (the golden tests compare them byte for byte); a request it does not run gets a :class:~grounded_decision.service.limits.Refusal.

PythonThreads

PythonThreads(count: int)

The threads that run requests; the event loop awaits each with a deadline.

A request past its deadline is abandoned, not stopped (Python cannot stop a thread): the bounded work per request (:class:WorkLimits) bounds how long it keeps its thread, and its session is dropped.

Source code in src/grounded_decision/service/connection.py

def __init__(self, count: int) -> None:
    self._pool = ThreadPoolExecutor(count, thread_name_prefix="gd-service")

run async

run(work: Callable[[], T], timeout: float) -> T

work() on a thread, within timeout seconds.

Raises:

Type	Description
TimeoutError	It did not finish in time (if it had not started, it never will).

Source code in src/grounded_decision/service/connection.py

async def run[T](self, work: Callable[[], T], timeout: float) -> T:
    """``work()`` on a thread, within ``timeout`` seconds.

    Raises:
        TimeoutError: It did not finish in time (if it had not
            started, it never will).
    """

    loop = asyncio.get_running_loop()
    return await asyncio.wait_for(loop.run_in_executor(self._pool, work), timeout)

close

close() -> None

Stops taking work; work not yet started is dropped.

Source code in src/grounded_decision/service/connection.py

def close(self) -> None:
    """Stops taking work; work not yet started is dropped."""

    self._pool.shutdown(wait=False, cancel_futures=True)

Answer dataclass

Answer(
    text: str,
    refusal: Refusal | None = None,
    close: bool = False,
)

What the service sends back for one message.

Attributes:

Name	Type	Description
text	str	The reply, as JSON text.
refusal	Refusal | None	Why the message was not run, or None.
close	bool	The connection must close after the reply (its session is lost).

ProtocolConnection dataclass

ProtocolConnection(
    threads: PythonThreads,
    limits: ServiceLimits,
    rate: RateLimiter | None = None,
    endpoint: Endpoint = Endpoint(),
    messages: int = 0,
    refused: int = 0,
)

One client's session: its endpoint, its rate and the service's limits.

Messages are answered one at a time, in order (the protocol's replies come in request order).

Attributes:

Name	Type	Description
threads	PythonThreads	Where the engine runs.
limits	ServiceLimits	The service's limits.
rate	RateLimiter | None	The connection's rate, or None for a one-message connection.
endpoint	Endpoint	The session's engine.
messages	int	Messages received.
refused	int	Messages answered with a refusal.

answer async

answer(text: str) -> Answer

The reply to one message (a protocol request as JSON text).

Source code in src/grounded_decision/service/connection.py

async def answer(self, text: str) -> Answer:
    """The reply to one message (a protocol request as JSON text)."""

    self.messages += 1
    limits = self.limits
    if len(text) > limits.max_message_bytes or (
        len(text.encode("utf-8")) > limits.max_message_bytes
    ):
        return self._refuse(Refusal.too_large(limits), None)
    try:
        message: JsonValue = json.loads(text)
    except ValueError:
        message = text  # the engine answers it as it answers handle_text
    identifier = request_id(message)
    if self.rate is not None and not self.rate.allow():
        return self._refuse(Refusal.rate(limits), identifier)
    refusal = WorkLimits(limits).check(message)
    if refusal is not None:
        return self._refuse(refusal, identifier)
    endpoint = self.endpoint
    try:
        reply = await self.threads.run(
            lambda: reply_text(endpoint.handle(message)), limits.request_timeout
        )
    except TimeoutError:
        return self._refuse(Refusal.timeout(limits), identifier, close=True)
    return Answer(reply)

refuse

refuse(refusal: Refusal) -> Answer

A refusal of a message that was never read (a binary frame).

Source code in src/grounded_decision/service/connection.py

def refuse(self, refusal: Refusal) -> Answer:
    """A refusal of a message that was never read (a binary frame)."""

    self.messages += 1
    return self._refuse(refusal, None)

app

The service as an ASGI application: its routes, over HTTP and WebSocket.

========================== ============================================== Route What it does ========================== ============================================== GET /healthz {"status": "ok", ...}: alive, versions, sessions in use POST /api/ One protocol request, one reply (a fresh session each time: describe, open, the demos) /api/decisions/... The reference decision service's HTTP API (POST /api/decisions/decide and the rest: the same paths and bodies as examples/reference_service.py) /ws/ (WebSocket) A protocol session: one per connection, every request and reply a text message ========================== ==============================================

Raw ASGI, no framework: four routes do not need one (design decision D77). :class:ServiceApp owns the routes, the session slots, the threads and the decision API; every request it will not run is answered with a protocol error (:mod:~grounded_decision.service.limits).

DECISIONS_PREFIX module-attribute

DECISIONS_PREFIX = '/api/decisions'

Where the reference decision service's paths are mounted.

TRY_AGAIN_LATER module-attribute

TRY_AGAIN_LATER = 1013

The WebSocket close code for "at capacity" and "too slow" (RFC 6455, IANA registry).

GOING_AWAY module-attribute

GOING_AWAY = 1001

The close code for an idle socket and for a server shutting down.

ServiceApp

ServiceApp(
    config: ServiceConfig,
    *,
    decisions: DecisionService | None = None,
    clock: Callable[[], float] = monotonic,
    version: str | None = None,
)

The service: health, protocol over HTTP and WebSocket, the decision API.

Stateless apart from each WebSocket's session, which lives as long as its connection. Use one per process.

Attributes:

Name	Type	Description
config		What it was started with.
slots		Sessions in use (WebSockets, and HTTP requests in flight).
threads		Where requests run.
decisions		The decision API's routes.

Source code in src/grounded_decision/service/app.py

def __init__(
    self,
    config: ServiceConfig,
    *,
    decisions: DecisionService | None = None,
    clock: Callable[[], float] = time.monotonic,
    version: str | None = None,
) -> None:
    self.config = config
    self.slots = SessionSlots(config.limits.max_sessions)
    self.threads = PythonThreads(config.threads)
    self.decisions = DecisionRoutes(
        decisions
        if decisions is not None
        else build_service(
            config.database,
            seed_demo=config.seed_demo_data,
            enforce=not config.shadow,
        )
    )
    self.log = EventLog()
    self._clock = clock
    self._connections = itertools.count(1)
    self.version = __version__ if version is None else version

close

close() -> None

Stops the threads (the server has stopped taking connections).

Source code in src/grounded_decision/service/app.py

def close(self) -> None:
    """Stops the threads (the server has stopped taking connections)."""

    self.threads.close()

health

health() -> dict[str, object]

What GET /healthz returns.

Source code in src/grounded_decision/service/app.py

def health(self) -> dict[str, object]:
    """What ``GET /healthz`` returns."""

    return {
        "status": "ok",
        "version": self.version,
        "protocol": PROTOCOL_VERSION,
        "sessions": self.slots.used,
        "capacity": self.slots.capacity,
    }

logs

Structured logs: one JSON object per line on standard output, with no visitor data.

What is logged: the service starting and stopping with its settings, each connection's end (how long, how many messages, how many refused), each refusal by limit, each HTTP request's route, status and time, and errors with their traceback. What is never logged: addresses, headers, user agents, message contents, game specs or moves. The access logs of the ASGI server (which carry the client's address) are off, its own logger is kept at warning, and the reference decision service's per-decision line (which carries the caller's entity ids) is kept at warning too.

Concepts and the classes that own them:

:class:JsonFormatter A log record as one JSON line. :class:EventLog The service's events, each with its fields.

QUIET_LOGGERS module-attribute

QUIET_LOGGERS = (
    "uvicorn",
    "uvicorn.error",
    "uvicorn.access",
    "grounded_decision.reference_service",
)

Loggers kept at warning: their info lines carry addresses or callers' ids.

JsonFormatter

Bases: Formatter

{"time", "level", "logger", "event", ...fields} on one line.

The fields come from the record's fields attribute (what :class:EventLog passes); an exception adds error and traceback.

EventLog dataclass

EventLog(
    logger: Logger = (lambda: getLogger(LOGGER_NAME))(),
)

The service's events, each a message and its fields.

Attributes:

Name	Type	Description
logger	Logger	Where they go.

configure_logging

configure_logging(
    level: str, stream: TextIO | None = None
) -> None

Send every log record to stream (standard output) as JSON lines.

Source code in src/grounded_decision/service/logs.py

def configure_logging(level: str, stream: TextIO | None = None) -> None:
    """Send every log record to ``stream`` (standard output) as JSON lines."""

    handler = logging.StreamHandler(stream if stream is not None else sys.stdout)
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    for existing in list(root.handlers):
        root.removeHandler(existing)
    root.addHandler(handler)
    root.setLevel(level.upper())
    quiet_loggers()

quiet_loggers

quiet_loggers() -> None

Keeps :data:QUIET_LOGGERS at warning or above (their info lines name visitors).

Source code in src/grounded_decision/service/logs.py

def quiet_loggers() -> None:
    """Keeps :data:`QUIET_LOGGERS` at ``warning`` or above (their ``info`` lines name visitors)."""

    for name in QUIET_LOGGERS:
        logger = logging.getLogger(name)
        logger.setLevel(max(logging.WARNING, logger.getEffectiveLevel()))

server

Running the service: in the foreground (grounded-decision serve) or on a thread (tests).

The ASGI server is uvicorn with its sans-I/O websockets protocol and h11 for HTTP, configured here and nowhere else: no access log, no Server header, the client's address never read from proxy headers, WebSocket frames capped above the message limit, a short keep-alive and a graceful shutdown that lets requests in flight finish.

Concepts and the classes that own them:

:class:ServiceServer The configured server around one :class:~.app.ServiceApp. :class:BackgroundService A server on a thread, on a free port, stopped on exit (tests, the load test).

ServiceServer dataclass

ServiceServer(
    config: ServiceConfig, app: ServiceApp | None = None
)

uvicorn, configured for the service.

Attributes:

Name	Type	Description
config	ServiceConfig	The service's settings.
app	ServiceApp | None	The application it serves (made from config by default).

port property

port: int

The bound port (useful with port 0), once started.

run

run() -> None

Serves until SIGINT or SIGTERM (or :meth:stop).

Source code in src/grounded_decision/service/server.py

def run(self) -> None:
    """Serves until SIGINT or SIGTERM (or :meth:`stop`)."""

    self.server.run()

BackgroundService

BackgroundService(
    config: ServiceConfig, app: ServiceApp | None = None
)

The service on a thread, on a free port of 127.0.0.1; a context manager.

Attributes:

Name	Type	Description
config		Its settings (the port is replaced by a free one).
server		The running server.

Source code in src/grounded_decision/service/server.py

def __init__(self, config: ServiceConfig, app: ServiceApp | None = None) -> None:
    self.config = replace(config, host="127.0.0.1", port=0)
    self.server = ServiceServer(self.config, app)
    self._thread = threading.Thread(
        target=self.server.run, name="gd-service-test", daemon=True
    )

serve

serve(config: ServiceConfig) -> None

grounded-decision serve: JSON logs on stdout, then serve in the foreground.

Source code in src/grounded_decision/service/server.py

def serve(config: ServiceConfig) -> None:
    """``grounded-decision serve``: JSON logs on stdout, then serve in the foreground."""

    configure_logging(config.log_level)
    log = EventLog()
    log.info("service.start", pid=os.getpid(), **config.summary())
    ServiceServer(config).run()

load

A load test of a running service: many WebSocket sessions stepping snake at once.

Each session opens a snake game and asks for one move per request, as fast as the replies come, until it has made its requests; a game that ends is opened again. The report gives the throughput (requests and moves a second, all sessions together) and the latency of one request (median, 95th and 99th percentile, worst), measured by the client::

python -m grounded_decision.service.load --url ws://127.0.0.1:8700/ws/ --sessions 32

make service-load runs it against a service it starts; tests/test_service.py runs a short one with a generous bound.

Concepts and the classes that own them:

:class:LoadTest What to run: the URL, the sessions, the requests each, the game. :class:LoadReport What it measured.

LoadReport dataclass

LoadReport(
    sessions: int,
    requests: int,
    moves: int,
    errors: int,
    seconds: float,
    latencies_ms: tuple[float, ...],
)

What a load test measured.

Attributes:

Name	Type	Description
sessions	int	Sessions run at once.
requests	int	step requests answered, all sessions together.
moves	int	Moves those requests played.
errors	int	Requests answered with an error (other than a finished game).
seconds	float	Wall-clock time from the first open to the last reply.
latencies_ms	tuple[float, ...]	Every request's round trip, in milliseconds, sorted.

percentile

percentile(share: float) -> float

The latency below which share of the requests fall (nearest rank).

Source code in src/grounded_decision/service/load.py

def percentile(self, share: float) -> float:
    """The latency below which ``share`` of the requests fall (nearest rank)."""

    if not self.latencies_ms:
        return 0.0
    rank = max(
        0,
        min(len(self.latencies_ms) - 1, round(share * len(self.latencies_ms)) - 1),
    )
    return self.latencies_ms[rank]

lines

lines() -> list[str]

The report, for people.

Source code in src/grounded_decision/service/load.py

def lines(self) -> list[str]:
    """The report, for people."""

    median = statistics.median(self.latencies_ms) if self.latencies_ms else 0.0
    worst = self.latencies_ms[-1] if self.latencies_ms else 0.0
    sessions = (
        f"sessions {self.sessions}, requests {self.requests:,},"
        f" moves {self.moves:,}, errors {self.errors}, {self.seconds:.2f} s"
    )
    throughput = (
        f"throughput {self.requests_per_second:,.0f} requests/s,"
        f" {self.moves_per_second:,.0f} moves/s"
    )
    latency = (
        f"latency median {median:.2f} ms, p95 {self.percentile(0.95):.2f} ms,"
        f" p99 {self.percentile(0.99):.2f} ms, max {worst:.2f} ms"
    )
    return [sessions, throughput, latency]

LoadTest dataclass

LoadTest(
    url: str,
    sessions: int = 16,
    requests: int = 200,
    size: str = "12x12",
    count: int = 1,
)

Sessions stepping snake at once against a service.

Attributes:

Name	Type	Description
url	str	The service's WebSocket (ws://127.0.0.1:8700/ws/).
sessions	int	How many at once.
requests	int	step requests per session.
size	str	The snake board.
count	int	Moves per request.

run

run() -> LoadReport

Runs the test (its own event loop) and returns what it measured.

Source code in src/grounded_decision/service/load.py

def run(self) -> LoadReport:
    """Runs the test (its own event loop) and returns what it measured."""

    return asyncio.run(self._run())

wait_for_health

wait_for_health(url: str, seconds: float) -> None

Waits until the service behind WebSocket url answers GET /healthz.

Raises:

Type	Description
TimeoutError	It did not within seconds.

Source code in src/grounded_decision/service/load.py

def wait_for_health(url: str, seconds: float) -> None:
    """Waits until the service behind WebSocket ``url`` answers ``GET /healthz``.

    Raises:
        TimeoutError: It did not within ``seconds``.
    """

    parts = urlsplit(url)
    scheme = "https" if parts.scheme == "wss" else "http"
    health = f"{scheme}://{parts.netloc}/healthz"
    deadline = time.monotonic() + seconds
    while True:
        try:
            with urllib.request.urlopen(health, timeout=2):
                return
        except OSError:
            if time.monotonic() > deadline:
                raise TimeoutError(
                    f"{health} did not answer in {seconds:g} s"
                ) from None
            time.sleep(0.2)

main

main(argv: Sequence[str] | None = None) -> int

python -m grounded_decision.service.load: run a load test and print its report.

Source code in src/grounded_decision/service/load.py

def main(argv: Sequence[str] | None = None) -> int:
    """``python -m grounded_decision.service.load``: run a load test and print its report."""

    parser = argparse.ArgumentParser(
        prog="python -m grounded_decision.service.load",
        description="WebSocket sessions stepping snake at once against a running service.",
    )
    parser.add_argument(
        "--url", default="ws://127.0.0.1:8700/ws/", help="the service's /ws/"
    )
    parser.add_argument("--sessions", type=int, default=16, help="sessions at once")
    parser.add_argument(
        "--requests", type=int, default=200, help="step requests per session"
    )
    parser.add_argument("--size", default="12x12", help="the snake board")
    parser.add_argument("--json", action="store_true", help="print the report as JSON")
    parser.add_argument(
        "--wait",
        type=float,
        default=0.0,
        help="seconds to wait for the service's /healthz",
    )
    arguments = parser.parse_args(argv)
    if arguments.wait > 0:
        wait_for_health(str(arguments.url), float(arguments.wait))
    report = LoadTest(
        str(arguments.url),
        int(arguments.sessions),
        int(arguments.requests),
        str(arguments.size),
    ).run()
    if arguments.json:
        print(
            json.dumps(
                {
                    "sessions": report.sessions,
                    "requests": report.requests,
                    "moves": report.moves,
                    "errors": report.errors,
                    "seconds": round(report.seconds, 3),
                    "requests_per_second": round(report.requests_per_second, 1),
                    "p50_ms": round(statistics.median(report.latencies_ms), 3),
                    "p95_ms": round(report.percentile(0.95), 3),
                    "p99_ms": round(report.percentile(0.99), 3),
                }
            )
        )
    else:
        print("\n".join(report.lines()))
    return 0 if report.errors == 0 else 1