From 0f562ca52dea2009696d549a72f136c68d285d05 Mon Sep 17 00:00:00 2001 From: crypto-a Date: Mon, 7 Sep 2026 12:59:25 -0400 Subject: [PATCH] feat(messaging): a NATS transport, so the bus has a client the fleet shares NATS JetStream is live on compliance-engine and nothing could talk to it, because no service had a client. This is that client, as transport policy rather than as an abstraction over anything. ADMITTED ON THE kit.clients ARGUMENT, and it arrived on its first copy rather than its third, so it is worth stating what carries it. This is not an abstraction over a domain: it is the ack policy, the redelivery behaviour, the reconnection settings and the header convention, and those are wrong in the same way in every service that decides them alone. They are also wrong invisibly, and are found during an incident, with a consumer that has been quietly reprocessing or quietly skipping. WHAT IT REFUSES TO HOLD is the half that keeps that honest. The payload is opaque bytes: no schema, no event registry, no versioning, because what an event IS belongs to the services that agree on it. It creates no streams and no consumers either, since retention and replica counts are operational decisions and a library that makes them makes them once, wrongly, in production. THE FOUR HEADERS ARE THE JUDGEMENT CALL. They are in because they are how tenancy and idempotency travel: without a fixed convention two services disagree about where the tenant id lives, and a redelivered message cannot be recognised as a duplicate. The event id rides as Nats-Msg-Id rather than a name of ours, which is not cosmetic: that is the header JetStream itself deduplicates on, so publishing is idempotent inside the stream's window for free instead of leaving the whole problem to consumers. AT LEAST ONCE, SAID PLAINLY. fetch returns unsettled messages and settling is an explicit ack, nak or term. Nothing acknowledges on the caller's behalf, because an auto-acking API turns at-least-once into at-most-once at the exact moment a handler raises. A message whose headers cannot be read is terminated rather than naked: it will be malformed next time too, and a nak would spend the whole attempt budget rediscovering that. A failed settle is logged and swallowed. The work happened; only the acknowledgement did not, so the message comes back and the handler's idempotency covers it. Raising there would tell a caller its work failed, which is worse than the truth. nats-py is an OPTIONAL EXTRA, for the reason telemetry is: a dependency here is one all ten repositories inherit and cannot refuse. The import is function local, and CI's minimal-install job now asserts kit.messaging imports without it. Connecting without the extra raises a BrokerUnavailable naming the extra, which is a failure a service that forgot can actually read. Metrics label the STREAM, never the subject. kit's own metrics docstring warns that one label carrying a tenant id turns one series into millions, and a subject carries one in every useful design. The four new instruments are registered in _build, without which record() would have dropped every one of them silently. Version goes to 0.7.0, which also fixes a real bug: pyproject and kit.__version__ both said 0.6.1 at the commit tagged v0.6.2, so three services pinned to v0.6.2 logged the wrong version at boot, defeating the field's whole purpose. test_version compares the source to the installed metadata, and both derive from the same file, so it never caught it. Gate green: ruff, pyright strict, 295 tests, 100% branch coverage, and the package imports with neither extra installed. Closes #14 --- .github/workflows/ci.yml | 2 +- README.md | 15 + pyproject.toml | 8 +- src/kit/__init__.py | 2 +- src/kit/messaging/__init__.py | 88 +++++ src/kit/messaging/_broker.py | 383 ++++++++++++++++++ src/kit/messaging/_errors.py | 54 +++ src/kit/messaging/_headers.py | 114 ++++++ src/kit/messaging/_health.py | 32 ++ src/kit/observability/__init__.py | 8 + src/kit/observability/_metrics.py | 24 ++ tests/test_messaging.py | 623 ++++++++++++++++++++++++++++++ uv.lock | 17 +- 13 files changed, 1365 insertions(+), 5 deletions(-) create mode 100644 src/kit/messaging/__init__.py create mode 100644 src/kit/messaging/_broker.py create mode 100644 src/kit/messaging/_errors.py create mode 100644 src/kit/messaging/_headers.py create mode 100644 src/kit/messaging/_health.py create mode 100644 tests/test_messaging.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e8d5fdc..b396ac3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -48,7 +48,7 @@ jobs: uv venv --python 3.14 .venv-minimal VIRTUAL_ENV=.venv-minimal uv pip install . VIRTUAL_ENV=.venv-minimal uv run --no-project python -c " - import kit, kit.health, kit.httpapi, kit.config, kit.observability, kit.email + import kit, kit.health, kit.httpapi, kit.config, kit.observability, kit.email, kit.messaging print('kit', kit.__version__, 'imports without the otel extra') " diff --git a/README.md b/README.md index 67f5572..9b1eaf9 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ from kit.health import Registry | `kit.clients` | Outbound calls: deadlines, retries with jitter, a circuit breaker, trace and request-id propagation, pluggable auth | | `kit.config` | Env loading, the service prefix, the `PORT` and `OTEL_*` exceptions, fail-fast process settings and fail-soft backing systems | | `kit.email` | Outbound mail: the message every provider agrees on, and one Resend adapter on top of `kit.clients` | +| `kit.messaging` | The event bus: publish and pull-subscribe over NATS JetStream, explicit ack, nak and term, and the four headers tenancy and idempotency travel in | | `kit.testing` | The contract tests a service inherits, so the shared behaviour is verified in every repository | About a thousand lines. The most important piece is also the smallest: roughly @@ -95,6 +96,20 @@ honest is what it refuses to hold. Templates, rendered copy and which person receives which message stay in the service, because those are the parts that differ, and a package holding them would be a mail product rather than a seam. +`kit.messaging` is admitted on the same argument and is worth checking against +it just as hard, because it arrived on its FIRST copy rather than its third. It +is not an abstraction over a domain: it is the ack policy, the redelivery +behaviour, the reconnection settings and the header convention, and those go +wrong the same way in every service that decides them alone. They also go wrong +invisibly, and are found during an incident, with a consumer that has been +quietly reprocessing or quietly skipping. + +The boundary that keeps it honest is again what it refuses to hold. The payload +is opaque bytes: there is no schema, no event registry and no versioning, because +what an event IS belongs to the services that agree on it. It creates no streams +and no consumers either, since retention and replica counts are operational +decisions, and a library that makes them makes them once, wrongly, in production. + Business models, ORM models, migrations, route trees and service settings never belong here at all. diff --git a/pyproject.toml b/pyproject.toml index 8bd0afe..8972a51 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,7 @@ [project] name = "scadable-kit" -version = "0.6.1" +version = "0.7.0" description = "Cross-cutting behaviour shared by every SCADABLE service" requires-python = ">=3.14,<3.15" license = { file = "LICENSE" } @@ -45,6 +45,12 @@ otel = [ # traceparent, B starts a new trace, and one request is two unlinked traces. "opentelemetry-instrumentation-httpx>=0.65b0", ] +# The bus is an extra for the same reason telemetry is: most services never +# touch it, and a dependency here is one every service in the fleet inherits and +# cannot refuse. `kit.messaging` imports cleanly without it, and raises a +# BrokerUnavailable naming this extra if something actually tries to connect, +# which is a failure a service that forgot can read. +nats = ["nats-py>=2.15,<3"] [build-system] requires = ["hatchling"] diff --git a/src/kit/__init__.py b/src/kit/__init__.py index 1e3d33d..844a8f0 100644 --- a/src/kit/__init__.py +++ b/src/kit/__init__.py @@ -20,4 +20,4 @@ across the fleet without opening every repository. """ -__version__ = "0.6.1" +__version__ = "0.7.0" diff --git a/src/kit/messaging/__init__.py b/src/kit/messaging/__init__.py new file mode 100644 index 0000000..118bb63 --- /dev/null +++ b/src/kit/messaging/__init__.py @@ -0,0 +1,88 @@ +"""The event bus, as transport policy the whole fleet shares. + + bus = MessageBus(Broker(name="events", servers=("nats://nats.nats:4222",))) + await bus.connect() + + await bus.publish( + "events.repository.indexed", + payload, + Envelope(tenant=tenant, event_id=uid, event_type="repository.indexed", + occurred_at=now), + ) + + subscription = await bus.subscribe(stream="EVENTS", subject="events.>", + durable="brain") + for message in await subscription.fetch(batch=8): + await handle(message) + await message.ack() + +WHY THIS IS IN THE KIT AT ALL, given that "code moves in on its third copy". This +is the `kit.clients` case rather than the database-helpers case: it is not an +abstraction over a domain, it is the ack policy, the redelivery behaviour, the +reconnection settings and the header convention, and those are wrong in the same +way in every service that decides them alone. They are also wrong in a way that +only shows up during an incident, when a consumer has been quietly reprocessing +or quietly skipping. + +THE BOUNDARY THAT KEEPS IT HONEST IS WHAT IT REFUSES TO HOLD. The payload is +opaque bytes. There is no schema, no event registry and no versioning: what an +event IS belongs to the services that agree on it, and a package holding that +would be a messaging product rather than a seam. It creates no streams and no +consumers either, because retention and replica counts are operational decisions +and a library that makes them makes them once, wrongly, in production. + +WHAT IS POLICY AND WHAT IS NOT. Timeouts, batch sizes, reconnection and nak +delays are parameters and yours to change per broker. The four header names are +not: they are how tenancy and idempotency travel, and a service that renames one +stops being readable by every other service on the bus. + +AT LEAST ONCE, SAID PLAINLY. `fetch` returns unsettled messages and settling is +an explicit `ack`, `nak` or `term`. Nothing here acknowledges on your behalf, so +a handler that raises gets its message back rather than losing it. Handlers must +be idempotent; `Envelope.event_id` is what they deduplicate on, and it rides the +wire as `Nats-Msg-Id` so the server suppresses duplicate publishes too. +""" + +from __future__ import annotations + +from kit.messaging._broker import ( + DEFAULT_FETCH_TIMEOUT_SECONDS, + Broker, + Message, + MessageBus, + Subscription, +) +from kit.messaging._errors import ( + BrokerError, + BrokerTimeout, + BrokerUnavailable, + MalformedMessage, +) +from kit.messaging._headers import ( + EVENT_ID, + EVENT_TYPE, + OCCURRED_AT, + TENANT, + Envelope, + envelope, +) +from kit.messaging._health import register_brokers + +__all__ = [ + "DEFAULT_FETCH_TIMEOUT_SECONDS", + "EVENT_ID", + "EVENT_TYPE", + "OCCURRED_AT", + "TENANT", + "Broker", + "BrokerError", + "BrokerTimeout", + "BrokerUnavailable", + "Envelope", + "MalformedMessage", + "Message", + "MessageBus", + "Subscription", + "envelope", + "register_brokers", +] diff --git a/src/kit/messaging/_broker.py b/src/kit/messaging/_broker.py new file mode 100644 index 0000000..eb6116d --- /dev/null +++ b/src/kit/messaging/_broker.py @@ -0,0 +1,383 @@ +"""The bus, as one object per broker, built where everything else is built. + +WHAT THIS OWNS. Connecting, publishing with the four headers, pulling a batch, +and the three ways a message can be settled. Reconnection is delegated to the +NATS client, whose parameters are set here rather than left at their defaults. + +WHAT IT REFUSES TO OWN, and this is the important half: + + - IT CREATES NO STREAMS AND NO CONSUMERS. Creating a stream carries retention, + replica and discard decisions, and a library that creates one on connect + eventually creates the wrong one in production, silently, because the call + is idempotent and the second caller's arguments are ignored. Streams are an + operational act. + + - IT DOES NOT AUTO-ACK. `fetch` hands back messages that are still unsettled, + and settling is an explicit call. An auto-acking API turns at-least-once + into at-most-once at the moment a handler raises, which is the one moment + anybody cares. + + - IT HOLDS NO SCHEMA. The payload is opaque bytes. + +THE IMPORT OF `nats` IS FUNCTION LOCAL, always, because the dependency is an +optional extra and every service in the fleet inherits this package whether or +not it touches a bus. A module-scope import would make `import kit.messaging` +fail at startup for the services that never asked for it. +""" + +from __future__ import annotations + +import logging +from collections.abc import Sequence +from dataclasses import dataclass, field +from typing import Any + +from kit.messaging._errors import BrokerTimeout, BrokerUnavailable, MalformedMessage +from kit.messaging._headers import Envelope, envelope +from kit.observability import ( + MESSAGING_CONSUMED, + MESSAGING_DELIVERIES, + MESSAGING_PUBLISHED, + MESSAGING_SETTLED, + record, +) + +log = logging.getLogger("kit.messaging") + +DEFAULT_CONNECT_TIMEOUT_SECONDS = 5.0 +DEFAULT_RECONNECT_WAIT_SECONDS = 2.0 +DEFAULT_FETCH_TIMEOUT_SECONDS = 5.0 +DEFAULT_PUBLISH_TIMEOUT_SECONDS = 5.0 + +RECONNECT_FOREVER = -1 +"""What `max_reconnect_attempts` means by default, and it is deliberate. + +A consumer that gives up reconnecting becomes a process that is running, healthy +by any liveness probe, and consuming nothing. Retrying forever turns that into a +visible gap in throughput instead, which is the failure somebody notices. +""" + + +@dataclass(frozen=True, slots=True) +class Broker: + """One bus, and everything this service knows about reaching it.""" + + name: str + """Short and STABLE. It becomes a metric label and a readiness check name, so + changing it starts a new time series and orphans the dashboard watching the + old one.""" + + servers: tuple[str, ...] = () + credentials_path: str = "" + """A `.creds` file, when the bus requires one. + + Empty today: the deployed bus has no authentication and is reached only + through a NetworkPolicy. It is carried anyway so that turning auth on is a + config change in one service rather than a release of this package.""" + + token: str = "" + connect_timeout_seconds: float = DEFAULT_CONNECT_TIMEOUT_SECONDS + reconnect_wait_seconds: float = DEFAULT_RECONNECT_WAIT_SECONDS + max_reconnect_attempts: int = RECONNECT_FOREVER + publish_timeout_seconds: float = DEFAULT_PUBLISH_TIMEOUT_SECONDS + + def __post_init__(self) -> None: + if not self.name.strip(): + raise ValueError("a broker needs a name; it labels metrics and health checks") + + @property + def configured(self) -> bool: + """Whether an operator set this up, answered without opening a socket.""" + return bool(self.servers) + + +@dataclass(frozen=True, slots=True) +class Message: + """One delivery, still unsettled. + + `delivered` is the attempt number this delivery is, counting from one. Above + one means the message has been redelivered, which is the signal a handler + uses to decide that something is poisonous rather than merely slow. + """ + + envelope: Envelope + payload: bytes + delivered: int + stream: str + _broker: str = field(repr=False, default="") + _message: Any = field(repr=False, default=None) + + async def ack(self) -> None: + """Done. Do not send this again.""" + await self._settle("ack") + + async def nak(self, delay_seconds: float | None = None) -> None: + """Not now. Send it again, optionally after a delay. + + The delay is the back pressure knob: naking without one on a dependency + that is down produces a hot loop against it. + """ + await self._settle("nak", delay_seconds) + + async def term(self) -> None: + """Never. Stop redelivering this message. + + For a message that will fail identically every time: a malformed + payload, or work whose subject no longer exists. A nak would retry it + until the consumer's limit and then dead-letter it anyway, having spent + the attempts to learn what was already known. + """ + await self._settle("term") + + async def _settle(self, outcome: str, delay_seconds: float | None = None) -> None: + try: + if outcome == "ack": + await self._message.ack() + elif outcome == "nak": + await self._message.nak(delay=delay_seconds) + else: + await self._message.term() + except Exception as error: + # A FAILED SETTLE IS NOT A FAILED HANDLER. The work happened; only + # the acknowledgement did not, so the message will be redelivered + # and the handler's idempotency is what saves it. Raising here would + # tell a caller its work failed, which is worse than the truth. + log.warning( + "could not settle message", + extra={"broker": self._broker, "outcome": outcome, "error": str(error)}, + ) + return + record(MESSAGING_SETTLED, 1, broker=self._broker, stream=self.stream, outcome=outcome) + + +class Subscription: + """A durable consumer, pulled in batches. + + PULL RATHER THAN PUSH, because a pull consumer takes work at the rate the + process can finish it. A push consumer is handed messages at the server's + rate, and a slow handler turns that into an ack timeout and redelivery of + work that is still running. + """ + + def __init__(self, broker: str, stream: str, subscription: Any) -> None: + self.broker = broker + self.stream = stream + self._subscription = subscription + + async def fetch( + self, batch: int = 1, timeout_seconds: float = DEFAULT_FETCH_TIMEOUT_SECONDS + ) -> Sequence[Message]: + """Take up to `batch` messages, or none. + + AN EMPTY BATCH IS NOT AN ERROR and is the ordinary case on a quiet + stream. The NATS client signals it with a timeout, which is caught here + and turned into an empty sequence, so a caller's loop does not have to + treat "nothing happened" as a failure. + """ + try: + raw = await self._subscription.fetch(batch, timeout=timeout_seconds) + except TimeoutError: + return () + except Exception as error: + raise BrokerUnavailable(self.broker, f"fetch failed: {error}") from None + + messages: list[Message] = [] + for one in raw: + try: + read = envelope(dict(one.headers or {})) + except ValueError as error: + # TERMINATED, NOT NAKED. It will be malformed next time too, and + # a nak would spend the whole attempt budget rediscovering that. + await _terminate(one) + log.warning( + "terminated an unreadable message", + extra={"broker": self.broker, "stream": self.stream, "error": str(error)}, + ) + continue + delivered = _delivery_count(one) + record(MESSAGING_CONSUMED, 1, broker=self.broker, stream=self.stream) + record(MESSAGING_DELIVERIES, delivered, broker=self.broker, stream=self.stream) + messages.append( + Message( + envelope=read, + payload=one.data, + delivered=delivered, + stream=self.stream, + _broker=self.broker, + _message=one, + ) + ) + return tuple(messages) + + +class MessageBus: + """The connection, built once in a composition root and passed around. + + Never a module-level singleton: a test substitutes one by passing a + different object, not by monkeypatching this module. + """ + + def __init__(self, broker: Broker, connection: Any | None = None) -> None: + """`connection` exists for tests, and for one real case. + + A test passes a stub and exercises the real header, settle and metric + code with no server and no sleep. The real case is a service that + already holds a connection and wants this package's policy on top of it. + """ + self.broker = broker + self._connection = connection + self._stream: Any = None if connection is None else connection.jetstream() + + @property + def connected(self) -> bool: + """What the client believes, without asking the network. + + A readiness check reads this. It does not ping the bus, for the reason + `kit.clients` does not call its upstreams: a probe that reaches out turns + one kubelet's polling interval into load on a shared dependency, from + every replica, forever. + """ + return self._connection is not None and bool(self._connection.is_connected) + + async def connect(self) -> None: + """Open the connection, or say why not. + + Idempotent: calling it twice is a no-op rather than a second connection, + because a composition root that retries should not leak one. + """ + if self._connection is not None: + return + if not self.broker.configured: + raise BrokerUnavailable(self.broker.name, "no servers configured") + + try: + import nats as _nats + except ImportError: + raise BrokerUnavailable( + self.broker.name, + "the nats extra is not installed; add scadable-kit[nats]", + ) from None + + options: dict[str, Any] = { + "servers": list(self.broker.servers), + "connect_timeout": self.broker.connect_timeout_seconds, + "reconnect_time_wait": self.broker.reconnect_wait_seconds, + "max_reconnect_attempts": self.broker.max_reconnect_attempts, + } + if self.broker.credentials_path: + options["user_credentials"] = self.broker.credentials_path + if self.broker.token: + options["token"] = self.broker.token + + # ERASED AT THE BOUNDARY, the way `_metrics` erases the OpenTelemetry + # SDK. `nats-py` ships no type information, so under pyright's strict + # mode every call through it is a partially-unknown type. Narrowing here + # keeps the strictness on OUR code, where it is worth having, rather + # than turning the setting off for the package. + nats: Any = _nats + + try: + self._connection = await nats.connect(**options) + except TimeoutError: + raise BrokerTimeout(self.broker.name, "connect timed out") from None + except Exception as error: + raise BrokerUnavailable(self.broker.name, f"connect failed: {error}") from None + self._stream = self._connection.jetstream() + + async def publish(self, subject: str, payload: bytes, message: Envelope) -> None: + """Publish, and wait for the stream to say it stored it. + + WAITING IS THE POINT. JetStream's publish returns an acknowledgement + that the message is persisted to the stream's replicas; not waiting for + it makes this a fire and forget call that returns success while the + message is still in flight, which is indistinguishable from working + right up until a leader election eats one. + + Deduplication comes free: the envelope's id rides as `Nats-Msg-Id`, and + the stream refuses a second copy inside its duplicate window. + """ + stream = self._require_stream() + try: + await stream.publish( + subject, + payload, + headers=message.headers(), + timeout=self.broker.publish_timeout_seconds, + ) + except TimeoutError: + raise BrokerTimeout(self.broker.name, "publish timed out") from None + except Exception as error: + raise BrokerUnavailable(self.broker.name, f"publish failed: {error}") from None + record(MESSAGING_PUBLISHED, 1, broker=self.broker.name, stream=subject.split(".")[0]) + + async def subscribe(self, stream: str, subject: str, durable: str) -> Subscription: + """Bind to a durable consumer that already exists. + + `durable` is a name the operator created. Binding rather than creating is + what keeps ack policy, max deliveries and the dead-letter arrangement in + one place instead of in whichever service connected first. + """ + pull = self._require_stream() + try: + subscription = await pull.pull_subscribe(subject, durable=durable, stream=stream) + except Exception as error: + raise BrokerUnavailable( + self.broker.name, f"could not bind consumer {durable!r}: {error}" + ) from None + return Subscription(self.broker.name, stream, subscription) + + async def close(self) -> None: + """Drain and close, so in-flight work finishes. + + `drain` rather than `close`: it stops taking new messages, lets what is + already in hand settle, and then closes. A bare close abandons them and + they come back as redeliveries to whoever is left. + """ + if self._connection is None: + return + try: + await self._connection.drain() + except Exception as error: + log.warning( + "could not drain the broker connection", + extra={"broker": self.broker.name, "error": str(error)}, + ) + self._connection = None + self._stream = None + + def _require_stream(self) -> Any: + if self._stream is None: + raise BrokerUnavailable(self.broker.name, "not connected") + return self._stream + + +async def _terminate(message: Any) -> None: + """Terminate a message we could not read, and never raise doing it.""" + try: + await message.term() + except Exception: + return + + +def _delivery_count(message: Any) -> int: + """Which attempt this is, counting from one. + + Absent metadata is reported as a first delivery rather than as zero: zero + would say "never delivered" about a message being held, and the number is + used to decide whether something is poisonous. + """ + metadata = getattr(message, "metadata", None) + delivered = getattr(metadata, "num_delivered", None) + if not isinstance(delivered, int) or delivered < 1: + return 1 + return delivered + + +__all__ = [ + "DEFAULT_FETCH_TIMEOUT_SECONDS", + "Broker", + "MalformedMessage", + "Message", + "MessageBus", + "Subscription", +] diff --git a/src/kit/messaging/_errors.py b/src/kit/messaging/_errors.py new file mode 100644 index 0000000..38b8bc3 --- /dev/null +++ b/src/kit/messaging/_errors.py @@ -0,0 +1,54 @@ +"""What a broker failure is, as types a caller can branch on. + +EVERY ONE OF THESE NAMES THE BROKER, for the reason `kit.clients._errors` gives +about upstreams: a fleet running more than one bus, or one service reading from +two, produces a log line that costs an incident responder the first ten minutes +working out which. + +THEY CARRY NO STATUS CODE AND NO SUBJECT. Which HTTP status a service renders a +broker failure as belongs to the API version doing the rendering. The subject is +left off deliberately too: it frequently carries a tenant id, and an exception +message is the one place that reliably reaches a log aggregator unredacted. +""" + +from __future__ import annotations + + +class BrokerError(Exception): + """Something went wrong talking to the bus.""" + + def __init__(self, broker: str, detail: str = "") -> None: + super().__init__(f"{broker}: {detail}" if detail else broker) + self.broker = broker + self.detail = detail + + +class BrokerUnavailable(BrokerError): + """The bus could not be reached, or would not take the message. + + Retrying LATER may work. This is the shape a publisher should let its own + retry policy see, and the shape a caller should never turn into a dropped + message: on this error the state change that produced the message has + usually already been committed. + """ + + +class BrokerTimeout(BrokerUnavailable): + """We gave up waiting. + + A subclass of Unavailable, deliberately, because to a caller deciding what to + do next it is the same class of problem. Separate because it is the one that + means we stopped rather than the bus said no, and the two lead to different + places when somebody goes looking. + """ + + +class MalformedMessage(BrokerError): + """A message arrived that this package will not hand on. + + Its headers do not carry the four fields a consumer needs, so there is no + tenant to scope to and no id to dedupe on. Not a subclass of Unavailable: + retrying changes nothing, because the message will be malformed next time + too. The consumer's move is to terminate it rather than nak it, which is why + the two settle verbs are separate. + """ diff --git a/src/kit/messaging/_headers.py b/src/kit/messaging/_headers.py new file mode 100644 index 0000000..7ce04a0 --- /dev/null +++ b/src/kit/messaging/_headers.py @@ -0,0 +1,114 @@ +"""What travels beside a message, and why these four and no others. + +THE PAYLOAD IS THE SERVICE'S BUSINESS AND THESE ARE NOT. A publisher and a +consumer have to agree on where the tenant lives and on what makes a message the +same message, before either can read a byte of the body. Left to each service, +those two answers diverge, and the divergence is invisible until a consumer is +silently processing another tenant's event or reprocessing one it already did. + +WHAT IS DELIBERATELY NOT HERE. There is no schema, no version field, no envelope +around the payload. The body is opaque bytes to this package. `kit` holds the +transport policy; what an event IS belongs to the services that agree on it. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import UTC, datetime + +TENANT = "Scadable-Tenant" +"""Which tenant the message belongs to. + +Not optional and not defaulted. A consumer scopes its work by this, and a +message that arrives without one cannot be scoped to anything, so it is refused +here rather than defaulted to something that would read as legitimate later. +""" + +EVENT_ID = "Nats-Msg-Id" +"""What makes this message the same message. NOT one of our own header names. + +`Nats-Msg-Id` is JetStream's own deduplication header: the server refuses a +second publish carrying an id it has already seen inside the stream's duplicate +window. Using our own name here would leave that mechanism switched off and +duplicate suppression entirely to consumers. + +So publishing is idempotent within the window for free, and past the window it +is still the id a consumer dedupes on. That is why the constant is named for +what it means to us and valued for what NATS does with it. +""" + +EVENT_TYPE = "Scadable-Event" +"""What kind of thing happened, so a consumer can route without parsing a body.""" + +OCCURRED_AT = "Scadable-Occurred-At" +"""When it happened, RFC 3339 in UTC. + +When it HAPPENED, not when it was published or delivered. Those differ whenever +a relay was behind, and the difference is the thing anybody debugging a lag +actually wants. +""" + + +@dataclass(frozen=True, slots=True) +class Envelope: + """The four headers, as one object, checked once. + + Frozen because a message's identity should not change between being received + and being settled. + """ + + tenant: str + event_id: str + event_type: str + occurred_at: datetime + + def __post_init__(self) -> None: + """Refuse a blank rather than carry one. + + A blank tenant scopes to nothing, a blank id makes every message unique + and defeats deduplication, and a blank type gives a consumer nothing to + route on. Each is a bug at the publisher that would otherwise be found + by the consumer, hours later, as absent data. + """ + for name, value in ( + ("tenant", self.tenant), + ("event_id", self.event_id), + ("event_type", self.event_type), + ): + if not value.strip(): + raise ValueError(f"{name} is required and was blank") + if self.occurred_at.tzinfo is None: + raise ValueError("occurred_at must be timezone aware") + + def headers(self) -> dict[str, str]: + """The wire form.""" + return { + TENANT: self.tenant, + EVENT_ID: self.event_id, + EVENT_TYPE: self.event_type, + OCCURRED_AT: self.occurred_at.astimezone(UTC).isoformat(), + } + + +def envelope(headers: Mapping[str, str] | None) -> Envelope: + """Read the four headers back, or say which one was missing. + + Raises `ValueError`, which the broker turns into a typed error naming itself. + A message that cannot be read is not a message this package will hand to a + consumer with fields quietly defaulted. + """ + present = headers or {} + occurred = present.get(OCCURRED_AT, "") + try: + at = datetime.fromisoformat(occurred) + except ValueError: + raise ValueError(f"{OCCURRED_AT} is not an RFC 3339 timestamp") from None + # A naive timestamp is ambiguous rather than wrong, and reading it as UTC + # would be a guess that looks like data. `Envelope` refuses it below. + return Envelope( + tenant=present.get(TENANT, ""), + event_id=present.get(EVENT_ID, ""), + event_type=present.get(EVENT_TYPE, ""), + occurred_at=at, + ) diff --git a/src/kit/messaging/_health.py b/src/kit/messaging/_health.py new file mode 100644 index 0000000..2f3329a --- /dev/null +++ b/src/kit/messaging/_health.py @@ -0,0 +1,32 @@ +"""The bus, in `/readyz`, without ever touching the network. + +INFORMATIONAL, NEVER BLOCKING, and the reason is the same one `kit.clients` gives +about an open breaker. A web replica that cannot reach the bus can still serve +every read it owns. Failing readiness would take it out of rotation, which turns +one component's outage into two and does not reconnect anything. + +It reads what the client already believes rather than sending a ping, because a +readiness probe that reaches out turns one kubelet's polling interval into load +on a shared dependency, from every replica, forever. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable, Iterable + +from kit.health import Registry +from kit.messaging._broker import MessageBus + + +def register_brokers(registry: Registry, buses: Iterable[MessageBus]) -> None: + """Declare one check per bus, named for the broker.""" + for bus in buses: + registry.add_informational(bus.broker.name, _check(bus)) + + +def _check(bus: MessageBus) -> Callable[[], Awaitable[None]]: + async def check() -> None: + if not bus.connected: + raise RuntimeError(f"{bus.broker.name} is not connected") + + return check diff --git a/src/kit/observability/__init__.py b/src/kit/observability/__init__.py index 3278d64..ddb1c2e 100644 --- a/src/kit/observability/__init__.py +++ b/src/kit/observability/__init__.py @@ -23,6 +23,10 @@ CLIENT_BREAKER_OPEN, CLIENT_DURATION, CLIENT_REQUESTS, + MESSAGING_CONSUMED, + MESSAGING_DELIVERIES, + MESSAGING_PUBLISHED, + MESSAGING_SETTLED, SERVER_DURATION, SERVER_REQUESTS, record, @@ -41,6 +45,10 @@ "CLIENT_BREAKER_OPEN", "CLIENT_DURATION", "CLIENT_REQUESTS", + "MESSAGING_CONSUMED", + "MESSAGING_DELIVERIES", + "MESSAGING_PUBLISHED", + "MESSAGING_SETTLED", "SERVER_DURATION", "SERVER_REQUESTS", "JSONFormatter", diff --git a/src/kit/observability/_metrics.py b/src/kit/observability/_metrics.py index a915a4a..edd3b96 100644 --- a/src/kit/observability/_metrics.py +++ b/src/kit/observability/_metrics.py @@ -40,6 +40,14 @@ CLIENT_ATTEMPTS = "http.client.attempts" CLIENT_BREAKER_OPEN = "http.client.breaker_open" +# MESSAGING. The label is the STREAM, never the subject: a subject carries a +# tenant id in most useful designs, and the module docstring above is about +# exactly that turning one series into millions. +MESSAGING_PUBLISHED = "messaging.published" +MESSAGING_CONSUMED = "messaging.consumed" +MESSAGING_SETTLED = "messaging.settled" +MESSAGING_DELIVERIES = "messaging.deliveries" + def start_metrics( *, @@ -128,6 +136,22 @@ def _build(meter: Any) -> dict[str, Any]: CLIENT_BREAKER_OPEN: meter.create_up_down_counter( CLIENT_BREAKER_OPEN, unit="1", description="Upstreams currently being shed" ), + MESSAGING_PUBLISHED: meter.create_counter( + MESSAGING_PUBLISHED, unit="1", description="Messages published, by stream" + ), + MESSAGING_CONSUMED: meter.create_counter( + MESSAGING_CONSUMED, unit="1", description="Messages received, by stream" + ), + MESSAGING_SETTLED: meter.create_counter( + MESSAGING_SETTLED, + unit="1", + description="Messages settled, by outcome: ack, nak or term", + ), + MESSAGING_DELIVERIES: meter.create_histogram( + MESSAGING_DELIVERIES, + unit="1", + description="Delivery attempt this message is on. Above 1 is redelivery.", + ), } diff --git a/tests/test_messaging.py b/tests/test_messaging.py new file mode 100644 index 0000000..06a782c --- /dev/null +++ b/tests/test_messaging.py @@ -0,0 +1,623 @@ +"""The bus, exercised against a stub connection rather than a server. + +The stub is the same seam `service_client(transport=...)` uses: the real header, +settle, metric and error code runs, with no socket and no sleep of consequence. +""" + +from __future__ import annotations + +import builtins +import sys +import types +from datetime import UTC, datetime, timedelta +from typing import Any + +import httpx +import pytest +from fastapi import FastAPI + +from kit.health import Registry +from kit.httpapi import install_conventions +from kit.messaging import ( + EVENT_ID, + OCCURRED_AT, + TENANT, + Broker, + BrokerTimeout, + BrokerUnavailable, + Envelope, + MessageBus, + envelope, + register_brokers, +) +from kit.messaging._broker import Subscription, _delivery_count +from kit.observability import _metrics + +WHEN = datetime(2026, 9, 7, 12, 0, tzinfo=UTC) + + +def an_envelope(**overrides: Any) -> Envelope: + fields: dict[str, Any] = { + "tenant": "org_1", + "event_id": "evt_1", + "event_type": "repository.indexed", + "occurred_at": WHEN, + } + fields.update(overrides) + return Envelope(**fields) + + +class FakeMetadata: + def __init__(self, delivered: int) -> None: + self.num_delivered = delivered + + +class FakeMessage: + """One delivery, recording how it was settled.""" + + def __init__( + self, + headers: dict[str, str] | None = None, + data: bytes = b"{}", + delivered: int = 1, + fails: bool = False, + ) -> None: + self.headers = headers if headers is not None else an_envelope().headers() + self.data = data + self.metadata = FakeMetadata(delivered) + self.settled: list[tuple[str, float | None]] = [] + self._fails = fails + + async def ack(self) -> None: + self._record("ack") + + async def nak(self, delay: float | None = None) -> None: + self._record("nak", delay) + + async def term(self) -> None: + self._record("term") + + def _record(self, outcome: str, delay: float | None = None) -> None: + if self._fails: + raise RuntimeError("the connection went away") + self.settled.append((outcome, delay)) + + +class FakeSubscription: + def __init__(self, batches: list[Any]) -> None: + self._batches = batches + + async def fetch(self, batch: int, timeout: float) -> Any: # noqa: ASYNC109 + result = self._batches.pop(0) + if isinstance(result, Exception): + raise result + return result + + +class FakeStream: + def __init__(self, *, publish_error: Exception | None = None) -> None: + self.published: list[tuple[str, bytes, dict[str, str], float]] = [] + self._publish_error = publish_error + self.subscription = FakeSubscription([]) + self.subscribe_error: Exception | None = None + + async def publish( + self, + subject: str, + payload: bytes, + headers: dict[str, str], + timeout: float, # noqa: ASYNC109 + ) -> None: + if self._publish_error is not None: + raise self._publish_error + self.published.append((subject, payload, headers, timeout)) + + async def pull_subscribe(self, subject: str, durable: str, stream: str) -> Any: + if self.subscribe_error is not None: + raise self.subscribe_error + return self.subscription + + +class FakeConnection: + def __init__(self, stream: FakeStream | None = None, *, drain_fails: bool = False) -> None: + self.is_connected = True + self._stream = stream or FakeStream() + self.drained = False + self._drain_fails = drain_fails + + def jetstream(self) -> FakeStream: + return self._stream + + async def drain(self) -> None: + if self._drain_fails: + raise RuntimeError("drain refused") + self.drained = True + + +def a_bus(connection: FakeConnection | None = None) -> MessageBus: + return MessageBus(Broker(name="events", servers=("nats://x:4222",)), connection) + + +# --- the envelope ------------------------------------------------------------ + + +def test_the_four_headers_go_out_and_come_back() -> None: + """A round trip, because the wire form is the contract between services.""" + read = envelope(an_envelope().headers()) + + assert read == an_envelope() + + +def test_the_event_id_rides_as_the_header_jetstream_dedupes_on() -> None: + """NOT one of our own names, and that is the whole point: the server refuses + a second publish carrying an id it has already seen.""" + assert EVENT_ID == "Nats-Msg-Id" + assert an_envelope().headers()[EVENT_ID] == "evt_1" + + +def test_an_occurred_at_is_normalised_to_utc() -> None: + """Two publishers in two zones must produce the same string for one moment.""" + elsewhere = WHEN.astimezone(_offset(hours=5)) + + assert an_envelope(occurred_at=elsewhere).headers()[OCCURRED_AT] == WHEN.isoformat() + + +@pytest.mark.parametrize("blank", ["", " "]) +@pytest.mark.parametrize("field", ["tenant", "event_id", "event_type"]) +def test_a_blank_required_field_is_refused(field: str, blank: str) -> None: + """A blank tenant scopes to nothing and a blank id defeats deduplication. + Both are publisher bugs that would otherwise be found by the consumer.""" + with pytest.raises(ValueError, match=field): + an_envelope(**{field: blank}) + + +def test_a_naive_timestamp_is_refused_rather_than_assumed_utc() -> None: + """Reading it as UTC would be a guess that then looks like data.""" + with pytest.raises(ValueError, match="timezone aware"): + an_envelope(occurred_at=datetime(2026, 9, 7, 12, 0)) # noqa: DTZ001 + + +def test_an_unparsable_timestamp_names_the_header() -> None: + with pytest.raises(ValueError, match=OCCURRED_AT): + envelope({**an_envelope().headers(), OCCURRED_AT: "yesterday"}) + + +def test_absent_headers_are_refused_by_name() -> None: + """`None` is what a message with no headers at all carries.""" + with pytest.raises(ValueError): + envelope(None) + + +# --- publishing -------------------------------------------------------------- + + +async def test_publishing_sends_the_headers_and_waits_for_the_stream() -> None: + """Waiting is the point: an unwaited publish reports success while the + message is still in flight.""" + connection = FakeConnection() + bus = a_bus(connection) + + await bus.publish("events.repository.indexed", b"payload", an_envelope()) + + subject, payload, headers, timeout = connection.jetstream().published[0] + assert subject == "events.repository.indexed" + assert payload == b"payload" + assert headers[TENANT] == "org_1" + assert timeout == bus.broker.publish_timeout_seconds + + +async def test_a_publish_timeout_is_its_own_type() -> None: + """Separate from Unavailable because it means we stopped, not that the bus + said no, and the two lead different places.""" + bus = a_bus(FakeConnection(FakeStream(publish_error=TimeoutError()))) + + with pytest.raises(BrokerTimeout, match="events"): + await bus.publish("events.x", b"", an_envelope()) + + +async def test_a_failed_publish_names_the_broker() -> None: + bus = a_bus(FakeConnection(FakeStream(publish_error=RuntimeError("no stream")))) + + with pytest.raises(BrokerUnavailable, match="events"): + await bus.publish("events.x", b"", an_envelope()) + + +async def test_publishing_before_connecting_is_refused() -> None: + bus = MessageBus(Broker(name="events", servers=("nats://x:4222",))) + + with pytest.raises(BrokerUnavailable, match="not connected"): + await bus.publish("events.x", b"", an_envelope()) + + +# --- consuming --------------------------------------------------------------- + + +async def test_fetch_returns_unsettled_messages() -> None: + """Nothing is acknowledged on the caller's behalf, so a handler that raises + gets its message back rather than losing it.""" + raw = FakeMessage() + subscription = Subscription("events", "EVENTS", FakeSubscription([[raw]])) + + [message] = await subscription.fetch() + + assert message.envelope.tenant == "org_1" + assert message.payload == b"{}" + assert raw.settled == [] + + +async def test_an_empty_batch_is_not_an_error() -> None: + """The ordinary case on a quiet stream. The client signals it as a timeout.""" + subscription = Subscription("events", "EVENTS", FakeSubscription([TimeoutError()])) + + assert await subscription.fetch() == () + + +async def test_a_failed_fetch_names_the_broker() -> None: + subscription = Subscription("events", "EVENTS", FakeSubscription([RuntimeError("gone")])) + + with pytest.raises(BrokerUnavailable, match="events"): + await subscription.fetch() + + +async def test_an_unreadable_message_is_terminated_not_naked() -> None: + """It will be malformed next time too, so a nak would spend the whole + attempt budget rediscovering that.""" + bad = FakeMessage(headers={TENANT: "org_1"}) + subscription = Subscription("events", "EVENTS", FakeSubscription([[bad]])) + + assert await subscription.fetch() == () + assert bad.settled == [("term", None)] + + +async def test_terminating_an_unreadable_message_never_raises() -> None: + """The connection may already be gone; that must not become the caller's + problem on a message we were dropping anyway.""" + bad = FakeMessage(headers={}, fails=True) + subscription = Subscription("events", "EVENTS", FakeSubscription([[bad]])) + + assert await subscription.fetch() == () + + +async def test_a_readable_message_survives_beside_an_unreadable_one() -> None: + """One poisonous message must not cost the batch.""" + subscription = Subscription( + "events", "EVENTS", FakeSubscription([[FakeMessage(headers={}), FakeMessage()]]) + ) + + assert len(await subscription.fetch()) == 1 + + +# --- settling ---------------------------------------------------------------- + + +async def test_the_three_settle_verbs_reach_the_message() -> None: + raw = [FakeMessage(), FakeMessage(), FakeMessage()] + subscription = Subscription("events", "EVENTS", FakeSubscription([raw])) + one, two, three = await subscription.fetch(batch=3) + + await one.ack() + await two.nak(delay_seconds=30.0) + await three.term() + + assert raw[0].settled == [("ack", None)] + assert raw[1].settled == [("nak", 30.0)] + assert raw[2].settled == [("term", None)] + + +async def test_a_failed_settle_does_not_raise() -> None: + """The work happened; only the acknowledgement did not. Raising would tell a + caller its work failed, which is worse than the truth: the message comes back + and the handler's idempotency covers it.""" + subscription = Subscription("events", "EVENTS", FakeSubscription([[FakeMessage(fails=True)]])) + [message] = await subscription.fetch() + + await message.ack() + + +async def test_a_redelivery_reports_which_attempt_it_is() -> None: + """Above one is how a handler tells poisonous from merely slow.""" + subscription = Subscription("events", "EVENTS", FakeSubscription([[FakeMessage(delivered=4)]])) + + [message] = await subscription.fetch() + + assert message.delivered == 4 + + +@pytest.mark.parametrize("metadata", [None, FakeMetadata(0), "not metadata"]) +def test_absent_delivery_metadata_reads_as_a_first_delivery(metadata: Any) -> None: + """Zero would say "never delivered" about a message being held.""" + + class Bare: + pass + + message = Bare() + if metadata is not None: + message.metadata = metadata # type: ignore[attr-defined] + + assert _delivery_count(message) == 1 + + +# --- connecting -------------------------------------------------------------- + + +async def test_connecting_twice_does_not_open_a_second_connection() -> None: + connection = FakeConnection() + bus = a_bus(connection) + + await bus.connect() + + assert bus.connected + + +async def test_connecting_with_no_servers_is_refused() -> None: + bus = MessageBus(Broker(name="events")) + + with pytest.raises(BrokerUnavailable, match="no servers"): + await bus.connect() + + +async def test_connecting_without_the_extra_names_the_extra( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The failure a service that forgot the extra can actually read.""" + real = builtins.__import__ + + def missing(name: str, *args: Any, **kwargs: Any) -> Any: + if name == "nats": + raise ImportError(name) + return real(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", missing) + bus = MessageBus(Broker(name="events", servers=("nats://x:4222",))) + + with pytest.raises(BrokerUnavailable, match=r"scadable-kit\[nats\]"): + await bus.connect() + + +class FakeNats: + """Stands in for the `nats` module, so the real connect path is exercised.""" + + def __init__(self, connection: Any = None, error: Exception | None = None) -> None: + self.connection = connection + self.error = error + self.options: dict[str, Any] = {} + + async def connect(self, **options: Any) -> Any: + self.options = options + if self.error is not None: + raise self.error + return self.connection + + +@pytest.fixture +def nats_module(monkeypatch: pytest.MonkeyPatch): + """Install a stub `nats` module for the duration of one test.""" + + def install(fake: FakeNats) -> FakeNats: + module = types.ModuleType("nats") + module.connect = fake.connect # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "nats", module) + return fake + + return install + + +async def test_connecting_passes_the_reconnect_settings_through(nats_module: Any) -> None: + """Retrying forever is deliberate: a consumer that gives up becomes a + process that is running, healthy by any liveness probe, and consuming + nothing.""" + fake = nats_module(FakeNats(FakeConnection())) + bus = MessageBus(Broker(name="events", servers=("nats://x:4222",))) + + await bus.connect() + + assert fake.options["max_reconnect_attempts"] == -1 + assert fake.options["servers"] == ["nats://x:4222"] + assert "user_credentials" not in fake.options + assert "token" not in fake.options + assert bus.connected + + +async def test_credentials_are_passed_when_set(nats_module: Any) -> None: + """Carried even though the deployed bus has no auth, so turning it on is a + config change in one service rather than a release of this package.""" + fake = nats_module(FakeNats(FakeConnection())) + bus = MessageBus( + Broker( + name="events", + servers=("nats://x:4222",), + credentials_path="/etc/nats/creds", + token="t", # noqa: S106 + ) + ) + + await bus.connect() + + assert fake.options["user_credentials"] == "/etc/nats/creds" + assert fake.options["token"] == "t" # noqa: S105 + + +async def test_a_connect_timeout_is_its_own_type(nats_module: Any) -> None: + nats_module(FakeNats(error=TimeoutError())) + bus = MessageBus(Broker(name="events", servers=("nats://x:4222",))) + + with pytest.raises(BrokerTimeout, match="connect timed out"): + await bus.connect() + + +async def test_a_refused_connect_names_the_broker(nats_module: Any) -> None: + nats_module(FakeNats(error=RuntimeError("no route to host"))) + bus = MessageBus(Broker(name="events", servers=("nats://x:4222",))) + + with pytest.raises(BrokerUnavailable, match="events"): + await bus.connect() + + +async def test_connecting_an_already_connected_bus_is_a_no_op(nats_module: Any) -> None: + """A composition root that retries should not leak a second connection.""" + fake = nats_module(FakeNats(FakeConnection())) + bus = a_bus(FakeConnection()) + + await bus.connect() + + assert fake.options == {} + + +async def test_a_broker_needs_a_name() -> None: + """It labels metrics and health checks, so a blank one is a series nobody + can find.""" + with pytest.raises(ValueError, match="name"): + Broker(name=" ") + + +def test_configured_answers_without_opening_a_socket() -> None: + assert not Broker(name="events").configured + assert Broker(name="events", servers=("nats://x:4222",)).configured + + +# --- subscribing ------------------------------------------------------------- + + +async def test_subscribing_binds_an_existing_durable_consumer() -> None: + """Binding rather than creating keeps ack policy and max deliveries in one + place instead of in whichever service connected first.""" + bus = a_bus(FakeConnection()) + + subscription = await bus.subscribe(stream="EVENTS", subject="events.>", durable="brain") + + assert subscription.stream == "EVENTS" + + +async def test_binding_a_missing_consumer_names_it() -> None: + stream = FakeStream() + stream.subscribe_error = RuntimeError("consumer not found") + bus = a_bus(FakeConnection(stream)) + + with pytest.raises(BrokerUnavailable, match="brain"): + await bus.subscribe(stream="EVENTS", subject="events.>", durable="brain") + + +# --- closing ----------------------------------------------------------------- + + +async def test_closing_drains_so_in_flight_work_finishes() -> None: + """A bare close abandons held messages and they come back as redeliveries.""" + connection = FakeConnection() + bus = a_bus(connection) + + await bus.close() + + assert connection.drained + assert not bus.connected + + +async def test_closing_an_unconnected_bus_is_a_no_op() -> None: + await MessageBus(Broker(name="events")).close() + + +async def test_a_failed_drain_does_not_raise_on_the_way_down() -> None: + bus = a_bus(FakeConnection(drain_fails=True)) + + await bus.close() + + assert not bus.connected + + +# --- readiness --------------------------------------------------------------- + + +async def readiness_of(registry: Registry) -> httpx.Response: + app = FastAPI() + install_conventions(app, readiness=registry) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + return await client.get("/readyz") + + +async def test_a_connected_bus_reports_ready() -> None: + registry = Registry() + register_brokers(registry, [a_bus(FakeConnection())]) + + response = await readiness_of(registry) + + assert response.json()["checks"]["events"] == "ready" + + +async def test_a_disconnected_bus_is_visible_but_does_not_block() -> None: + """THE decision, asserted. A replica that cannot reach the bus still serves + every read it owns, so failing readiness would turn one component's outage + into two and would not reconnect anything.""" + registry = Registry() + register_brokers(registry, [MessageBus(Broker(name="events"))]) + + response = await readiness_of(registry) + + assert response.status_code == 200, "a disconnected bus took the service out of rotation" + assert response.json()["status"] == "ready" + assert response.json()["checks"]["events"] == "error" + + +# --- metrics ----------------------------------------------------------------- + + +class Recorder: + def __init__(self) -> None: + self.calls: list[tuple[float, dict[str, Any]]] = [] + + def add(self, value: float, attributes: dict[str, Any]) -> None: + self.calls.append((value, attributes)) + + +@pytest.fixture +def recorders(monkeypatch: pytest.MonkeyPatch) -> dict[str, Recorder]: + """Every messaging instrument, so a name absent from `_build` shows up here + as a missing key rather than as silence.""" + made = { + name: Recorder() + for name in ( + _metrics.MESSAGING_PUBLISHED, + _metrics.MESSAGING_CONSUMED, + _metrics.MESSAGING_SETTLED, + _metrics.MESSAGING_DELIVERIES, + ) + } + monkeypatch.setattr(_metrics, "_instruments", dict(made)) + return made + + +async def test_the_metric_label_is_the_stream_never_the_subject( + recorders: dict[str, Recorder], +) -> None: + """A subject carries a tenant id in most useful designs, and one label + holding one turns a single series into millions.""" + await a_bus(FakeConnection()).publish("events.repository.indexed", b"", an_envelope()) + + _, attributes = recorders[_metrics.MESSAGING_PUBLISHED].calls[0] + assert attributes == {"broker": "events", "stream": "events"} + + +async def test_settling_is_counted_by_outcome(recorders: dict[str, Recorder]) -> None: + subscription = Subscription("events", "EVENTS", FakeSubscription([[FakeMessage()]])) + [message] = await subscription.fetch() + + await message.nak() + + _, attributes = recorders[_metrics.MESSAGING_SETTLED].calls[0] + assert attributes["outcome"] == "nak" + + +async def test_a_failed_settle_is_not_counted_as_settled( + recorders: dict[str, Recorder], +) -> None: + """Counting it would report an acknowledgement that never reached the bus.""" + subscription = Subscription("events", "EVENTS", FakeSubscription([[FakeMessage(fails=True)]])) + [message] = await subscription.fetch() + + await message.ack() + + assert recorders[_metrics.MESSAGING_SETTLED].calls == [] + + +def _offset(hours: int) -> Any: + """A fixed offset, without importing zoneinfo for one assertion.""" + return __import__("datetime").timezone(timedelta(hours=hours)) diff --git a/uv.lock b/uv.lock index 9c3fa0e..c8ac7ec 100644 --- a/uv.lock +++ b/uv.lock @@ -243,6 +243,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, ] +[[package]] +name = "nats-py" +version = "2.15.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/02/f0/fc5e93f2b0dd14a202590ad9d30eda1955ea872039b5204357348d0f4b1e/nats_py-2.15.0.tar.gz", hash = "sha256:6622c547d9a7d2313d9c147d46c386188f4ec2c7b5c9f9a0438a4d1b55f54a93", size = 75995, upload-time = "2026-06-05T07:34:03.904Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/db/a8/b55606c7c621fb813c8ec78baf201d2c78bf6051091ec0c7ada572999e95/nats_py-2.15.0-py3-none-any.whl", hash = "sha256:9f8d36aa52a9926a88b8f1d70cf1fdce0ad387941479b500ee9ab3e51073cefd", size = 90334, upload-time = "2026-06-05T07:34:02.81Z" }, +] + [[package]] name = "nodeenv" version = "1.10.0" @@ -623,7 +632,7 @@ wheels = [ [[package]] name = "scadable-kit" -version = "0.6.1" +version = "0.7.0" source = { editable = "." } dependencies = [ { name = "fastapi" }, @@ -633,6 +642,9 @@ dependencies = [ ] [package.optional-dependencies] +nats = [ + { name = "nats-py" }, +] otel = [ { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "opentelemetry-instrumentation-fastapi" }, @@ -654,6 +666,7 @@ dev = [ requires-dist = [ { name = "fastapi", specifier = ">=0.141,<0.142" }, { name = "httpx", specifier = ">=0.28,<0.29" }, + { name = "nats-py", marker = "extra == 'nats'", specifier = ">=2.15,<3" }, { name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'otel'", specifier = ">=1.44,<2" }, { name = "opentelemetry-instrumentation-fastapi", marker = "extra == 'otel'", specifier = ">=0.65b0" }, { name = "opentelemetry-instrumentation-httpx", marker = "extra == 'otel'", specifier = ">=0.65b0" }, @@ -661,7 +674,7 @@ requires-dist = [ { name = "pydantic", specifier = ">=2.13,<3" }, { name = "pydantic-settings", specifier = ">=2.7,<3" }, ] -provides-extras = ["otel"] +provides-extras = ["otel", "nats"] [package.metadata.requires-dev] dev = [