From 4b3edf107457d87935e7246c22d375a833b27798 Mon Sep 17 00:00:00 2001 From: livesite Date: Thu, 27 Aug 2026 21:57:49 +0800 Subject: [PATCH] release: add WebSocket TTL live subscriptions for 0.10.0 --- Cargo.lock | 96 +- Cargo.toml | 6 +- README.md | 215 +++- console-ui/package.json | 2 +- console-ui/src/layout/AppShell.tsx | 2 +- crates/discovery/Cargo.toml | 1 + crates/discovery/src/main.rs | 26 +- crates/discovery/src/server.rs | 31 +- crates/operator/src/controller/mod.rs | 26 + crates/operator/src/controller/preflight.rs | 32 +- crates/operator/src/controller/status.rs | 1 + crates/operator/src/crd.rs | 38 + crates/operator/src/lib.rs | 2 +- crates/operator/src/resources.rs | 49 +- crates/queue/src/broker.rs | 23 +- crates/queue/src/broker/expiration.rs | 4 + crates/queue/src/broker_tests.rs | 93 ++ crates/queue/src/topic.rs | 11 + crates/server/src/config.rs | 51 + crates/server/src/config/environment.rs | 21 + crates/server/src/config/tests.rs | 13 + crates/server/src/http.rs | 39 +- crates/server/src/http/kodo_compat.rs | 7 + crates/server/src/http/websocket.rs | 1061 +++++++++++++++++ crates/server/src/main.rs | 9 + crates/server/src/metrics.rs | 44 + crates/server/src/tcp.rs | 8 +- crates/server/src/tcp/ephemeral.rs | 31 +- deploy/helm/rustqueue/Chart.yaml | 4 +- .../crds/rustqueue.io_rustqueues.yaml | 17 + deploy/helm/rustqueue/templates/NOTES.txt | 7 + deploy/helm/rustqueue/templates/queue.yaml | 7 + deploy/helm/rustqueue/values.yaml | 10 +- docs/architecture/share-nothing-v7.md | 126 +- docs/operations/kubernetes.md | 81 +- docs/releases/v0.10.0.md | 204 ++++ fuzz/Cargo.lock | 122 +- rustqueue.example.toml | 9 + scripts/acceptance-multi-broker-k8s.sh | 4 +- site/index.html | 6 +- 40 files changed, 2431 insertions(+), 108 deletions(-) create mode 100644 crates/server/src/http/websocket.rs create mode 100644 docs/releases/v0.10.0.md diff --git a/Cargo.lock b/Cargo.lock index 022585e..26974c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -245,6 +245,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", + "base64 0.22.1", "bytes", "form_urlencoded", "futures-util", @@ -263,8 +264,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1", "sync_wrapper", "tokio", + "tokio-tungstenite", "tower", "tower-layer", "tower-service", @@ -2025,10 +2028,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" dependencies = [ "libc", - "rand_chacha", + "rand_chacha 0.3.1", "rand_core 0.6.4", ] +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha 0.9.0", + "rand_core 0.9.5", +] + [[package]] name = "rand" version = "0.10.2" @@ -2050,6 +2063,16 @@ dependencies = [ "rand_core 0.6.4", ] +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + [[package]] name = "rand_core" version = "0.6.4" @@ -2059,6 +2082,15 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "rand_core" version = "0.10.1" @@ -2283,7 +2315,7 @@ dependencies = [ [[package]] name = "rustqueue-bench" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "clap", @@ -2295,7 +2327,7 @@ dependencies = [ [[package]] name = "rustqueue-console" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "axum", @@ -2323,7 +2355,7 @@ dependencies = [ [[package]] name = "rustqueue-discovery" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "axum", @@ -2340,13 +2372,14 @@ dependencies = [ "sha2", "tokio", "tower", + "tower-http", "tracing", "tracing-subscriber", ] [[package]] name = "rustqueue-operator" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "axum", @@ -2373,7 +2406,7 @@ dependencies = [ [[package]] name = "rustqueue-protocol" -version = "0.9.0" +version = "0.10.0" dependencies = [ "bytes", "serde", @@ -2383,7 +2416,7 @@ dependencies = [ [[package]] name = "rustqueue-proxy" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "axum", @@ -2404,7 +2437,7 @@ dependencies = [ [[package]] name = "rustqueue-queue" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "bytes", @@ -2426,7 +2459,7 @@ dependencies = [ [[package]] name = "rustqueue-server" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "async-compression", @@ -2462,7 +2495,7 @@ dependencies = [ [[package]] name = "rustqueue-storage" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "crc32c", @@ -2475,14 +2508,14 @@ dependencies = [ [[package]] name = "rustqueue-telemetry" -version = "0.9.0" +version = "0.10.0" dependencies = [ "serde", ] [[package]] name = "rustqueuectl" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "clap", @@ -2713,6 +2746,17 @@ dependencies = [ "unsafe-libyaml", ] +[[package]] +name = "sha1" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest", +] + [[package]] name = "sha2" version = "0.10.9" @@ -2980,6 +3024,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -3174,6 +3230,22 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.5", + "sha1", + "thiserror", +] + [[package]] name = "typenum" version = "1.20.1" diff --git a/Cargo.toml b/Cargo.toml index 65cb892..147851a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,7 @@ members = [ exclude = ["fuzz"] [workspace.package] -version = "0.9.0" +version = "0.10.0" edition = "2021" license = "Apache-2.0" rust-version = "1.88" @@ -25,7 +25,7 @@ rust-version = "1.88" anyhow = "1.0.95" async-compression = { version = "0.4.42", features = ["tokio", "deflate"] } flate2 = "1.1" -axum = "0.8.1" +axum = { version = "0.8.1", features = ["ws"] } axum-server = { version = "0.7.2", features = ["tls-rustls"] } bytes = { version = "1.9.0", features = ["serde"] } chrono = { version = "0.4.39", default-features = false, features = ["clock", "serde"] } @@ -51,7 +51,7 @@ tokio-rustls = "0.26.1" tokio-snappy = "0.2.0" toml = "0.8.19" tower = "0.5.2" -tower-http = { version = "0.6.2", features = ["fs", "timeout", "trace"] } +tower-http = { version = "0.6.2", features = ["cors", "fs", "timeout", "trace"] } tracing = "0.1.41" tracing-subscriber = { version = "0.3.19", features = ["env-filter", "json"] } rustqueue-telemetry = { path = "crates/telemetry" } diff --git a/README.md b/README.md index c4faa7d..1865bf0 100644 --- a/README.md +++ b/README.md @@ -11,14 +11,14 @@ [NSQ performance boundaries](docs/architecture/nsq-performance.md) · [Kubernetes operations](docs/operations/kubernetes.md) · [Console operations](docs/operations/console.md) · -[v0.9.0 release](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.9.0) +[v0.10.0 release](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.10.0) -RustQueue 0.9.0 is a Kubernetes-native, NSQ V2-compatible message queue for +RustQueue 0.10.0 is a Kubernetes-native, NSQ V2-compatible message queue for trusted internal networks. It is written in Rust and uses a deliberately simple share-nothing model: each Broker owns one durable RWO PVC, while Kubernetes provides scheduling, rollout and discovery. -> Current release: [v0.9.0](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.9.0). +> Current release: [v0.10.0](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.10.0). > RustQueue is a production candidate for workloads that accept single-PVC > durability and at-least-once delivery. It does not replicate messages between > Brokers and is not an HA replacement for a replicated log. @@ -32,6 +32,7 @@ The complete architecture and reliability contract is documented in | --- | --- | | Durability | Default `PUB`/`MPUB`/`DPUB` return after local segment `fsync`; opt-in `write_ack` and `nsq_relaxed` return after append with explicit crash-loss windows; `FIN`/`REQ` use a durable channel WAL | | Delivery | At least once; a restart may redeliver a message without a durable `FIN` | +| Live subscriptions | Optional WebSocket `rustqueue.live.v1` stream for existing `TTL_DISCARD` Topics; best effort, no replay or ACK | | Compatibility | NSQ V2 core commands, lookup, standard Stats fields, TLS/mTLS, AUTH, Snappy, Deflate, fan-out and ephemeral channels | | Kodo | Default-off compatibility profile: stable publish Gateways from `/nodes`, real Broker owners from `/lookup`, and no upstream Kodo change | | Message size | Conservative 20 MiB default; stable 100 MiB protocol/storage ceiling; Kodo profile validates exactly 104,857,500 bytes | @@ -44,54 +45,43 @@ messages stored on that Broker are lost. Configure disk pressure protection, monitor the exported metrics, and choose PVC/storage failure policies that fit your workload before deploying to production. -## What's new in 0.9.0 - -- **Topic-level TTL discard.** Managed Topics can choose `TTL_DISCARD` with a - positive whole-second `messageTtlSeconds`; `RELIABLE` remains the default and - requires no TTL. The TTL starts at the Broker write timestamp and is an - absolute lifetime. -- **One final expiration boundary.** Expiry applies to unrouted, deferred, - requeued and in-flight messages. It evicts the message from every Channel, - never writes a DLQ entry, and lets `TTL_DISCARD` retry normally until the TTL - rather than ending through retention or maximum-attempt DLQ rules. -- **Safe online policy changes.** Topic creation can set the policy, and - `/v1/manage/topics/configure` supports the `CONFIGURE` operation for existing - Topics. Enabling or shortening TTL previews possible backlog loss and - requires the exact Topic name; visible multi-owner migrations block policy - changes. Extending or disabling TTL cannot revive an already expired message. -- **Deadline-bound delivery.** Reservation and socket handoff recheck the - absolute deadline. If a frame cannot finish before expiry, the consumer - connection closes instead of leaving a partially parseable NSQ frame; a - complete frame that is not `FIN`ed still expires at the same boundary. -- **Wire and storage compatibility.** `PUB`, `MPUB` and `DPUB` keep their - existing `publish_ack_mode` acknowledgement behavior, and NSQ TCP/HTTP - publishing remains unchanged. Existing and auto-created Topics stay - `RELIABLE`; old manifests default to that mode, and disk format v7 needs no - migration. - -See the [v0.9.0 release notes](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.9.0) -and [NSQ performance boundaries](docs/architecture/nsq-performance.md) for -the contract and benchmark interpretation. - -## Download 0.9.0 +## What's new in 0.10.0 + +- **TTL live WebSocket subscriptions.** The default-off WebSocket endpoint + streams only existing `TTL_DISCARD` Topics. It is a best-effort feed with no + replay, reconnect resume or client ACK; consumers use Discovery `/lookup` + and connect to every returned Topic owner. +- **Bounded live delivery.** Each connection uses one ephemeral Channel and at + most one reservation. Frame memory, total connections and per-Topic + connections are bounded; a slow or expired write closes the connection and + releases the reservation. +- **Existing protocol compatibility.** Topic TTL discard remains the v0.9.0 + capability, while NSQ TCP/HTTP publishing, ACK modes, storage format v7 and + existing `RELIABLE` behavior remain unchanged. + +See the [v0.10.0 release notes](docs/releases/v0.10.0.md) and [NSQ performance +boundaries](docs/architecture/nsq-performance.md) for the contract and +benchmark interpretation. + +## Download 0.10.0 Every release contains native Linux binaries, the Console UI, source, the Helm Chart and a checksum manifest: | Asset | Contents | | --- | --- | -| `rustqueue-0.9.0-linux-x86_64.tar.gz` | Linux x86_64 binaries, Console UI and example configuration | -| `rustqueue-0.9.0-linux-aarch64.tar.gz` | Linux ARM64 binaries, Console UI and example configuration | -| `rustqueue-0.9.0-source.tar.gz` | Source archive for the tagged commit | -| `rustqueue-0.9.0.tgz` | Helm Chart | -| `SHA256SUMS-0.9.0` | SHA-256 checksums for every downloadable artifact | +| `rustqueue-0.10.0-linux-x86_64.tar.gz` | Linux x86_64 binaries, Console UI and example configuration | +| `rustqueue-0.10.0-linux-aarch64.tar.gz` | Linux ARM64 binaries, Console UI and example configuration | +| `rustqueue-0.10.0-source.tar.gz` | Source archive for the tagged commit | +| `rustqueue-0.10.0.tgz` | Helm Chart | +| `SHA256SUMS-0.10.0` | SHA-256 checksums for every downloadable artifact | ```sh arch="$(uname -m)" -curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.9.0/rustqueue-0.9.0-linux-${arch}.tar.gz" -curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.9.0/SHA256SUMS-0.9.0" -sha256sum --check --ignore-missing SHA256SUMS-0.9.0 -tar -xzf "rustqueue-0.9.0-linux-${arch}.tar.gz" +curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.10.0/rustqueue-0.10.0-linux-${arch}.tar.gz" +curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.10.0/SHA256SUMS-0.10.0" +sha256sum --check --ignore-missing SHA256SUMS-0.10.0 +tar -xzf "rustqueue-0.10.0-linux-${arch}.tar.gz" ``` ## Architecture @@ -182,7 +172,7 @@ operator -> eligible nodes -> StatefulSet ordinal + retained RWO PVC sent is returned as ambiguous and is never retried automatically. - `rustqueue-operator`: creates the StatefulSet, retained PVCs, discovery, proxy, RBAC, disruption budgets, PVC expansion and drain-aware one-at-a-time - rolling updates. The 0.9 Kodo profile adds atomic Discovery cutover, + rolling updates. The Kodo compatibility profile adds atomic Discovery cutover, producer-restart fencing and fail-closed decommissioning. - `rustqueue-console`: Kubernetes and broker observability backend serving the bilingual Carbon UI, with default-off native Topic/Channel management. @@ -193,6 +183,130 @@ The broker implements `IDENTIFY`, `AUTH`, `SUB`, `PUB`, `MPUB`, `DPUB`, `RDY`, `FIN`, `REQ`, `TOUCH`, `NOP`, and `CLS`, including TLS/mTLS, Snappy, Deflate, output buffering, sampling, fan-out and ephemeral channels. +## WebSocket TTL live subscriptions + +RustQueue can expose a best-effort live feed for short-lived data without +changing the NSQ TCP contract. WebSocket support is disabled by default. When +enabled, each Broker's HTTP listener accepts: + +```text +GET /v1/ws/topics/{topic} +``` + +The request must negotiate the `rustqueue.live.v1` subprotocol. The Topic must +already exist on that Broker and have `deliveryMode=TTL_DISCARD`; the endpoint +never creates a Topic. A missing Topic and a `RELIABLE` Topic are rejected during +the handshake with the stable WebSocket error response. One connection covers +one Topic and starts at the current tail, so messages published before the +connection—or while it is disconnected—are not replayed. + +Because RustQueue is share-nothing, clients first call Discovery `/lookup`, +then open one WebSocket to the Broker HTTP address for each healthy owner. A +client can union those streams and use the broker-scoped message ID in the +binary frame for migration-time deduplication. The producer proxy is not a +WebSocket fan-out layer. + +### Authentication and browser policy + +WebSocket authentication reuses the existing subscription permission and +checks the virtual Channel `websocket#ephemeral`. Service clients may send +`Authorization: Bearer ` in the HTTP upgrade request. Browser clients +may complete the `auth_required` / `auth` text-control exchange; credentials +must never be placed in a query parameter. Browser requests with an `Origin` +header must exactly match `websocket.allowed_origins`; a request without an +Origin is treated as a non-browser client and still goes through normal +authentication. Discovery's read-only `/lookup` endpoint applies the same +allowed-origin list. + +### Binary data frame + +Every delivered message is one complete binary WebSocket frame. The first 24 +bytes are big-endian metadata, followed by the raw message body: + +| Offset | Size | Field | +| --- | ---: | --- | +| `0..4` | 4 | ASCII magic `RQW1` | +| `4..6` | 2 | Header length, fixed at `24` | +| `6..8` | 2 | Delivery attempts | +| `8..16` | 8 | Broker timestamp, signed `i64` nanoseconds | +| `16..24` | 8 | Broker-scoped message ID, unsigned `u64` | +| `24..` | — | Original message body | + +Text frames are control messages. Clients should handle `auth_required`, +`ready`, and stable error codes and ignore unknown fields for forward +compatibility. There is no WebSocket ACK: a message is considered delivered by +RustQueue only after the complete frame is written, which does not mean the +client has processed it. + +The HTTP `101 Switching Protocols` response establishes only the transport. A +live subscription begins at the server's `{"type":"ready",...}` control +frame. No message capture is promised before `ready`, including while the +server is waiting for the `auth` control frame; clients must not interpret the +upgrade or `auth_required` frame as the start boundary. + +### Stable v1 error codes + +Handshake failures reuse the HTTP `ApiError` shape: the stable code is in the +JSON `message` field and the human-readable explanation is in `detail`. After +upgrade, terminal errors are sent as a text control such as +`{"type":"error","code":"E_TOPIC_CLOSED","detail":"..."}` and then the +connection is closed. Clients should reconnect or repeat `/lookup` according +to the code: + +| Code | Meaning | +| --- | --- | +| `E_WS_QUERY_FORBIDDEN` | Query parameters are not accepted; use the path and headers only. | +| `E_WS_ORIGIN` | The browser `Origin` is not an exact configured allowlist entry. | +| `E_WS_SUBPROTOCOL` | The upgrade must include `rustqueue.live.v1`. | +| `E_BAD_TOPIC` | The Topic does not exist on this Broker. | +| `E_WS_TTL_REQUIRED` | The existing Topic is not `TTL_DISCARD` or changed away from it. | +| `E_WS_CAPACITY` | Connection, per-Topic, Channel or frame-memory capacity is exhausted; back off. | +| `E_WS_CONTROL` | An unsupported client text/binary data frame was received. | +| `E_WS_HEARTBEAT` | The peer missed the heartbeat liveness window; reconnect. | +| `E_AUTH_TIMEOUT` | The required `auth` control was not received before the AUTH deadline. | +| `E_BAD_AUTH` | The AUTH control was non-text, malformed, or missing a valid `auth`/`secret` payload. | +| `E_AUTH_OVERLOADED` | The AUTH service or its capacity gate is exhausted; retry later. | +| `E_AUTH_FAILED` | The authentication service or credential validation failed. | +| `E_UNAUTHORIZED` | AUTH succeeded but does not permit this Topic subscription. | +| `E_TOPIC_CLOSED` | The Topic was deleted, tombstoned, or otherwise closed. | +| `E_OWNER_CHANGED` | The Topic owner/fence changed; repeat Discovery `/lookup`. | +| `E_DRAINING` | The Broker is draining and will not keep the live subscription. | +| `E_SHUTDOWN` | The Broker is shutting down; reconnect after a healthy owner is discovered. | + +The stream deliberately has no reliable backlog. At most one message is +reserved per connection. A failed or timed-out write cancels that reservation +and closes the connection; RustQueue does not retry the frame over WebSocket. +The write deadline is the smaller of the socket timeout and the message's +remaining TTL. Paused Topics stop taking messages. Topic deletion, migration, +owner shutdown, or a change to `RELIABLE` closes affected connections so the +client can perform `/lookup` again. + +The node-wide frame budget and connection limits are bounded by this default +runtime configuration: + +```toml +[websocket] +enabled = false +max_connections = 1024 +max_connections_per_topic = 256 +frame_inflight_bytes = 134217728 +allowed_origins = [] +``` + +The same fields can be overridden for a Broker with +`RUSTQUEUE_WEBSOCKET_ENABLED`, `RUSTQUEUE_WEBSOCKET_MAX_CONNECTIONS`, +`RUSTQUEUE_WEBSOCKET_MAX_CONNECTIONS_PER_TOPIC`, +`RUSTQUEUE_WEBSOCKET_FRAME_INFLIGHT_BYTES`, and +`RUSTQUEUE_WEBSOCKET_ALLOWED_ORIGINS` (a comma-separated list of exact HTTP +origins). Wildcards and origins containing a path are rejected. + +Helm values, the RustQueue Operator spec, and the CRD expose the same settings +through their Kubernetes naming conventions. Enabling the feature reuses the +existing Broker HTTP Service and NetworkPolicy; it does not add a port or an +Ingress. Monitor connection count, completed frame messages/bytes, +authentication failures, capacity rejections, non-TTL rejections and slow +consumer closes before increasing the limits. + ## Local development The host does not need Rust. Builds and tests use the pinned Docker toolchain: @@ -240,7 +354,7 @@ kubectl label node worker-1 rustqueue.io/eligible=true helm upgrade --install rustqueue deploy/helm/rustqueue \ --namespace rustqueue --create-namespace \ - --set queue.image=registry.example/rustqueue:0.9.0 \ + --set queue.image=registry.example/rustqueue:0.10.0 \ --set queue.storageClassName=ssd-rwo ``` @@ -254,7 +368,7 @@ scrub, upgrade or the NSQ-compatible admin API. Client TLS is optional and always supplied through an existing Kubernetes Secret; the operator does not run a CA. -### Kodo compatibility in 0.9 +### Kodo compatibility Kodo compatibility is implemented entirely by RustQueue as a separate, default-off deployment profile. Kodo continues to use RustQueue Discovery: @@ -477,7 +591,7 @@ test-only direct Pod placement; production anti-affinity is unchanged. A unit fixture covers discovery indexing for 500 brokers. No 500-broker deployment or load test is part of the functional gate. -The v0.9.0 CI/CD workflow publishes a Release only after the non-Kubernetes +The v0.10.0 CI/CD workflow publishes a Release only after the non-Kubernetes release gate, both native Linux builds, packaging and checksum verification succeed. The v0.8.0 Kodo compatibility baseline additionally passed the unmodified Kodo source replay, an exact 104,857,500-byte `PUB`/`DPUB` with one @@ -534,6 +648,8 @@ Native broker endpoints: - `GET /v1/registry` - `GET /v1/registry/head` - `GET /v1/capabilities` +- `GET /v1/ws/topics/{topic}` (WebSocket, only when enabled and only for + `TTL_DISCARD` Topics; negotiate `rustqueue.live.v1`) - `GET|POST /v1/drain` - `GET /v1/stats` - `GET /v1/observe` (console token, no message bodies) @@ -580,7 +696,7 @@ writes. Keep both relaxed profiles separate from durable-PUB results. ## Storage and upgrades -RustQueue 0.9 keeps disk format v7. Format v7 is a clean break: a v6 or older +RustQueue 0.10 keeps disk format v7. Format v7 is a clean break: a v6 or older directory is refused and there is no in-place migration. Within v7, record tags and existing fields are append-only. Every binary declares its reader/writer feature range and protocol message/body @@ -653,6 +769,7 @@ non-Kubernetes gate runs in GitHub Actions. ## Non-goals -RustQueue 0.9 does not provide message replication, backups, exactly-once +RustQueue 0.10 does not provide message replication, backups, exactly-once delivery, a global channel catalog, online data migration, cross-region -replication, or Broker/PVC lifecycle controls in Console. +replication, reliable WebSocket replay/resume/ACK semantics, or Broker/PVC +lifecycle controls in Console. diff --git a/console-ui/package.json b/console-ui/package.json index 2f126e4..04922f5 100644 --- a/console-ui/package.json +++ b/console-ui/package.json @@ -1,7 +1,7 @@ { "name": "rustqueue-console-ui", "private": true, - "version": "0.9.0", + "version": "0.10.0", "packageManager": "pnpm@11.9.0", "type": "module", "scripts": { diff --git a/console-ui/src/layout/AppShell.tsx b/console-ui/src/layout/AppShell.tsx index 04654c8..41dfa02 100644 --- a/console-ui/src/layout/AppShell.tsx +++ b/console-ui/src/layout/AppShell.tsx @@ -57,7 +57,7 @@ export function AppShell({ page, onPage, dark, onTheme, onRefresh, children }: { onClick={(event) => { event.preventDefault(); onPage(item.id); }} >{t(`nav.${item.id}`)})} -
{t('app.readOnly')}v0.9.0
+
{t('app.readOnly')}v0.10.0
{children}
diff --git a/crates/discovery/Cargo.toml b/crates/discovery/Cargo.toml index 36bfa38..070e18e 100644 --- a/crates/discovery/Cargo.toml +++ b/crates/discovery/Cargo.toml @@ -25,6 +25,7 @@ sha2.workspace = true tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true +tower-http.workspace = true [dev-dependencies] tower.workspace = true diff --git a/crates/discovery/src/main.rs b/crates/discovery/src/main.rs index 2a439f2..3841b95 100644 --- a/crates/discovery/src/main.rs +++ b/crates/discovery/src/main.rs @@ -42,6 +42,12 @@ struct Cli { kodo_gateway_address: Option, #[arg(long, env = "RUSTQUEUE_KODO_CLEANUP_ENABLED", default_value_t = false)] kodo_cleanup_enabled: bool, + #[arg( + long, + env = "RUSTQUEUE_WEBSOCKET_ALLOWED_ORIGINS", + value_delimiter = ',' + )] + websocket_allowed_origins: Vec, } #[tokio::main] @@ -56,6 +62,24 @@ async fn main() -> anyhow::Result<()> { if cli.endpoint_slice_timeout_ms == 0 { anyhow::bail!("EndpointSlice timeout must be greater than zero"); } + if cli.websocket_allowed_origins.iter().any(|origin| { + let authority = origin + .strip_prefix("http://") + .or_else(|| origin.strip_prefix("https://")); + origin == "*" + || authority.is_none_or(|authority| { + authority.is_empty() + || authority.contains('/') + || authority.contains('?') + || authority.contains('#') + || authority.contains(',') + || authority.trim() != authority + }) + }) { + anyhow::bail!( + "WebSocket allowed origins must be exact HTTP origins without paths or wildcards" + ); + } if cli.kodo_cleanup_enabled { anyhow::bail!( "Kodo automatic cleanup is disabled until cluster-wide atomic deletion is available" @@ -103,6 +127,6 @@ async fn main() -> anyhow::Result<()> { ); tokio::select! { result = refresh => result, - result = serve(cli.address, directory) => result, + result = serve(cli.address, directory, cli.websocket_allowed_origins) => result, } } diff --git a/crates/discovery/src/server.rs b/crates/discovery/src/server.rs index 0beb498..0238190 100644 --- a/crates/discovery/src/server.rs +++ b/crates/discovery/src/server.rs @@ -9,6 +9,8 @@ use axum::{Json, Router}; use serde::Deserialize; use serde_json::{json, Value}; use std::net::SocketAddr; +use std::time::Duration; +use tower_http::cors::{AllowOrigin, CorsLayer}; #[derive(Deserialize)] struct TopicQuery { @@ -22,10 +24,29 @@ struct ChannelQuery { } pub fn router(directory: Directory) -> Router { + router_with_origins(directory, &[]) +} + +pub fn router_with_origins(directory: Directory, allowed_origins: &[String]) -> Router { + let lookup_route = if allowed_origins.is_empty() { + get(lookup) + } else { + let origins = allowed_origins + .iter() + .filter_map(|origin| HeaderValue::from_str(origin).ok()) + .collect::>(); + get(lookup).layer( + CorsLayer::new() + .allow_origin(AllowOrigin::list(origins)) + .allow_methods([axum::http::Method::GET]) + .allow_headers([header::ACCEPT]) + .max_age(Duration::from_secs(600)), + ) + }; Router::new() .route("/ping", get(|| async { "OK" })) .route("/info", get(info)) - .route("/lookup", get(lookup)) + .route("/lookup", lookup_route) .route("/topics", get(topics)) .route("/channels", get(channels)) .route("/nodes", get(nodes)) @@ -82,10 +103,14 @@ async fn nsq_content_negotiation(request: Request, next: Next) -> Response response } -pub async fn serve(address: SocketAddr, directory: Directory) -> anyhow::Result<()> { +pub async fn serve( + address: SocketAddr, + directory: Directory, + allowed_origins: Vec, +) -> anyhow::Result<()> { let listener = tokio::net::TcpListener::bind(address).await?; tracing::info!(%address, "RustQueue discovery API listening"); - axum::serve(listener, router(directory)).await?; + axum::serve(listener, router_with_origins(directory, &allowed_origins)).await?; Ok(()) } diff --git a/crates/operator/src/controller/mod.rs b/crates/operator/src/controller/mod.rs index 29deb73..0aa4fd5 100644 --- a/crates/operator/src/controller/mod.rs +++ b/crates/operator/src/controller/mod.rs @@ -1380,6 +1380,32 @@ fn validate(cluster: &RustQueue, active_feature_floor: u32) -> anyhow::Result<() ) { bail!("publishAckMode must be durable, write_ack, or nsq_relaxed"); } + if cluster.spec.websocket.max_connections == 0 + || cluster.spec.websocket.max_connections > tokio::sync::Semaphore::MAX_PERMITS + || cluster.spec.websocket.max_connections_per_topic == 0 + || cluster.spec.websocket.max_connections_per_topic > cluster.spec.websocket.max_connections + || cluster.spec.websocket.frame_inflight_bytes + < cluster.spec.max_message_bytes.saturating_add(24) + || cluster.spec.websocket.frame_inflight_bytes > u32::MAX as usize + { + bail!("WebSocket limits must fit one maximum message and satisfy per-Topic <= global connections"); + } + if cluster.spec.websocket.allowed_origins.iter().any(|origin| { + let authority = origin + .strip_prefix("http://") + .or_else(|| origin.strip_prefix("https://")); + origin == "*" + || authority.is_none_or(|authority| { + authority.is_empty() + || authority.contains('/') + || authority.contains('?') + || authority.contains('#') + || authority.contains(',') + || authority.trim() != authority + }) + }) { + bail!("WebSocket allowedOrigins must be exact HTTP origins without paths or wildcards"); + } validate_message_storage_contract( cluster.spec.max_message_bytes, effective_storage_feature_level(cluster.spec.storage_feature_level, active_feature_floor), diff --git a/crates/operator/src/controller/preflight.rs b/crates/operator/src/controller/preflight.rs index 68d6d8e..e87e005 100644 --- a/crates/operator/src/controller/preflight.rs +++ b/crates/operator/src/controller/preflight.rs @@ -28,6 +28,8 @@ struct BinaryCapabilities { maximum_message_bytes: usize, #[serde(default = "legacy_maximum_batch_bytes")] maximum_batch_bytes: usize, + #[serde(default)] + websocket_ttl_live_v1: bool, } #[derive(Clone, Debug, Deserialize)] @@ -142,6 +144,7 @@ pub(super) async fn target_image( cluster.spec.storage_feature_level, cluster.spec.max_message_bytes, required_batch_bytes(cluster), + cluster.spec.websocket.enabled, ) .map_or_else( |message| Ok(Outcome::Blocked(message)), @@ -279,6 +282,7 @@ fn validate_binary( desired: u32, message_bytes: usize, batch_bytes: usize, + websocket_required: bool, ) -> Result<(), String> { if capabilities.binary_version.trim().is_empty() || capabilities.data_format != DATA_FORMAT { return Err("target image does not advertise RustQueue format v7".into()); @@ -299,6 +303,9 @@ fn validate_binary( capabilities.maximum_message_bytes, capabilities.maximum_batch_bytes )); } + if websocket_required && !capabilities.websocket_ttl_live_v1 { + return Err("target image does not support WebSocket TTL live subscriptions".into()); + } Ok(()) } @@ -313,6 +320,7 @@ fn validate_report( desired.max(report.storage.active_writer_feature_level), message_bytes, batch_bytes, + false, )?; if report.storage.data_format != DATA_FORMAT { return Err("PVC compatibility state is not format v7".into()); @@ -341,13 +349,14 @@ fn legacy_maximum_batch_bytes() -> usize { fn probe_contract(cluster: &RustQueue, image: &str) -> String { format!( - "{}\0{}\0{}\0{}\0{}\0{}\0{}", + "{}\0{}\0{}\0{}\0{}\0{}\0{}\0{}", cluster.metadata.uid.as_deref().unwrap_or_default(), image, cluster.spec.image_pull_policy, cluster.spec.storage_feature_level, cluster.spec.max_message_bytes, required_batch_bytes(cluster), + cluster.spec.websocket.enabled, cluster.spec.rollout.retry_nonce, ) } @@ -462,13 +471,16 @@ mod tests { maximum_writer_feature_level: maximum, maximum_message_bytes: 100 * 1024 * 1024, maximum_batch_bytes: 128 * 1024 * 1024, + websocket_ttl_live_v1: true, } } #[test] fn target_image_must_support_the_requested_feature() { - assert!(validate_binary(&binary(2), 2, 100 * 1024 * 1024, 128 * 1024 * 1024).is_ok()); - assert!(validate_binary(&binary(1), 2, 20 * 1024 * 1024, 64 * 1024 * 1024).is_err()); + assert!( + validate_binary(&binary(2), 2, 100 * 1024 * 1024, 128 * 1024 * 1024, false).is_ok() + ); + assert!(validate_binary(&binary(1), 2, 20 * 1024 * 1024, 64 * 1024 * 1024, false).is_err()); } #[test] @@ -483,7 +495,19 @@ mod tests { .unwrap(); assert_eq!(legacy.maximum_message_bytes, legacy_maximum_message_bytes()); assert_eq!(legacy.maximum_batch_bytes, legacy_maximum_batch_bytes()); - assert!(validate_binary(&legacy, 2, 100 * 1024 * 1024, 128 * 1024 * 1024).is_err()); + assert!(validate_binary(&legacy, 2, 100 * 1024 * 1024, 128 * 1024 * 1024, false).is_err()); + } + + #[test] + fn websocket_capability_is_required_only_when_enabled() { + let mut capabilities = binary(2); + capabilities.websocket_ttl_live_v1 = false; + assert!( + validate_binary(&capabilities, 2, 20 * 1024 * 1024, 64 * 1024 * 1024, true,).is_err() + ); + assert!( + validate_binary(&capabilities, 2, 20 * 1024 * 1024, 64 * 1024 * 1024, false,).is_ok() + ); } #[test] diff --git a/crates/operator/src/controller/status.rs b/crates/operator/src/controller/status.rs index 3060f7f..a9a4f8d 100644 --- a/crates/operator/src/controller/status.rs +++ b/crates/operator/src/controller/status.rs @@ -225,6 +225,7 @@ mod tests { proxy_node_selector: BTreeMap::new(), proxy_tcp_max_connection_age_seconds: 300, discovery_replicas: 2, + websocket: crate::crd::WebSocketSpec::default(), kodo_compatibility: crate::crd::KodoCompatibility::default(), maintenance: None, rollout: RolloutPolicy::default(), diff --git a/crates/operator/src/crd.rs b/crates/operator/src/crd.rs index 74dec16..3005453 100644 --- a/crates/operator/src/crd.rs +++ b/crates/operator/src/crd.rs @@ -82,6 +82,8 @@ pub struct RustQueueSpec { #[serde(default = "default_discovery_replicas")] pub discovery_replicas: i32, #[serde(default)] + pub websocket: WebSocketSpec, + #[serde(default)] pub kodo_compatibility: KodoCompatibility, #[serde(default)] pub maintenance: Option, @@ -93,6 +95,33 @@ pub struct RustQueueSpec { pub broker_resources: WorkloadResources, } +#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)] +#[serde(rename_all = "camelCase")] +pub struct WebSocketSpec { + #[serde(default)] + pub enabled: bool, + #[serde(default = "default_websocket_max_connections")] + pub max_connections: usize, + #[serde(default = "default_websocket_max_connections_per_topic")] + pub max_connections_per_topic: usize, + #[serde(default = "default_websocket_frame_inflight_bytes")] + pub frame_inflight_bytes: usize, + #[serde(default)] + pub allowed_origins: Vec, +} + +impl Default for WebSocketSpec { + fn default() -> Self { + Self { + enabled: false, + max_connections: default_websocket_max_connections(), + max_connections_per_topic: default_websocket_max_connections_per_topic(), + frame_inflight_bytes: default_websocket_frame_inflight_bytes(), + allowed_origins: Vec::new(), + } + } +} + #[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)] #[serde(rename_all = "camelCase")] pub struct KodoCompatibility { @@ -375,6 +404,15 @@ fn default_max_detailed_metric_series() -> usize { fn default_discovery_replicas() -> i32 { 2 } +fn default_websocket_max_connections() -> usize { + 1_024 +} +fn default_websocket_max_connections_per_topic() -> usize { + 256 +} +fn default_websocket_frame_inflight_bytes() -> usize { + 128 * 1024 * 1024 +} fn default_rollout_timeout() -> u64 { 600 } diff --git a/crates/operator/src/lib.rs b/crates/operator/src/lib.rs index 08d3a3b..eda579f 100644 --- a/crates/operator/src/lib.rs +++ b/crates/operator/src/lib.rs @@ -5,7 +5,7 @@ pub mod resources; pub use crd::{ BrokerMaintenance, BrokerScheduling, BrokerToleration, KodoCompatibility, OperationStatus, - RolloutPolicy, RustQueue, RustQueueCondition, RustQueueSpec, RustQueueStatus, + RolloutPolicy, RustQueue, RustQueueCondition, RustQueueSpec, RustQueueStatus, WebSocketSpec, WorkloadResources, }; pub use management_crd::{ diff --git a/crates/operator/src/resources.rs b/crates/operator/src/resources.rs index 29229d8..01bd6f8 100644 --- a/crates/operator/src/resources.rs +++ b/crates/operator/src/resources.rs @@ -255,6 +255,12 @@ pub fn build(input: BuildInput<'_>) -> anyhow::Result { json!({"name": "RUSTQUEUE_BROKER_SERVICE", "value": broker_service_name}), json!({"name": "RUSTQUEUE_REGISTRY_TOKEN_FILE", "value": "/run/secrets/rustqueue/registry-token"}), ]; + if !cluster.spec.websocket.allowed_origins.is_empty() { + discovery_env.push(json!({ + "name": "RUSTQUEUE_WEBSOCKET_ALLOWED_ORIGINS", + "value": cluster.spec.websocket.allowed_origins.join(",") + })); + } if input.advertise_kodo_gateways { let cleanup_enabled = input.activate_kodo_cleanup && cluster.spec.kodo_compatibility.effective_cleanup_enabled(); @@ -533,8 +539,26 @@ fn broker_config(cluster: &RustQueue, secret_name: &str) -> String { } else { "" }; + let websocket_origins = cluster + .spec + .websocket + .allowed_origins + .iter() + .map(|origin| serde_json::to_string(origin).expect("origin is serializable")) + .collect::>() + .join(", "); + let websocket = if cluster.spec.websocket.enabled { + format!( + "[websocket]\nenabled = true\nmax_connections = {}\nmax_connections_per_topic = {}\nframe_inflight_bytes = {}\nallowed_origins = [{websocket_origins}]\n\n", + cluster.spec.websocket.max_connections, + cluster.spec.websocket.max_connections_per_topic, + cluster.spec.websocket.frame_inflight_bytes, + ) + } else { + String::new() + }; let mut output = format!( - "{kodo_network}[storage]\ndata_path = \"/data\"\nfeature_level = {}\nmax_segment_bytes = {max_segment_bytes}\nmin_free_bytes = {}\ndisk_high_watermark_percent = {}\ndisk_low_watermark_percent = {}\nprotective_eviction_enabled = {}\ndisk_pressure_grace_seconds = {}\nmaintenance_startup_delay_seconds = {}\n\n[queue]\nbootstrap_retention_seconds = {}\nmax_message_bytes = {}\nmax_topics = {}\nmax_publish_workers = {}\npublish_worker_idle_seconds = {}\npublish_ack_mode = \"{}\"\nrelaxed_sync_messages = {}\nrelaxed_sync_bytes = {}\nrelaxed_sync_interval_ms = {}\n\n[limits]\nmax_body_bytes = {max_body_bytes}\nnode_publish_inflight_bytes = {node_publish_inflight_bytes}\nconnection_publish_inflight_bytes = {connection_publish_inflight_bytes}\nnode_delivery_inflight_bytes = {}\nconnection_delivery_inflight_bytes = {}\ndisconnect_on_retriable_publish_error = false\n\n[metrics]\ndetailed_queue_metrics = {}\nmax_detailed_series = {}\n\n[security]\nadmin_token_file = \"/run/secrets/rustqueue/admin-token\"\n{publish_token}registry_token_file = \"/run/secrets/rustqueue/registry-token\"\nconsole_token_file = \"/run/secrets/rustqueue/console-token\"\n{kodo_cleanup_token}console_management_enabled = {}\nkodo_cleanup_enabled = {}\n# secret: {secret_name}\n", + "{kodo_network}{websocket}[storage]\ndata_path = \"/data\"\nfeature_level = {}\nmax_segment_bytes = {max_segment_bytes}\nmin_free_bytes = {}\ndisk_high_watermark_percent = {}\ndisk_low_watermark_percent = {}\nprotective_eviction_enabled = {}\ndisk_pressure_grace_seconds = {}\nmaintenance_startup_delay_seconds = {}\n\n[queue]\nbootstrap_retention_seconds = {}\nmax_message_bytes = {}\nmax_topics = {}\nmax_publish_workers = {}\npublish_worker_idle_seconds = {}\npublish_ack_mode = \"{}\"\nrelaxed_sync_messages = {}\nrelaxed_sync_bytes = {}\nrelaxed_sync_interval_ms = {}\n\n[limits]\nmax_body_bytes = {max_body_bytes}\nnode_publish_inflight_bytes = {node_publish_inflight_bytes}\nconnection_publish_inflight_bytes = {connection_publish_inflight_bytes}\nnode_delivery_inflight_bytes = {}\nconnection_delivery_inflight_bytes = {}\ndisconnect_on_retriable_publish_error = false\n\n[metrics]\ndetailed_queue_metrics = {}\nmax_detailed_series = {}\n\n[security]\nadmin_token_file = \"/run/secrets/rustqueue/admin-token\"\n{publish_token}registry_token_file = \"/run/secrets/rustqueue/registry-token\"\nconsole_token_file = \"/run/secrets/rustqueue/console-token\"\n{kodo_cleanup_token}console_management_enabled = {}\nkodo_cleanup_enabled = {}\n# secret: {secret_name}\n", cluster.spec.storage_feature_level, cluster.spec.min_free_bytes, cluster.spec.disk_high_watermark_percent, @@ -666,6 +690,7 @@ mod tests { proxy_node_selector: BTreeMap::new(), proxy_tcp_max_connection_age_seconds: 300, discovery_replicas: 2, + websocket: crate::crd::WebSocketSpec::default(), kodo_compatibility: crate::crd::KodoCompatibility::default(), maintenance: None, rollout: crate::crd::RolloutPolicy::default(), @@ -847,6 +872,28 @@ mod tests { ); } + #[test] + fn websocket_broker_config_is_opt_in_and_renders_limits_and_origins() { + let default_config = broker_config(&cluster(), "queue-auth"); + assert!(!default_config.contains("[websocket]")); + + let mut websocket_cluster = cluster(); + websocket_cluster.spec.websocket.enabled = true; + websocket_cluster.spec.websocket.allowed_origins = vec![ + "https://console.example.test".into(), + "http://127.0.0.1:14151".into(), + ]; + let websocket_config = broker_config(&websocket_cluster, "queue-auth"); + + assert!(websocket_config.contains("[websocket]\nenabled = true")); + assert!(websocket_config.contains("max_connections = 1024")); + assert!(websocket_config.contains("max_connections_per_topic = 256")); + assert!(websocket_config.contains("frame_inflight_bytes = 134217728")); + assert!(websocket_config.contains( + "allowed_origins = [\"https://console.example.test\", \"http://127.0.0.1:14151\"]" + )); + } + #[test] fn kodo_profile_renders_three_stable_gateways_and_large_message_limits() { let mut cluster = cluster(); diff --git a/crates/queue/src/broker.rs b/crates/queue/src/broker.rs index bd9eafa..a41f6f4 100644 --- a/crates/queue/src/broker.rs +++ b/crates/queue/src/broker.rs @@ -466,6 +466,23 @@ impl Broker { } pub async fn create_channel(&self, topic: &str, channel: &str) -> Result<(), BrokerError> { + self.create_channel_inner(topic, channel, true).await + } + + pub async fn create_channel_existing( + &self, + topic: &str, + channel: &str, + ) -> Result<(), BrokerError> { + self.create_channel_inner(topic, channel, false).await + } + + async fn create_channel_inner( + &self, + topic: &str, + channel: &str, + create_topic: bool, + ) -> Result<(), BrokerError> { validate_name(topic).map_err(|_| BrokerError::InvalidTopic)?; validate_channel(channel)?; self.ensure_management_access(topic, Some(channel))?; @@ -475,7 +492,11 @@ impl Broker { self.storage_task(move || { let _lifecycle = broker.inner.topic_lifecycle.lock(); broker.ensure_management_access(&topic, Some(&channel))?; - let handle = broker.get_or_create_topic_locked(&topic)?; + let handle = if create_topic { + broker.get_or_create_topic_locked(&topic)? + } else { + broker.topic(&topic)? + }; let _commit_gate = handle.commit_gate.lock(); let _channel_commit_gate = handle.channel_commit_gate.lock(); if handle diff --git a/crates/queue/src/broker/expiration.rs b/crates/queue/src/broker/expiration.rs index 3358b2e..be68d2a 100644 --- a/crates/queue/src/broker/expiration.rs +++ b/crates/queue/src/broker/expiration.rs @@ -131,6 +131,10 @@ impl Broker { Ok(self.topic(topic)?.subscribe_policy()) } + pub fn topic_reliable_policy_epoch(&self, topic: &str) -> Result { + Ok(self.topic(topic)?.reliable_policy_epoch()) + } + pub async fn configure_topic_policy( &self, topic: &str, diff --git a/crates/queue/src/broker_tests.rs b/crates/queue/src/broker_tests.rs index e8ffa2a..b75152b 100644 --- a/crates/queue/src/broker_tests.rs +++ b/crates/queue/src/broker_tests.rs @@ -1428,6 +1428,24 @@ async fn broker_rejects_unbounded_durable_topic_creation() { assert_eq!(broker.topic_names(), vec!["first"]); } +#[tokio::test] +async fn create_channel_existing_rejects_missing_topic_without_creating_it() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + ..BrokerConfig::default() + }) + .unwrap(); + + let error = broker + .create_channel_existing("events", "workers") + .await + .unwrap_err(); + + assert!(matches!(error, BrokerError::TopicNotFound)); + assert!(broker.topic_names().is_empty()); +} + #[tokio::test] async fn sealed_backlog_uses_bounded_metadata_residency() { let root = tempdir().unwrap(); @@ -2940,6 +2958,81 @@ async fn ttl_policy_changes_apply_to_backlog_without_reviving_expired_messages() assert_eq!(broker.stats().topics[0].ttl_discarded_messages, 1); } +#[tokio::test] +async fn ttl_policy_watch_keeps_reliable_epoch_across_coalesced_changes() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + ..BrokerConfig::default() + }) + .unwrap(); + broker.create_topic("policy-watch").await.unwrap(); + broker + .configure_topic_policy( + "policy-watch", + TopicPolicy { + delivery_mode: DeliveryMode::TtlDiscard, + message_ttl_seconds: Some(30), + }, + ) + .await + .unwrap(); + + let mut policy_changes = broker.subscribe_topic_policy("policy-watch").unwrap(); + let initial_epoch = broker.topic_reliable_policy_epoch("policy-watch").unwrap(); + + broker + .configure_topic_policy( + "policy-watch", + TopicPolicy { + delivery_mode: DeliveryMode::TtlDiscard, + message_ttl_seconds: Some(60), + }, + ) + .await + .unwrap(); + assert_eq!( + broker.topic_reliable_policy_epoch("policy-watch").unwrap(), + initial_epoch + ); + policy_changes.changed().await.unwrap(); + assert_eq!( + *policy_changes.borrow_and_update(), + TopicPolicy { + delivery_mode: DeliveryMode::TtlDiscard, + message_ttl_seconds: Some(60), + } + ); + + broker + .configure_topic_policy("policy-watch", TopicPolicy::default()) + .await + .unwrap(); + broker + .configure_topic_policy( + "policy-watch", + TopicPolicy { + delivery_mode: DeliveryMode::TtlDiscard, + message_ttl_seconds: Some(90), + }, + ) + .await + .unwrap(); + + policy_changes.changed().await.unwrap(); + assert_eq!( + *policy_changes.borrow_and_update(), + TopicPolicy { + delivery_mode: DeliveryMode::TtlDiscard, + message_ttl_seconds: Some(90), + } + ); + assert_eq!( + broker.topic_reliable_policy_epoch("policy-watch").unwrap(), + initial_epoch + 1 + ); +} + #[tokio::test] async fn ttl_discard_evicts_durable_ephemeral_deferred_requeued_and_in_flight_state() { let root = tempdir().unwrap(); diff --git a/crates/queue/src/topic.rs b/crates/queue/src/topic.rs index 0ed3270..9016d4a 100644 --- a/crates/queue/src/topic.rs +++ b/crates/queue/src/topic.rs @@ -22,6 +22,7 @@ use rustqueue_storage::{RecordHeader, RecordKind, SegmentLog}; use std::collections::HashMap; use std::fs::File; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -42,6 +43,7 @@ pub(crate) struct TopicHandle { pub state: Mutex, pub wake: tokio::sync::watch::Sender, policy_changes: tokio::sync::watch::Sender, + reliable_policy_epoch: AtomicU64, } pub(crate) struct Topic { @@ -97,6 +99,7 @@ impl TopicHandle { state: Mutex::new(topic), wake, policy_changes, + reliable_policy_epoch: AtomicU64::new(0), })) } @@ -154,6 +157,7 @@ impl TopicHandle { }), wake, policy_changes, + reliable_policy_epoch: AtomicU64::new(0), })) } @@ -165,6 +169,9 @@ impl TopicHandle { pub fn set_policy(&self, topic: &mut Topic, policy: TopicPolicy) -> Result { let changed = topic.set_policy(policy)?; if changed { + if policy.delivery_mode == DeliveryMode::Reliable { + self.reliable_policy_epoch.fetch_add(1, Ordering::AcqRel); + } self.policy_changes.send_replace(policy); self.signal(); } @@ -174,6 +181,10 @@ impl TopicHandle { pub fn subscribe_policy(&self) -> tokio::sync::watch::Receiver { self.policy_changes.subscribe() } + + pub fn reliable_policy_epoch(&self) -> u64 { + self.reliable_policy_epoch.load(Ordering::Acquire) + } } impl Topic { diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index 6ee62c1..0d94352 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -19,6 +19,7 @@ pub struct RuntimeCapabilities { storage: rustqueue_storage::BinaryCapabilities, pub maximum_message_bytes: usize, pub maximum_batch_bytes: usize, + pub websocket_ttl_live_v1: bool, } pub fn runtime_capabilities() -> RuntimeCapabilities { @@ -26,6 +27,7 @@ pub fn runtime_capabilities() -> RuntimeCapabilities { storage: rustqueue_storage::binary_capabilities(), maximum_message_bytes: MAX_SUPPORTED_MESSAGE_BYTES, maximum_batch_bytes: MAX_SUPPORTED_BATCH_BYTES, + websocket_ttl_live_v1: true, } } @@ -34,6 +36,7 @@ pub fn runtime_capabilities() -> RuntimeCapabilities { pub struct Config { pub node: NodeConfig, pub network: NetworkConfig, + pub websocket: WebSocketConfig, pub storage: StorageConfig, pub queue: QueueConfig, pub security: SecurityConfig, @@ -43,6 +46,16 @@ pub struct Config { pub log_format: String, } +#[derive(Clone, Debug, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct WebSocketConfig { + pub enabled: bool, + pub max_connections: usize, + pub max_connections_per_topic: usize, + pub frame_inflight_bytes: usize, + pub allowed_origins: Vec, +} + #[derive(Clone, Debug, Deserialize)] #[serde(default, deny_unknown_fields)] pub struct NodeConfig { @@ -174,6 +187,7 @@ impl Default for Config { Self { node: NodeConfig::default(), network: NetworkConfig::default(), + websocket: WebSocketConfig::default(), storage: StorageConfig::default(), queue: QueueConfig::default(), security: SecurityConfig::default(), @@ -185,6 +199,18 @@ impl Default for Config { } } +impl Default for WebSocketConfig { + fn default() -> Self { + Self { + enabled: false, + max_connections: 1_024, + max_connections_per_topic: 256, + frame_inflight_bytes: 128 * 1024 * 1024, + allowed_origins: Vec::new(), + } + } +} + impl Default for NodeConfig { fn default() -> Self { Self { @@ -403,6 +429,31 @@ impl Config { if !(1..=9).contains(&self.network.max_deflate_level) { bail!("network.max_deflate_level must be in 1..=9"); } + if self.websocket.max_connections == 0 + || self.websocket.max_connections > tokio::sync::Semaphore::MAX_PERMITS + || self.websocket.max_connections_per_topic == 0 + || self.websocket.max_connections_per_topic > self.websocket.max_connections + || self.websocket.frame_inflight_bytes < self.queue.max_message_bytes.saturating_add(24) + || self.websocket.frame_inflight_bytes > u32::MAX as usize + { + bail!("websocket connection limits must be non-zero and frame_inflight_bytes must fit at least one maximum message and u32"); + } + if self.websocket.allowed_origins.iter().any(|origin| { + let authority = origin + .strip_prefix("http://") + .or_else(|| origin.strip_prefix("https://")); + origin == "*" + || authority.is_none_or(|authority| { + authority.is_empty() + || authority.contains('/') + || authority.contains('?') + || authority.contains('#') + || authority.contains(',') + || authority.trim() != authority + }) + }) { + bail!("websocket.allowed_origins must contain exact HTTP origins without paths or wildcards"); + } if self.security.kodo_cleanup_enabled { bail!( "Kodo automatic cleanup is disabled until cluster-wide atomic deletion is available" diff --git a/crates/server/src/config/environment.rs b/crates/server/src/config/environment.rs index deeff37..e692912 100644 --- a/crates/server/src/config/environment.rs +++ b/crates/server/src/config/environment.rs @@ -27,6 +27,27 @@ impl Config { )?; set_from_env("RUSTQUEUE_TCP_ADDRESS", &mut self.network.tcp_address)?; set_from_env("RUSTQUEUE_HTTP_ADDRESS", &mut self.network.http_address)?; + set_from_env("RUSTQUEUE_WEBSOCKET_ENABLED", &mut self.websocket.enabled)?; + set_from_env( + "RUSTQUEUE_WEBSOCKET_MAX_CONNECTIONS", + &mut self.websocket.max_connections, + )?; + set_from_env( + "RUSTQUEUE_WEBSOCKET_MAX_CONNECTIONS_PER_TOPIC", + &mut self.websocket.max_connections_per_topic, + )?; + set_from_env( + "RUSTQUEUE_WEBSOCKET_FRAME_INFLIGHT_BYTES", + &mut self.websocket.frame_inflight_bytes, + )?; + if let Ok(value) = env::var("RUSTQUEUE_WEBSOCKET_ALLOWED_ORIGINS") { + self.websocket.allowed_origins = value + .split(',') + .map(str::trim) + .filter(|origin| !origin.is_empty()) + .map(str::to_owned) + .collect(); + } if let Ok(value) = env::var("RUSTQUEUE_KODO_HTTP_ADDRESS") { self.network.kodo_http_address = Some( value diff --git a/crates/server/src/config/tests.rs b/crates/server/src/config/tests.rs index 4132b77..3f3d2ef 100644 --- a/crates/server/src/config/tests.rs +++ b/crates/server/src/config/tests.rs @@ -66,6 +66,19 @@ fn binary_capabilities_advertise_protocol_limits() { ); } +#[test] +fn websocket_defaults_are_opt_in_and_capability_is_advertised() { + let config = Config::default(); + assert!(!config.websocket.enabled); + assert_eq!(config.websocket.max_connections, 1_024); + assert_eq!(config.websocket.max_connections_per_topic, 256); + assert_eq!(config.websocket.frame_inflight_bytes, 128 * 1024 * 1024); + assert!(config.websocket.allowed_origins.is_empty()); + + let capabilities = serde_json::to_value(runtime_capabilities()).unwrap(); + assert_eq!(capabilities["websocket_ttl_live_v1"], true); +} + #[test] fn kodo_cleanup_is_disabled_until_cluster_wide_deletion_is_atomic() { let root = tempfile::tempdir().unwrap(); diff --git a/crates/server/src/http.rs b/crates/server/src/http.rs index 40bd6fe..eb8dec0 100644 --- a/crates/server/src/http.rs +++ b/crates/server/src/http.rs @@ -5,6 +5,7 @@ mod manage; mod native; mod nsq_stats; mod tokens; +mod websocket; use compat::*; use helpers::*; @@ -12,11 +13,14 @@ use manage::*; use native::*; use nsq_stats::*; use tokens::TokenSet; +use websocket::{lookup_cors, WebSocketRuntime}; use crate::admission::PublishAdmission; +use crate::auth::Authenticator; use crate::config::Config; use crate::metrics::Metrics; use crate::subscriptions::SubscriptionRegistry; +use crate::tcp::EphemeralConsumers; use axum::body::{Body, Bytes}; use axum::extract::{Query, State}; use axum::http::{header, HeaderMap, HeaderValue, Request, StatusCode}; @@ -45,6 +49,10 @@ struct AppState { delivering: Arc, publish_admission: Arc, subscriptions: SubscriptionRegistry, + authenticator: Option>, + ephemeral_consumers: EphemeralConsumers, + websocket: WebSocketRuntime, + shutdown: tokio::sync::watch::Receiver, started_at: i64, } @@ -115,6 +123,8 @@ pub async fn serve( delivering: Arc, publish_admission: Arc, subscriptions: SubscriptionRegistry, + authenticator: Option>, + ephemeral_consumers: EphemeralConsumers, shutdown: tokio::sync::watch::Receiver, ) -> anyhow::Result<()> { let state = app_state( @@ -125,14 +135,21 @@ pub async fn serve( delivering, publish_admission, subscriptions, + authenticator, + ephemeral_consumers, + shutdown.clone(), )?; + let lookup_route = match lookup_cors(&state.config.websocket.allowed_origins) { + Some(cors) => get(lookup).layer(cors), + None => get(lookup), + }; let mut router = Router::new() .route("/ping", get(ping)) .route("/info", get(info_handler)) .route("/pub", post(publish)) .route("/mpub", post(multi_publish)) .route("/stats", get(stats)) - .route("/lookup", get(lookup)) + .route("/lookup", lookup_route) .route("/topics", get(topics)) .route("/channels", get(channels)) .route("/nodes", get(nodes)) @@ -157,6 +174,9 @@ pub async fn serve( .route("/v1/observe/head", get(observe_head)) .route("/v1/storage/scrub", post(scrub)) .layer(middleware::from_fn(nsq_content_negotiation)); + if state.config.websocket.enabled { + router = router.route("/v1/ws/topics/{topic}", get(websocket::subscribe_topic)); + } if state.config.security.console_management_enabled { router = router .route("/v1/manage/topics/{action}", post(manage_topic)) @@ -178,9 +198,12 @@ pub async fn serve( let listener = TcpListener::bind(address).await?; info!(%address, "HTTP API listening"); let reloader = tokio::spawn(tokens.reload(token_shutdown)); - let result = axum::serve(listener, router) - .with_graceful_shutdown(wait_for_shutdown(shutdown)) - .await; + let result = axum::serve( + listener, + router.into_make_service_with_connect_info::(), + ) + .with_graceful_shutdown(wait_for_shutdown(shutdown)) + .await; reloader.abort(); result?; Ok(()) @@ -197,6 +220,7 @@ async fn wait_for_shutdown(mut shutdown: tokio::sync::watch::Receiver) { } } +#[allow(clippy::too_many_arguments)] fn app_state( config: Arc, broker: Arc, @@ -205,6 +229,9 @@ fn app_state( delivering: Arc, publish_admission: Arc, subscriptions: SubscriptionRegistry, + authenticator: Option>, + ephemeral_consumers: EphemeralConsumers, + shutdown: tokio::sync::watch::Receiver, ) -> anyhow::Result { Ok(AppState { tokens: TokenSet::from_config(&config)?, @@ -215,6 +242,10 @@ fn app_state( delivering, publish_admission, subscriptions, + authenticator, + ephemeral_consumers, + websocket: WebSocketRuntime::new(&config.websocket), + shutdown, started_at: unix_seconds(), }) } diff --git a/crates/server/src/http/kodo_compat.rs b/crates/server/src/http/kodo_compat.rs index 2e31846..f55286e 100644 --- a/crates/server/src/http/kodo_compat.rs +++ b/crates/server/src/http/kodo_compat.rs @@ -23,6 +23,9 @@ pub(crate) async fn serve_kodo_compat( delivering, publish_admission, subscriptions, + None, + EphemeralConsumers::default(), + shutdown.clone(), )?; let tokens = state.tokens.clone(); let token_shutdown = shutdown.clone(); @@ -69,6 +72,10 @@ mod tests { delivering: Arc::new(AtomicBool::new(true)), publish_admission: Arc::new(PublishAdmission::new(1024, metrics)), subscriptions: SubscriptionRegistry::default(), + authenticator: None, + ephemeral_consumers: EphemeralConsumers::default(), + websocket: WebSocketRuntime::new(&Config::default().websocket), + shutdown: tokio::sync::watch::channel(false).1, started_at: 1, }; diff --git a/crates/server/src/http/websocket.rs b/crates/server/src/http/websocket.rs new file mode 100644 index 0000000..dea7794 --- /dev/null +++ b/crates/server/src/http/websocket.rs @@ -0,0 +1,1061 @@ +use super::*; +use crate::auth::{AuthError, AuthSession, Authenticator}; +use crate::subscriptions::ClientIdentity; +use crate::tcp::EphemeralConsumers; +use axum::extract::connect_info::ConnectInfo; +use axum::extract::ws::{CloseFrame, Message, Utf8Bytes, WebSocket, WebSocketUpgrade}; +use axum::extract::Path; +use bytes::{BufMut, BytesMut}; +use parking_lot::Mutex; +use rustqueue_queue::{BrokerError, Delivery, DeliveryGuard, DeliveryMode, TopicPolicy}; +use serde::Deserialize; +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; +use tower_http::cors::{AllowOrigin, CorsLayer}; +use tracing::debug; + +const SUBPROTOCOL: &str = "rustqueue.live.v1"; +const AUTH_CHANNEL: &str = "websocket#ephemeral"; +const FRAME_MAGIC: &[u8; 4] = b"RQW1"; +const FRAME_HEADER_BYTES: usize = 24; +const MAX_CONTROL_BYTES: usize = 64 * 1024; +const CONTROL_WRITE_TIMEOUT: Duration = Duration::from_secs(1); + +#[derive(Clone)] +pub(super) struct WebSocketRuntime { + connections: Arc, + frame_bytes: Arc, + topics: Arc>>, + max_connections_per_topic: usize, + next_id: Arc, +} + +pub(super) struct ConnectionAdmission { + runtime: WebSocketRuntime, + topic: String, + _connection: OwnedSemaphorePermit, +} + +#[derive(Debug)] +enum AdmissionError { + Global, + Topic, +} + +#[derive(Deserialize)] +struct AuthControl { + #[serde(rename = "type")] + kind: String, + secret: String, +} + +struct SessionAuth { + session: Option, + secret: Option, +} + +struct ControlAuthError { + code: &'static str, + detail: String, +} + +impl WebSocketRuntime { + pub(super) fn new(config: &crate::config::WebSocketConfig) -> Self { + Self { + connections: Arc::new(Semaphore::new(config.max_connections)), + frame_bytes: Arc::new(Semaphore::new(config.frame_inflight_bytes)), + topics: Arc::new(Mutex::new(HashMap::new())), + max_connections_per_topic: config.max_connections_per_topic, + next_id: Arc::new(AtomicU64::new(0)), + } + } + + fn admit(&self, topic: &str) -> Result<(u64, ConnectionAdmission), AdmissionError> { + let connection = Arc::clone(&self.connections) + .try_acquire_owned() + .map_err(|_| AdmissionError::Global)?; + let mut topics = self.topics.lock(); + let count = topics.entry(topic.to_owned()).or_default(); + if *count >= self.max_connections_per_topic { + return Err(AdmissionError::Topic); + } + *count += 1; + let id = self.next_id.fetch_add(1, Ordering::Relaxed).wrapping_add(1); + Ok(( + id, + ConnectionAdmission { + runtime: self.clone(), + topic: topic.to_owned(), + _connection: connection, + }, + )) + } + + fn reserve_frame(&self, bytes: usize) -> Option { + let permits = u32::try_from(bytes.max(1)).ok()?; + Arc::clone(&self.frame_bytes) + .try_acquire_many_owned(permits) + .ok() + } +} + +impl Drop for ConnectionAdmission { + fn drop(&mut self) { + let mut topics = self.runtime.topics.lock(); + let remove = topics.get_mut(&self.topic).is_some_and(|count| { + *count = count.saturating_sub(1); + *count == 0 + }); + if remove { + topics.remove(&self.topic); + } + } +} + +pub(super) fn lookup_cors(allowed_origins: &[String]) -> Option { + let origins = allowed_origins + .iter() + .filter_map(|origin| HeaderValue::from_str(origin).ok()) + .collect::>(); + (!origins.is_empty()).then(|| { + CorsLayer::new() + .allow_origin(AllowOrigin::list(origins)) + .allow_methods([axum::http::Method::GET]) + .allow_headers([header::ACCEPT]) + .max_age(Duration::from_secs(600)) + }) +} + +pub(super) async fn subscribe_topic( + State(state): State, + Path(topic): Path, + ConnectInfo(peer): ConnectInfo, + uri: axum::http::Uri, + headers: HeaderMap, + ws: WebSocketUpgrade, +) -> Result { + if uri.query().is_some() { + return Err(ApiError::bad_request( + "E_WS_QUERY_FORBIDDEN", + "WebSocket subscriptions do not accept query parameters", + )); + } + ensure_origin_allowed(&state.config.websocket.allowed_origins, &headers)?; + if !ws + .requested_protocols() + .any(|protocol| protocol == SUBPROTOCOL) + { + return Err(ApiError::bad_request( + "E_WS_SUBPROTOCOL", + format!("Sec-WebSocket-Protocol must include {SUBPROTOCOL}"), + )); + } + let reliable_policy_epoch = state.broker.topic_reliable_policy_epoch(&topic)?; + let policy = state.broker.topic_policy(&topic)?; + if policy.delivery_mode != DeliveryMode::TtlDiscard { + state + .metrics + .websocket_non_ttl_rejections + .fetch_add(1, Ordering::Relaxed); + return Err(ApiError::conflict( + "E_WS_TTL_REQUIRED", + "WebSocket subscriptions require a TTL_DISCARD Topic", + )); + } + if !state.delivering.load(Ordering::Acquire) { + return Err(ApiError::unavailable( + "E_NOT_READY", + "Broker is not accepting new subscriptions", + )); + } + let (id, admission) = state.websocket.admit(&topic).map_err(|error| { + state + .metrics + .websocket_capacity_rejections + .fetch_add(1, Ordering::Relaxed); + match error { + AdmissionError::Global => ApiError { + status: StatusCode::TOO_MANY_REQUESTS, + code: "E_WS_CAPACITY", + detail: "WebSocket connection capacity is exhausted".into(), + }, + AdmissionError::Topic => ApiError { + status: StatusCode::TOO_MANY_REQUESTS, + code: "E_WS_CAPACITY", + detail: "WebSocket Topic connection capacity is exhausted".into(), + }, + } + })?; + let session_auth = authenticate_header(&state, peer, &topic, &headers).await?; + let max_write_buffer = state + .config + .queue + .max_message_bytes + .saturating_add(FRAME_HEADER_BYTES + 1024); + Ok(ws + .protocols([SUBPROTOCOL]) + .write_buffer_size(0) + .max_write_buffer_size(max_write_buffer) + .max_message_size(MAX_CONTROL_BYTES) + .max_frame_size(MAX_CONTROL_BYTES) + .on_upgrade(move |socket| { + run_session( + socket, + state, + peer, + topic, + id, + admission, + session_auth, + reliable_policy_epoch, + ) + }) + .into_response()) +} + +fn ensure_origin_allowed(allowed: &[String], headers: &HeaderMap) -> Result<(), ApiError> { + let Some(origin) = headers.get(header::ORIGIN) else { + return Ok(()); + }; + let allowed = origin + .to_str() + .ok() + .is_some_and(|origin| allowed.iter().any(|candidate| candidate == origin)); + if allowed { + Ok(()) + } else { + Err(ApiError { + status: StatusCode::FORBIDDEN, + code: "E_WS_ORIGIN", + detail: "WebSocket Origin is not allowed".into(), + }) + } +} + +async fn authenticate_header( + state: &AppState, + peer: SocketAddr, + topic: &str, + headers: &HeaderMap, +) -> Result { + let Some(authenticator) = state.authenticator.as_deref() else { + return Ok(SessionAuth { + session: None, + secret: None, + }); + }; + let Some(value) = headers.get(header::AUTHORIZATION) else { + return Ok(SessionAuth { + session: None, + secret: None, + }); + }; + let secret = value + .to_str() + .ok() + .and_then(|value| value.strip_prefix("Bearer ")) + .filter(|secret| !secret.is_empty()) + .ok_or_else(|| { + observe_auth_failure(state); + ApiError { + status: StatusCode::UNAUTHORIZED, + code: "E_AUTH_FAILED", + detail: "Authorization must use a non-empty Bearer token".into(), + } + })? + .to_owned(); + let session = authenticate(authenticator, peer, topic, &secret) + .await + .map_err(|error| auth_api_error(state, error))?; + Ok(SessionAuth { + session: Some(session), + secret: Some(secret), + }) +} + +#[allow(clippy::too_many_arguments)] +async fn run_session( + mut socket: WebSocket, + state: AppState, + peer: SocketAddr, + topic: String, + id: u64, + _admission: ConnectionAdmission, + mut auth: SessionAuth, + reliable_policy_epoch: u64, +) { + state + .metrics + .websocket_connections + .fetch_add(1, Ordering::Relaxed); + let result = run_session_inner( + &mut socket, + &state, + peer, + &topic, + id, + &mut auth, + reliable_policy_epoch, + ) + .await; + state + .metrics + .websocket_connections + .fetch_sub(1, Ordering::Relaxed); + if let Err(error) = result { + debug!(%peer, %topic, connection_id = id, %error, "WebSocket subscriber disconnected"); + } +} + +async fn run_session_inner( + socket: &mut WebSocket, + state: &AppState, + peer: SocketAddr, + topic: &str, + id: u64, + auth: &mut SessionAuth, + reliable_policy_epoch: u64, +) -> anyhow::Result<()> { + if state.authenticator.is_some() && auth.session.is_none() { + send_control(socket, json!({"type": "auth_required"})).await?; + match read_auth(socket, state, peer, topic).await { + Ok(session) => *auth = session, + Err(error) => { + observe_auth_failure(state); + close_control(socket, error.code, &error.detail, 1008).await; + anyhow::bail!(error.detail); + } + } + } + ensure_authorized(auth.session.as_ref(), topic)?; + let mut policy_changes = match state.broker.subscribe_topic_policy(topic) { + Ok(changes) => changes, + Err(error) => { + handle_broker_error(socket, error).await; + anyhow::bail!("Topic became unavailable before subscription readiness"); + } + }; + if policy_changes.borrow().delivery_mode != DeliveryMode::TtlDiscard { + close_control( + socket, + "E_WS_TTL_REQUIRED", + "Topic no longer uses TTL_DISCARD", + 1008, + ) + .await; + anyhow::bail!("Topic stopped using TTL_DISCARD before subscription readiness"); + } + let channel: Arc = Arc::from(format!("ws.{}.{}#ephemeral", state.config.node.id, id)); + let cleanup = match EphemeralCleanup::register( + state.ephemeral_consumers.clone(), + Arc::clone(&state.broker), + topic, + channel.as_ref(), + ) + .await + { + Ok(cleanup) => cleanup, + Err(error) => { + let (code, close) = match &error { + BrokerError::ChannelLimit => { + state + .metrics + .websocket_capacity_rejections + .fetch_add(1, Ordering::Relaxed); + ("E_WS_CAPACITY", 1013) + } + BrokerError::TopicNotFound + | BrokerError::TopicRetiring + | BrokerError::TopicTombstoned => ("E_TOPIC_CLOSED", 1001), + BrokerError::ManagementUnavailable => ("E_OWNER_CHANGED", 1012), + _ => ("E_BROKER", 1011), + }; + close_control(socket, code, &error.to_string(), close).await; + return Err(error.into()); + } + }; + if policy_changes.has_changed().is_err() { + close_control( + socket, + "E_TOPIC_CLOSED", + "Topic was removed while creating the live Channel", + 1001, + ) + .await; + anyhow::bail!("Topic was removed while creating the live Channel"); + } + let current_policy = match state.broker.topic_policy(topic) { + Ok(policy) => policy, + Err(error) => { + handle_broker_error(socket, error).await; + anyhow::bail!("Topic became unavailable while creating the live Channel"); + } + }; + if current_policy.delivery_mode != DeliveryMode::TtlDiscard { + close_control( + socket, + "E_WS_TTL_REQUIRED", + "Topic no longer uses TTL_DISCARD", + 1008, + ) + .await; + anyhow::bail!("Topic stopped using TTL_DISCARD while creating the live Channel"); + } + if reliable_policy_changed(state, topic, reliable_policy_epoch)? { + close_control( + socket, + "E_WS_TTL_REQUIRED", + "Topic switched to RELIABLE before subscription readiness", + 1008, + ) + .await; + anyhow::bail!("Topic switched to RELIABLE before subscription readiness"); + } + if state.broker.topic_policy(topic)?.delivery_mode != DeliveryMode::TtlDiscard { + close_control( + socket, + "E_WS_TTL_REQUIRED", + "Topic no longer uses TTL_DISCARD", + 1008, + ) + .await; + anyhow::bail!("Topic stopped using TTL_DISCARD before subscription readiness"); + } + let identity = ClientIdentity { + client_id: format!("ws-{id}"), + hostname: "websocket".into(), + remote_address: peer.to_string(), + user_agent: "rustqueue.live.v1".into(), + authed: auth.session.is_some(), + ..ClientIdentity::default() + }; + let lease = state + .subscriptions + .register(topic, channel.as_ref(), identity) + .map_err(|_| anyhow::anyhow!("subscription deletion is in progress"))?; + lease.update_flow(1, 0); + send_control( + socket, + json!({ + "type": "ready", + "topic": topic, + "subscription_id": id.to_string(), + "start": "tail", + "ack": "auto" + }), + ) + .await?; + + let mut shutdown = state.shutdown.clone(); + let heartbeat = Duration::from_millis(state.config.limits.heartbeat_interval_ms); + let mut heartbeat_tick = tokio::time::interval(heartbeat); + heartbeat_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + heartbeat_tick.tick().await; + let mut last_peer_activity = Instant::now(); + loop { + if !state.delivering.load(Ordering::Acquire) { + close_control(socket, "E_DRAINING", "Broker is draining", 1001).await; + break; + } + if reliable_policy_changed(state, topic, reliable_policy_epoch)? { + close_control( + socket, + "E_WS_TTL_REQUIRED", + "Topic switched to RELIABLE", + 1008, + ) + .await; + break; + } + if let Err(error) = refresh_auth(auth, state, peer, topic).await { + close_control(socket, error.code, &error.detail, 1008).await; + anyhow::bail!(error.detail); + } + let fetch = state.broker.fetch_batch_retained( + topic, + channel.as_ref(), + 1, + state.config.queue.max_message_bytes, + Duration::from_secs(1), + Some( + state + .config + .message_timeout() + .saturating_add(Duration::from_millis( + state.config.limits.tcp_command_timeout_ms, + )), + ), + ); + tokio::pin!(fetch); + let batch = tokio::select! { + biased; + changed = shutdown.changed() => { + let _ = changed; + close_control(socket, "E_SHUTDOWN", "Broker is shutting down", 1001).await; + break; + } + changed = policy_changes.changed() => { + if changed.is_err() { + close_control(socket, "E_TOPIC_CLOSED", "Topic was removed", 1001).await; + break; + } + if policy_changes.borrow_and_update().delivery_mode != DeliveryMode::TtlDiscard { + close_control(socket, "E_WS_TTL_REQUIRED", "Topic no longer uses TTL_DISCARD", 1008).await; + break; + } + continue; + } + incoming = socket.recv() => { + if !handle_incoming(socket, incoming, &mut last_peer_activity).await? { + break; + } + continue; + } + _ = heartbeat_tick.tick() => { + if last_peer_activity.elapsed() >= heartbeat.saturating_mul(2) { + state.metrics.websocket_slow_disconnects.fetch_add(1, Ordering::Relaxed); + close_control(socket, "E_WS_HEARTBEAT", "WebSocket heartbeat timed out", 1008).await; + break; + } + let ping = tokio::time::timeout( + Duration::from_millis(state.config.limits.tcp_command_timeout_ms), + socket.send(Message::Ping(bytes::Bytes::new())), + ) + .await; + match ping { + Ok(result) => result?, + Err(_) => { + state.metrics.websocket_slow_disconnects.fetch_add(1, Ordering::Relaxed); + anyhow::bail!("WebSocket heartbeat write timed out"); + } + } + continue; + } + result = &mut fetch => match result { + Ok(batch) => batch, + Err(error) => { + handle_broker_error(socket, error).await; + break; + } + }, + }; + let (mut deliveries, mut guard) = batch.into_parts(); + let Some(delivery) = deliveries.pop() else { + continue; + }; + lease.update_flow(0, 1); + let outcome = deliver_one( + socket, + state, + topic, + &channel, + delivery, + &mut guard, + &mut policy_changes, + reliable_policy_epoch, + ) + .await; + lease.update_flow(1, 0); + match outcome { + Ok(true) => { + lease.observe_delivery(); + lease.observe_finish(); + } + Ok(false) => continue, + Err(error) => { + return Err(error); + } + } + } + drop(lease); + drop(cleanup); + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +async fn deliver_one( + socket: &mut WebSocket, + state: &AppState, + topic: &str, + channel: &Arc, + delivery: Delivery, + guard: &mut DeliveryGuard, + policy_changes: &mut tokio::sync::watch::Receiver, + reliable_policy_epoch: u64, +) -> anyhow::Result { + let Some(expiration_ns) = state + .broker + .delivery_expiration_ns(topic, delivery.timestamp_ns)? + else { + anyhow::bail!("Topic stopped using TTL_DISCARD"); + }; + if remaining_until(expiration_ns).is_zero() { + return Ok(false); + } + let frame_bytes = FRAME_HEADER_BYTES.saturating_add(delivery.body.len()); + let Some(_frame_hold) = state.websocket.reserve_frame(frame_bytes) else { + state + .metrics + .websocket_capacity_rejections + .fetch_add(1, Ordering::Relaxed); + close_control( + socket, + "E_WS_CAPACITY", + "WebSocket frame memory capacity is exhausted", + 1013, + ) + .await; + anyhow::bail!("WebSocket frame memory capacity is exhausted"); + }; + let frame = encode_delivery(&delivery); + let write_deadline = Instant::now() + .checked_add(Duration::from_millis( + state.config.limits.tcp_command_timeout_ms, + )) + .unwrap_or_else(Instant::now); + let send = socket.send(Message::Binary(frame)); + tokio::pin!(send); + let mut shutdown = state.shutdown.clone(); + loop { + let expiration_ns = state + .broker + .delivery_expiration_ns(topic, delivery.timestamp_ns)? + .ok_or_else(|| anyhow::anyhow!("Topic stopped using TTL_DISCARD"))?; + let ttl_remaining = remaining_until(expiration_ns); + let socket_remaining = write_deadline.saturating_duration_since(Instant::now()); + if ttl_remaining.is_zero() { + anyhow::bail!("Message TTL expired during WebSocket frame write"); + } + if socket_remaining.is_zero() { + state + .metrics + .websocket_slow_disconnects + .fetch_add(1, Ordering::Relaxed); + anyhow::bail!("WebSocket frame write timed out"); + } + let timeout = ttl_remaining.min(socket_remaining); + tokio::select! { + biased; + changed = shutdown.changed() => { + let _ = changed; + anyhow::bail!("Broker shut down during WebSocket frame write"); + } + changed = policy_changes.changed() => { + if changed.is_err() { + anyhow::bail!("Topic closed during WebSocket frame write"); + } + if reliable_policy_changed(state, topic, reliable_policy_epoch)? { + anyhow::bail!("Topic switched to RELIABLE during WebSocket frame write"); + } + if policy_changes.borrow_and_update().delivery_mode != DeliveryMode::TtlDiscard { + anyhow::bail!("Topic stopped using TTL_DISCARD during WebSocket frame write"); + } + continue; + } + result = &mut send => { + result?; + break; + } + _ = tokio::time::sleep(timeout) => { + if socket_remaining <= ttl_remaining { + state.metrics.websocket_slow_disconnects.fetch_add(1, Ordering::Relaxed); + anyhow::bail!("WebSocket frame write timed out"); + } + anyhow::bail!("Message TTL expired during WebSocket frame write"); + } + } + } + let Some(token) = guard.accept_with_token(delivery.id) else { + anyhow::bail!("WebSocket delivery reservation disappeared"); + }; + state + .broker + .finish_delivery_shared(topic, Arc::clone(channel), delivery.id, token) + .await?; + state + .metrics + .websocket_messages + .fetch_add(1, Ordering::Relaxed); + state + .metrics + .websocket_bytes + .fetch_add(delivery.body.len() as u64, Ordering::Relaxed); + Ok(true) +} + +fn reliable_policy_changed( + state: &AppState, + topic: &str, + expected_epoch: u64, +) -> Result { + Ok(state.broker.topic_reliable_policy_epoch(topic)? != expected_epoch) +} + +fn encode_delivery(delivery: &Delivery) -> bytes::Bytes { + let mut frame = BytesMut::with_capacity(FRAME_HEADER_BYTES + delivery.body.len()); + frame.extend_from_slice(FRAME_MAGIC); + frame.put_u16(FRAME_HEADER_BYTES as u16); + frame.put_u16(delivery.attempts); + frame.put_i64(delivery.timestamp_ns); + frame.put_u64(delivery.id); + frame.extend_from_slice(&delivery.body); + frame.freeze() +} + +async fn read_auth( + socket: &mut WebSocket, + state: &AppState, + peer: SocketAddr, + topic: &str, +) -> Result { + let message = tokio::time::timeout( + Duration::from_millis(state.config.limits.auth_timeout_ms), + socket.recv(), + ) + .await + .map_err(|_| ControlAuthError { + code: "E_AUTH_TIMEOUT", + detail: "WebSocket AUTH timed out".into(), + })? + .ok_or_else(|| ControlAuthError { + code: "E_AUTH_FAILED", + detail: "WebSocket closed before AUTH".into(), + })? + .map_err(|error| ControlAuthError { + code: "E_AUTH_FAILED", + detail: error.to_string(), + })?; + let Message::Text(text) = message else { + return Err(ControlAuthError { + code: "E_BAD_AUTH", + detail: "WebSocket AUTH must be a text control frame".into(), + }); + }; + let request: AuthControl = + serde_json::from_str(text.as_str()).map_err(|error| ControlAuthError { + code: "E_BAD_AUTH", + detail: error.to_string(), + })?; + if request.kind != "auth" || request.secret.is_empty() { + return Err(ControlAuthError { + code: "E_BAD_AUTH", + detail: "WebSocket AUTH frame is invalid".into(), + }); + } + let authenticator = state + .authenticator + .as_deref() + .ok_or_else(|| ControlAuthError { + code: "E_AUTH_DISABLED", + detail: "WebSocket AUTH is disabled".into(), + })?; + let session = authenticate(authenticator, peer, topic, &request.secret) + .await + .map_err(|error| ControlAuthError { + code: auth_error_code(&error), + detail: "WebSocket AUTH failed".into(), + })?; + Ok(SessionAuth { + session: Some(session), + secret: Some(request.secret), + }) +} + +async fn refresh_auth( + auth: &mut SessionAuth, + state: &AppState, + peer: SocketAddr, + topic: &str, +) -> Result<(), ControlAuthError> { + let Some(session) = auth.session.as_ref() else { + return Ok(()); + }; + if !session.is_expired() { + ensure_authorized(Some(session), topic).map_err(|_| ControlAuthError { + code: "E_UNAUTHORIZED", + detail: "WebSocket subscription is no longer authorized".into(), + })?; + return Ok(()); + } + let authenticator = state + .authenticator + .as_deref() + .ok_or_else(|| ControlAuthError { + code: "E_AUTH_FAILED", + detail: "WebSocket AUTH configuration is unavailable".into(), + })?; + let secret = auth.secret.as_deref().ok_or_else(|| ControlAuthError { + code: "E_AUTH_FAILED", + detail: "WebSocket AUTH state is unavailable".into(), + })?; + let session = authenticate(authenticator, peer, topic, secret) + .await + .map_err(|error| { + observe_auth_failure(state); + ControlAuthError { + code: auth_error_code(&error), + detail: "WebSocket AUTH refresh failed".into(), + } + })?; + auth.session = Some(session); + Ok(()) +} + +async fn authenticate( + authenticator: &Authenticator, + peer: SocketAddr, + topic: &str, + secret: &str, +) -> Result { + let session = authenticator + .authenticate(&peer.ip().to_string(), false, "", secret.as_bytes()) + .await?; + if !session.can_subscribe(topic, AUTH_CHANNEL) { + return Err(AuthError::Unauthorized); + } + Ok(session) +} + +fn ensure_authorized(session: Option<&AuthSession>, topic: &str) -> anyhow::Result<()> { + if session.is_some_and(|session| !session.can_subscribe(topic, AUTH_CHANNEL)) { + anyhow::bail!("WebSocket subscription is not authorized"); + } + Ok(()) +} + +fn auth_api_error(state: &AppState, error: AuthError) -> ApiError { + observe_auth_failure(state); + let (status, code, detail) = match error { + AuthError::Unauthorized => ( + StatusCode::FORBIDDEN, + "E_UNAUTHORIZED", + "AUTH does not permit this Topic subscription", + ), + AuthError::Overloaded => ( + StatusCode::SERVICE_UNAVAILABLE, + "E_AUTH_OVERLOADED", + "AUTH capacity is exhausted", + ), + _ => ( + StatusCode::SERVICE_UNAVAILABLE, + "E_AUTH_FAILED", + "AUTH service failed", + ), + }; + ApiError { + status, + code, + detail: detail.into(), + } +} + +fn observe_auth_failure(state: &AppState) { + state + .metrics + .websocket_auth_failures + .fetch_add(1, Ordering::Relaxed); + state.metrics.auth_failures.fetch_add(1, Ordering::Relaxed); +} + +fn auth_error_code(error: &AuthError) -> &'static str { + match error { + AuthError::Unauthorized => "E_UNAUTHORIZED", + AuthError::Overloaded => "E_AUTH_OVERLOADED", + _ => "E_AUTH_FAILED", + } +} + +async fn handle_incoming( + socket: &mut WebSocket, + incoming: Option>, + last_peer_activity: &mut Instant, +) -> anyhow::Result { + let Some(message) = incoming else { + return Ok(false); + }; + let message = message?; + *last_peer_activity = Instant::now(); + match message { + Message::Close(_) => Ok(false), + Message::Ping(_) => Ok(true), + Message::Pong(_) => Ok(true), + Message::Text(_) | Message::Binary(_) => { + close_control(socket, "E_WS_CONTROL", "Unexpected client data frame", 1008).await; + Ok(false) + } + } +} + +async fn handle_broker_error(socket: &mut WebSocket, error: BrokerError) { + let (code, close) = match &error { + BrokerError::TopicNotFound | BrokerError::TopicRetiring | BrokerError::TopicTombstoned => { + ("E_TOPIC_CLOSED", 1001) + } + BrokerError::ManagementUnavailable => ("E_OWNER_CHANGED", 1012), + _ => ("E_BROKER", 1011), + }; + close_control(socket, code, &error.to_string(), close).await; +} + +async fn send_control(socket: &mut WebSocket, value: Value) -> anyhow::Result<()> { + tokio::time::timeout( + CONTROL_WRITE_TIMEOUT, + socket.send(Message::Text(serde_json::to_string(&value)?.into())), + ) + .await + .map_err(|_| anyhow::anyhow!("WebSocket control write timed out"))??; + Ok(()) +} + +async fn close_control(socket: &mut WebSocket, code: &'static str, detail: &str, close_code: u16) { + let _ = tokio::time::timeout(CONTROL_WRITE_TIMEOUT, async { + send_control( + socket, + json!({"type": "error", "code": code, "detail": detail}), + ) + .await?; + socket + .send(Message::Close(Some(CloseFrame { + code: close_code, + reason: Utf8Bytes::from_static(code), + }))) + .await?; + anyhow::Ok(()) + }) + .await; +} + +fn remaining_until(deadline_ns: i64) -> Duration { + let remaining = deadline_ns.saturating_sub(now_ns()); + Duration::from_nanos(remaining.max(0) as u64) +} + +fn now_ns() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + .min(i64::MAX as u128) as i64 +} + +struct EphemeralCleanup { + consumers: EphemeralConsumers, + broker: Arc, + topic: String, + channel: String, +} + +impl EphemeralCleanup { + async fn register( + consumers: EphemeralConsumers, + broker: Arc, + topic: &str, + channel: &str, + ) -> Result { + consumers.register_existing(&broker, topic, channel).await?; + Ok(Self { + consumers, + broker, + topic: topic.into(), + channel: channel.into(), + }) + } +} + +impl Drop for EphemeralCleanup { + fn drop(&mut self) { + let consumers = self.consumers.clone(); + let broker = Arc::clone(&self.broker); + let topic = self.topic.clone(); + let channel = self.channel.clone(); + tokio::spawn(async move { + consumers.unregister(&broker, &topic, &channel).await; + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn runtime(max_connections: usize, max_connections_per_topic: usize) -> WebSocketRuntime { + let config = crate::config::WebSocketConfig { + max_connections, + max_connections_per_topic, + ..Default::default() + }; + WebSocketRuntime::new(&config) + } + + #[test] + fn encodes_rqw1_golden_frame() { + let delivery = Delivery { + id: 0x0102_0304_0506_0708, + timestamp_ns: -0x0102_0304_0506_0708, + attempts: 0x090a, + body: Arc::<[u8]>::from(&b"hello"[..]), + }; + + let frame = encode_delivery(&delivery); + + assert_eq!( + frame.as_ref(), + &[ + b'R', b'Q', b'W', b'1', 0, 24, 9, 10, 0xfe, 0xfd, 0xfc, 0xfb, 0xfa, 0xf9, 0xf8, + 0xf8, 1, 2, 3, 4, 5, 6, 7, 8, b'h', b'e', b'l', b'l', b'o', + ] + ); + } + + #[test] + fn admission_enforces_global_and_per_topic_limits_and_releases() { + let runtime = runtime(2, 1); + let (_, first) = runtime.admit("events").unwrap(); + assert!(matches!( + runtime.admit("events"), + Err(AdmissionError::Topic) + )); + let (_, second) = runtime.admit("other").unwrap(); + assert!(matches!( + runtime.admit("third"), + Err(AdmissionError::Global) + )); + + drop(first); + let (_, replacement) = runtime.admit("events").unwrap(); + assert!(matches!( + runtime.admit("other"), + Err(AdmissionError::Global) + )); + + drop(second); + drop(replacement); + assert!(runtime.admit("events").is_ok()); + } + + #[test] + fn origin_check_allows_missing_and_exact_origin_only() { + let allowed = ["https://app.example".to_owned()]; + let missing = HeaderMap::new(); + assert!(ensure_origin_allowed(&allowed, &missing).is_ok()); + + let mut exact = HeaderMap::new(); + exact.insert( + header::ORIGIN, + HeaderValue::from_static("https://app.example"), + ); + assert!(ensure_origin_allowed(&allowed, &exact).is_ok()); + + let mut near_miss = HeaderMap::new(); + near_miss.insert( + header::ORIGIN, + HeaderValue::from_static("https://app.example.evil"), + ); + let error = ensure_origin_allowed(&allowed, &near_miss).unwrap_err(); + assert_eq!(error.status, StatusCode::FORBIDDEN); + assert_eq!(error.code, "E_WS_ORIGIN"); + } +} diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index a9a891b..94c5ba7 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -11,6 +11,7 @@ mod tls; use admission::PublishAdmission; use anyhow::Context; +use auth::Authenticator; use clap::Parser; use config::Config; use metrics::Metrics; @@ -106,6 +107,10 @@ async fn main() -> anyhow::Result<()> { Arc::clone(&metrics), )); let subscriptions = SubscriptionRegistry::default(); + let authenticator = Authenticator::new(&config) + .map_err(anyhow::Error::msg)? + .map(Arc::new); + let ephemeral_consumers = tcp::EphemeralConsumers::default(); let initially_pressured = disk_guard::initialize(&config, &publish_admission, &metrics)?; let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); @@ -125,6 +130,8 @@ async fn main() -> anyhow::Result<()> { Arc::clone(&delivering), Arc::clone(&publish_admission), subscriptions.clone(), + authenticator.clone(), + ephemeral_consumers.clone(), shutdown_rx.clone(), std::time::Duration::from_secs(config.shutdown.grace_seconds) .saturating_sub(std::time::Duration::from_millis(250)), @@ -137,6 +144,8 @@ async fn main() -> anyhow::Result<()> { Arc::clone(&delivering), Arc::clone(&publish_admission), subscriptions.clone(), + authenticator, + ephemeral_consumers, shutdown_rx.clone(), )); let mut kodo_http_task = tokio::spawn(http::serve_kodo_compat( diff --git a/crates/server/src/metrics.rs b/crates/server/src/metrics.rs index d11e48b..a53c360 100644 --- a/crates/server/src/metrics.rs +++ b/crates/server/src/metrics.rs @@ -7,6 +7,13 @@ use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; #[derive(Default)] pub struct Metrics { pub tcp_connections: AtomicI64, + pub websocket_connections: AtomicI64, + pub websocket_messages: AtomicU64, + pub websocket_bytes: AtomicU64, + pub websocket_auth_failures: AtomicU64, + pub websocket_capacity_rejections: AtomicU64, + pub websocket_slow_disconnects: AtomicU64, + pub websocket_non_ttl_rejections: AtomicU64, pub publish_messages: AtomicU64, pub publish_bytes: AtomicU64, pub publish_inflight_bytes: AtomicI64, @@ -36,6 +43,13 @@ pub struct Metrics { #[derive(Clone, Debug, Serialize)] pub struct RuntimeMetricsSnapshot { pub tcp_connections: i64, + pub websocket_connections: i64, + pub websocket_messages: u64, + pub websocket_bytes: u64, + pub websocket_auth_failures: u64, + pub websocket_capacity_rejections: u64, + pub websocket_slow_disconnects: u64, + pub websocket_non_ttl_rejections: u64, pub publish_messages: u64, pub publish_bytes: u64, pub publish_inflight_bytes: i64, @@ -300,6 +314,15 @@ impl Metrics { pub fn snapshot(&self) -> RuntimeMetricsSnapshot { RuntimeMetricsSnapshot { tcp_connections: self.tcp_connections.load(Ordering::Relaxed), + websocket_connections: self.websocket_connections.load(Ordering::Relaxed), + websocket_messages: self.websocket_messages.load(Ordering::Relaxed), + websocket_bytes: self.websocket_bytes.load(Ordering::Relaxed), + websocket_auth_failures: self.websocket_auth_failures.load(Ordering::Relaxed), + websocket_capacity_rejections: self + .websocket_capacity_rejections + .load(Ordering::Relaxed), + websocket_slow_disconnects: self.websocket_slow_disconnects.load(Ordering::Relaxed), + websocket_non_ttl_rejections: self.websocket_non_ttl_rejections.load(Ordering::Relaxed), publish_messages: self.publish_messages.load(Ordering::Relaxed), publish_bytes: self.publish_bytes.load(Ordering::Relaxed), publish_inflight_bytes: self.publish_inflight_bytes.load(Ordering::Relaxed), @@ -332,6 +355,20 @@ impl Metrics { concat!( "# TYPE rustqueue_tcp_connections gauge\n", "rustqueue_tcp_connections {}\n", + "# TYPE rustqueue_websocket_connections gauge\n", + "rustqueue_websocket_connections {}\n", + "# TYPE rustqueue_websocket_messages_total counter\n", + "rustqueue_websocket_messages_total {}\n", + "# TYPE rustqueue_websocket_bytes_total counter\n", + "rustqueue_websocket_bytes_total {}\n", + "# TYPE rustqueue_websocket_auth_failures_total counter\n", + "rustqueue_websocket_auth_failures_total {}\n", + "# TYPE rustqueue_websocket_capacity_rejections_total counter\n", + "rustqueue_websocket_capacity_rejections_total {}\n", + "# TYPE rustqueue_websocket_slow_disconnects_total counter\n", + "rustqueue_websocket_slow_disconnects_total {}\n", + "# TYPE rustqueue_websocket_non_ttl_rejections_total counter\n", + "rustqueue_websocket_non_ttl_rejections_total {}\n", "# TYPE rustqueue_publish_messages_total counter\n", "rustqueue_publish_messages_total {}\n", "# TYPE rustqueue_publish_bytes_total counter\n", @@ -382,6 +419,13 @@ impl Metrics { "rustqueue_protective_evicted_messages_total {}\n" ), self.tcp_connections.load(Ordering::Relaxed), + self.websocket_connections.load(Ordering::Relaxed), + self.websocket_messages.load(Ordering::Relaxed), + self.websocket_bytes.load(Ordering::Relaxed), + self.websocket_auth_failures.load(Ordering::Relaxed), + self.websocket_capacity_rejections.load(Ordering::Relaxed), + self.websocket_slow_disconnects.load(Ordering::Relaxed), + self.websocket_non_ttl_rejections.load(Ordering::Relaxed), self.publish_messages.load(Ordering::Relaxed), self.publish_bytes.load(Ordering::Relaxed), self.publish_inflight_bytes.load(Ordering::Relaxed), diff --git a/crates/server/src/tcp.rs b/crates/server/src/tcp.rs index 68c898c..694b7ab 100644 --- a/crates/server/src/tcp.rs +++ b/crates/server/src/tcp.rs @@ -13,7 +13,7 @@ use channel_ops::*; use codec::*; use commands::*; use dead_letter::*; -use ephemeral::*; +pub(crate) use ephemeral::EphemeralConsumers; use session::*; use time::*; use writer::*; @@ -224,16 +224,14 @@ pub async fn serve( delivering: Arc, publish_admission: Arc, subscriptions: SubscriptionRegistry, + authenticator: Option>, + ephemeral_consumers: EphemeralConsumers, mut shutdown: watch::Receiver, shutdown_grace: Duration, ) -> anyhow::Result<()> { let listener = TcpListener::bind(config.network.tcp_address).await?; let tls_acceptor = tls::acceptor(config.security.tls.as_ref())?; - let authenticator = Authenticator::new(&config) - .map_err(anyhow::Error::msg)? - .map(Arc::new); let permits = Arc::new(Semaphore::new(config.limits.max_connections)); - let ephemeral_consumers = EphemeralConsumers::default(); info!(address = %config.network.tcp_address, "NSQ TCP listener ready"); let mut sessions = JoinSet::new(); diff --git a/crates/server/src/tcp/ephemeral.rs b/crates/server/src/tcp/ephemeral.rs index 6f4a0d3..8d00a18 100644 --- a/crates/server/src/tcp/ephemeral.rs +++ b/crates/server/src/tcp/ephemeral.rs @@ -1,16 +1,35 @@ use super::*; #[derive(Clone, Default)] -pub(super) struct EphemeralConsumers { +pub(crate) struct EphemeralConsumers { counts: Arc>>, } impl EphemeralConsumers { - pub async fn register( + pub(crate) async fn register( &self, broker: &Broker, topic: &str, channel: &str, + ) -> Result<(), BrokerError> { + self.register_inner(broker, topic, channel, true).await + } + + pub(crate) async fn register_existing( + &self, + broker: &Broker, + topic: &str, + channel: &str, + ) -> Result<(), BrokerError> { + self.register_inner(broker, topic, channel, false).await + } + + async fn register_inner( + &self, + broker: &Broker, + topic: &str, + channel: &str, + create_topic: bool, ) -> Result<(), BrokerError> { let counts = Arc::clone(&self.counts); let broker = broker.clone(); @@ -19,7 +38,11 @@ impl EphemeralConsumers { let (reply, result) = tokio::sync::oneshot::channel(); tokio::spawn(async move { let mut counts = counts.lock().await; - let outcome = broker.create_channel(&topic, &channel).await; + let outcome = if create_topic { + broker.create_channel(&topic, &channel).await + } else { + broker.create_channel_existing(&topic, &channel).await + }; complete_registration(&mut counts, &broker, &topic, &channel, outcome, reply).await; }); let commit = result @@ -28,7 +51,7 @@ impl EphemeralConsumers { commit.send(()).map_err(|_| BrokerError::StorageUnavailable) } - pub async fn unregister(&self, broker: &Broker, topic: &str, channel: &str) { + pub(crate) async fn unregister(&self, broker: &Broker, topic: &str, channel: &str) { let counts = Arc::clone(&self.counts); let broker = broker.clone(); let topic = topic.to_owned(); diff --git a/deploy/helm/rustqueue/Chart.yaml b/deploy/helm/rustqueue/Chart.yaml index 990e63f..38df640 100644 --- a/deploy/helm/rustqueue/Chart.yaml +++ b/deploy/helm/rustqueue/Chart.yaml @@ -2,8 +2,8 @@ apiVersion: v2 name: rustqueue description: Kubernetes-native share-nothing NSQ-compatible message queue type: application -version: 0.9.0 -appVersion: "0.9.0" +version: 0.10.0 +appVersion: "0.10.0" kubeVersion: ">=1.28.0-0" keywords: - messaging diff --git a/deploy/helm/rustqueue/crds/rustqueue.io_rustqueues.yaml b/deploy/helm/rustqueue/crds/rustqueue.io_rustqueues.yaml index 9a4fa76..8ed4374 100644 --- a/deploy/helm/rustqueue/crds/rustqueue.io_rustqueues.yaml +++ b/deploy/helm/rustqueue/crds/rustqueue.io_rustqueues.yaml @@ -68,6 +68,23 @@ spec: default: {} proxyTcpMaxConnectionAgeSeconds: {type: integer, format: int64, minimum: 0, default: 300} discoveryReplicas: {type: integer, format: int32, minimum: 2, default: 2} + websocket: + type: object + default: + enabled: false + maxConnections: 1024 + maxConnectionsPerTopic: 256 + frameInflightBytes: 134217728 + allowedOrigins: [] + properties: + enabled: {type: boolean, default: false} + maxConnections: {type: integer, format: int64, minimum: 1, default: 1024} + maxConnectionsPerTopic: {type: integer, format: int64, minimum: 1, default: 256} + frameInflightBytes: {type: integer, format: int64, minimum: 25, maximum: 4294967295, default: 134217728} + allowedOrigins: + type: array + default: [] + items: {type: string, pattern: '^https?://[^/?#,\s]+$'} kodoCompatibility: type: object default: diff --git a/deploy/helm/rustqueue/templates/NOTES.txt b/deploy/helm/rustqueue/templates/NOTES.txt index cb9be5b..1630bfc 100644 --- a/deploy/helm/rustqueue/templates/NOTES.txt +++ b/deploy/helm/rustqueue/templates/NOTES.txt @@ -10,6 +10,13 @@ Producer endpoints (node-local proxy Service): Consumer lookup endpoint: http://{{ .Values.queue.name }}-discovery.{{ .Release.Namespace }}.svc:4161 + +WebSocket TTL live subscriptions (when enabled) connect directly to each +Broker HTTP address returned by Discovery /lookup: + ws:///v1/ws/topics/ + (use wss:// when the Broker HTTP endpoint is served with TLS) + Subprotocol: rustqueue.live.v1 + Only existing TTL_DISCARD Topics are accepted; there is no replay or ACK. {{- end }} {{- if and .Values.queue.create .Values.console.enabled }} RustQueue Console is available inside the cluster at: diff --git a/deploy/helm/rustqueue/templates/queue.yaml b/deploy/helm/rustqueue/templates/queue.yaml index cf2010f..a8591d2 100644 --- a/deploy/helm/rustqueue/templates/queue.yaml +++ b/deploy/helm/rustqueue/templates/queue.yaml @@ -46,6 +46,13 @@ spec: maxDetailedMetricSeries: {{ .Values.queue.maxDetailedMetricSeries | int64 }} consoleManagementEnabled: {{ and .Values.console.enabled .Values.console.management.enabled }} discoveryReplicas: {{ .Values.queue.discoveryReplicas }} + websocket: + enabled: {{ .Values.queue.websocket.enabled }} + maxConnections: {{ .Values.queue.websocket.maxConnections | int64 }} + maxConnectionsPerTopic: {{ .Values.queue.websocket.maxConnectionsPerTopic | int64 }} + frameInflightBytes: {{ .Values.queue.websocket.frameInflightBytes | int64 }} + allowedOrigins: + {{- toYaml .Values.queue.websocket.allowedOrigins | nindent 6 }} proxyNodeSelector: {{- toYaml .Values.queue.proxyNodeSelector | nindent 4 }} proxyTcpMaxConnectionAgeSeconds: {{ .Values.queue.proxyTcpMaxConnectionAgeSeconds | int64 }} diff --git a/deploy/helm/rustqueue/values.yaml b/deploy/helm/rustqueue/values.yaml index 0dd8186..8729276 100644 --- a/deploy/helm/rustqueue/values.yaml +++ b/deploy/helm/rustqueue/values.yaml @@ -5,7 +5,7 @@ operator: replicas: 2 image: repository: rustqueue-operator - tag: "0.9.0" + tag: "0.10.0" pullPolicy: IfNotPresent log: rustqueue_operator=info resources: @@ -20,7 +20,7 @@ operator: queue: create: true name: rustqueue - image: rustqueue:0.9.0 + image: rustqueue:0.10.0 imagePullPolicy: IfNotPresent minBrokers: 1 maxBrokers: 500 @@ -57,6 +57,12 @@ queue: registrySecretName: "" clientTlsSecretName: "" discoveryReplicas: 2 + websocket: + enabled: false + maxConnections: 1024 + maxConnectionsPerTopic: 256 + frameInflightBytes: 134217728 + allowedOrigins: [] proxyNodeSelector: {} # Rotate opaque TCP tunnels so a small producer pool is redistributed as the # Broker fleet changes. Set to 0 to keep connections pinned indefinitely. diff --git a/docs/architecture/share-nothing-v7.md b/docs/architecture/share-nothing-v7.md index 6c12c5a..c671d37 100644 --- a/docs/architecture/share-nothing-v7.md +++ b/docs/architecture/share-nothing-v7.md @@ -1,7 +1,7 @@ # RustQueue format v7 share-nothing architecture Status: accepted implementation contract -Target release: 0.9.0 +Target release: 0.10.0 Data format: v7, clean directories only ## 1. Goal @@ -125,7 +125,109 @@ TTL cannot restore an already expired prefix. the fleet over time, but it cannot turn 16 streams into 500-way concurrent disk I/O. -### 2.5 Ordering and identity +### 2.5 WebSocket TTL live subscriptions + +The Broker HTTP listener optionally exposes a best-effort live subscription for +short-lived Topics: + +```text +GET /v1/ws/topics/{topic} +Sec-WebSocket-Protocol: rustqueue.live.v1 +``` + +The feature is default-off. A handshake succeeds only when the addressed +Topic already exists locally and its persisted policy is `TTL_DISCARD`; the +endpoint never creates a Topic, and `RELIABLE` or unknown Topics are rejected +with a stable error code. One connection represents one Topic and begins at +the current tail. There is no historical replay, reconnect resume, consumer +ACK or WebSocket retry. + +Clients first query Discovery `/lookup` and then connect directly to every +healthy Broker owner returned for the Topic. The producer proxy is not a +WebSocket fan-out layer. During a visible owner migration, clients may union +the streams and deduplicate using the broker-scoped message ID in the frame. +Discovery's read-only lookup response uses the same allowed-origin policy as +the Broker WebSocket endpoint. + +Each message is written as one complete binary frame. Its first 24 bytes are +big-endian metadata, followed by the original body: + +| Offset | Size | Field | +| --- | ---: | --- | +| `0..4` | 4 | ASCII `RQW1` | +| `4..6` | 2 | Header length, fixed at `24` | +| `6..8` | 2 | Delivery attempts | +| `8..16` | 8 | Broker timestamp, signed `i64` nanoseconds | +| `16..24` | 8 | Broker-scoped message ID, unsigned `u64` | +| `24..` | — | Original message body | + +Text frames are control messages. `auth_required`, client `auth`, `ready` and +stable error codes are part of the v1 control contract. Subscription +authorization reuses the existing AUTH rules with the virtual Channel +`websocket#ephemeral`. Service clients can send an HTTP +`Authorization: Bearer` header during upgrade; browser clients must use the +control exchange and must not put credentials in a query string. A browser +Origin must exactly match the configured allowlist. A missing Origin is +allowed for non-browser clients but does not bypass authentication. + +The HTTP `101 Switching Protocols` response establishes only the transport. +The live subscription boundary is the server's `{"type":"ready",...}` control +frame. No message capture is promised before `ready`, including during the +AUTH exchange after `auth_required`; an HTTP upgrade is not a capture +boundary. + +Handshake failures reuse the HTTP `ApiError` shape: the stable code is in the +JSON `message` field and the human-readable explanation is in `detail`. Once +upgraded, terminal errors use a text control of the form +`{"type":"error","code":"...","detail":"..."}` before close: + +| Code | Meaning | +| --- | --- | +| `E_WS_QUERY_FORBIDDEN` | Query parameters are forbidden; use the path and headers. | +| `E_WS_ORIGIN` | Browser Origin is not an exact configured allowlist entry. | +| `E_WS_SUBPROTOCOL` | `rustqueue.live.v1` was not negotiated. | +| `E_BAD_TOPIC` | Topic does not exist on this Broker. | +| `E_WS_TTL_REQUIRED` | Existing Topic is not `TTL_DISCARD` or changed away from it. | +| `E_WS_CAPACITY` | Connection, Channel or frame-memory capacity is exhausted. | +| `E_WS_CONTROL` | Unsupported client text/binary data frame was received. | +| `E_WS_HEARTBEAT` | Peer missed the heartbeat liveness window. | +| `E_AUTH_TIMEOUT` | `auth` was not received before the AUTH deadline. | +| `E_BAD_AUTH` | AUTH control was non-text, malformed, or missing a valid `auth`/`secret`. | +| `E_AUTH_OVERLOADED` | AUTH service or capacity gate is exhausted; retry later. | +| `E_AUTH_FAILED` | Authentication service or credential validation failed. | +| `E_UNAUTHORIZED` | AUTH does not permit the Topic subscription. | +| `E_TOPIC_CLOSED` | Topic was deleted, tombstoned, or closed. | +| `E_OWNER_CHANGED` | Topic owner/fence changed; perform a fresh lookup. | +| `E_DRAINING` | Broker is draining and closes the live subscription. | +| `E_SHUTDOWN` | Broker is shutting down; reconnect after discovery. | + +The delivery path creates one ephemeral Channel per connection and reserves at +most one message at a time. A reservation is released when the complete frame +cannot be written, when the connection closes, or when the Broker is shutting +down; there is no reliable backlog. The write deadline is the smaller of the +socket timeout and the message's remaining TTL. If a frame cannot finish by +that deadline, the connection closes rather than leaving a partial frame. +Paused Topics stop taking messages. Topic deletion, migration, owner shutdown, +or a change to `RELIABLE` closes affected connections and leaves reconnection +to the client's next lookup. + +Runtime defaults are bounded and opt-in: + +```toml +[websocket] +enabled = false +max_connections = 1024 +max_connections_per_topic = 256 +frame_inflight_bytes = 134217728 +allowed_origins = [] +``` + +The node-wide frame budget and connection caps protect publishers from slow +consumers. Metrics are aggregate by default and include connection count, +completed frame messages/bytes, authentication failures, capacity rejections, +non-TTL Topic rejections and slow-consumer closes. + +### 2.6 Ordering and identity - Storage append order exists only inside one broker/topic log. - There is no cross-broker order and no global delivery completion order. @@ -144,6 +246,8 @@ TTL cannot restore an already expired prefix. - Client TLS/mTLS, AUTH, Snappy, and Deflate. - Local topic segments and channel WAL/checkpoints. - Local health, metrics, stats, drain, scrub, and disk-pressure endpoints. +- Optional best-effort WebSocket delivery for existing `TTL_DISCARD` Topics; + it shares the Broker HTTP listener and does not add a listener or port. - No OpenRaft node, internal replication server, leader routing, peer discovery, or membership controller. @@ -169,6 +273,9 @@ Each replica: `/lookup` returns the healthy owners during a partial Broker outage while the complete-inventory metric remains false. +WebSocket clients use `/lookup` only for owner discovery. Discovery does not +proxy or multiplex the WebSocket data path. + Discovery state is derived and is never authoritative message metadata. A restart reconstructs the complete index from ready brokers. Replicas do not coordinate and clients may union their answers exactly as with multiple @@ -366,6 +473,12 @@ move to discovery. Native v6 cluster APIs are removed, including partition, replica, migration, rebalance, transfer-leader, federation, and cluster operation resources. +The optional native Broker endpoint `GET /v1/ws/topics/{topic}` is available +only when WebSocket support is enabled and the Topic is `TTL_DISCARD`. Clients +must negotiate `rustqueue.live.v1`; a successful frame write is best effort and +does not acknowledge application processing. `/v1/capabilities` advertises the +`websocket_ttl_live_v1` capability when the binary supports this protocol. + `/ping` remains process liveness. Broker readiness requires local storage, clock, and admission health only; it has no cluster-wide dependency. @@ -373,6 +486,8 @@ clock, and admission health only; it has no cluster-wide dependency. - The deployment targets a trusted internal Kubernetes network. - Client-facing broker TLS/mTLS and external AUTH remain optional and compatible. +- WebSocket Origin checks are exact matches against `websocket.allowed_origins`; + the same allowlist is used by Discovery's read-only `/lookup` response. - Certificates are mounted from user-provided Kubernetes Secrets. - There is no operator-managed CA. - Discovery-to-broker polling uses a bounded bearer token, dedicated Service @@ -478,6 +593,11 @@ The implementation is complete only when all of the following pass: 14. Console management acceptance creates a real three-owner topic, forces the Console Pod down after one owner is durably recorded, and verifies the new Pod completes the remaining owners without reapplying the finished owner. +15. WebSocket acceptance verifies the `rustqueue.live.v1` handshake and 24-byte + `RQW1` frame layout, rejects missing and `RELIABLE` Topics without creating + them, confirms no replay across connect/disconnect, and exercises AUTH, + exact Origin checks, TTL deadline closure, Topic pause/delete/migration, + connection and frame-budget rejection, and reservation cleanup. ## 10. Explicit non-goals @@ -487,5 +607,7 @@ The implementation is complete only when all of the following pass: - global deduplication, order, channel catalog, or atomic management; - online broker data migration; - consumer connection fan-out reduction; +- reliable WebSocket delivery, replay, resume, client ACK or a shared + WebSocket consumer group; - cross-region replication; - Broker/PVC lifecycle controls in Console. diff --git a/docs/operations/kubernetes.md b/docs/operations/kubernetes.md index e3480e5..e373562 100644 --- a/docs/operations/kubernetes.md +++ b/docs/operations/kubernetes.md @@ -71,6 +71,83 @@ kubectl apply -f deploy/helm/rustqueue/crds kubectl wait --for=condition=Established crd/rustqueuetopics.rustqueue.io --timeout=60s ``` +## WebSocket TTL live subscriptions + +The optional WebSocket feed is disabled by default and is available only for +Topics whose persisted `deliveryMode` is `TTL_DISCARD`. It is served by each +Broker's existing HTTP listener at: + +```text +GET /v1/ws/topics/{topic} +Sec-WebSocket-Protocol: rustqueue.live.v1 +``` + +The endpoint does not create Topics. A missing Topic or a `RELIABLE` Topic is +rejected during the handshake with a stable error code. One connection covers +one Topic and starts at the current tail; there is no replay, reconnect resume, +WebSocket ACK or WebSocket-layer retry. A complete frame write only means that +RustQueue finished the socket write—it does not mean that the client processed +the message. + +Discover owners first, then connect directly to every Broker HTTP address in +the Discovery `/lookup` response. Do not send WebSocket traffic through the +producer proxy or assume that one Broker is a global Topic owner. During a +visible migration, union the owner streams and deduplicate with the message ID +in each binary frame. Owner shutdown, migration, Topic deletion, or a change +to `RELIABLE` closes affected connections; repeat `/lookup` before reconnecting. + +Configure the same values in the runtime `[websocket]` block, the Helm +`queue.websocket` values, and the RustQueue Operator/CRD `websocket` spec: + +```toml +[websocket] +enabled = false +max_connections = 1024 +max_connections_per_topic = 256 +frame_inflight_bytes = 134217728 +allowed_origins = [] +``` + +For example, a deployment can opt in with equivalent Helm fields: + +```sh +helm upgrade rustqueue deploy/helm/rustqueue \ + --namespace rustqueue --reuse-values \ + --set queue.websocket.enabled=true \ + --set queue.websocket.maxConnections=1024 \ + --set queue.websocket.maxConnectionsPerTopic=256 \ + --set queue.websocket.frameInflightBytes=134217728 +``` + +`allowed_origins` is an exact browser Origin allowlist. Browser upgrades with +an Origin not in the list are rejected; a missing Origin is allowed for a +service client but still requires normal authentication. Discovery's +read-only `/lookup` endpoint uses the same allowlist. Service clients can use +an `Authorization: Bearer` header during upgrade. Browser clients use the +`auth_required` / `auth` text-control exchange, and credentials must not be +placed in a query parameter. Authorization reuses the existing subscription +permission and the virtual Channel `websocket#ephemeral`. + +The binary data frame begins with 24 bytes of big-endian metadata followed by +the raw message body: `RQW1` at `0..4`, header length `24` at `4..6`, attempts +at `6..8`, signed Broker nanosecond timestamp at `8..16`, and unsigned +broker-scoped message ID at `16..24`. Text frames carry `auth_required`, +`ready`, and stable error controls. A client should ignore unknown control +fields so the protocol can evolve. + +Each connection has at most one outstanding reservation. The Broker releases +the reservation and closes the connection if the frame cannot be completely +written before the smaller of the socket timeout and remaining message TTL. +Pause stops new WebSocket deliveries without creating reliable backlog. The +global frame budget, per-topic and total connection caps are fail-closed; watch +connection count, frame messages/bytes, authentication failures, capacity +rejections, non-TTL rejections and slow-consumer closes before raising limits. + +The feature reuses the existing Broker HTTP Service and NetworkPolicy. It does +not add a port or create a default Ingress. If external browser access is +needed, put the existing HTTP endpoint behind the deployment's TLS and +authentication boundary and keep the Origin allowlist narrow. + ## Targeted Broker maintenance ```sh @@ -101,7 +178,7 @@ Canary approval is optional: ```sh helm upgrade rustqueue deploy/helm/rustqueue \ --namespace rustqueue \ - --set queue.image=registry.example/rustqueue:0.9.0 \ + --set queue.image=registry.example/rustqueue:0.10.0 \ --set queue.rollout.requireCanaryApproval=true rustqueuectl -n rustqueue rollout approve @@ -113,7 +190,7 @@ Useful controls: rustqueuectl -n rustqueue rollout pause rustqueuectl -n rustqueue rollout resume rustqueuectl -n rustqueue rollout retry -rustqueuectl -n rustqueue rollout rollback registry.example/rustqueue:0.9.0 +rustqueuectl -n rustqueue rollout rollback registry.example/rustqueue:0.10.0 rustqueuectl -n rustqueue rollout forward ``` diff --git a/docs/releases/v0.10.0.md b/docs/releases/v0.10.0.md new file mode 100644 index 0000000..2479aad --- /dev/null +++ b/docs/releases/v0.10.0.md @@ -0,0 +1,204 @@ +# RustQueue 0.10.0 + +RustQueue 0.10.0 adds an optional best-effort WebSocket receive path for live +consumers of existing `TTL_DISCARD` Topics. Topic TTL discard itself was +introduced in 0.9.0 and remains the prerequisite; this release does not change +the NSQ publish contract, reliable Topics, or v7 storage format. + +## WebSocket TTL live subscription + +When enabled, each Broker's existing HTTP listener accepts one Topic per +connection: + +```text +GET /v1/ws/topics/{topic} +Sec-WebSocket-Protocol: rustqueue.live.v1 +``` + +The Topic must already exist on that Broker and use `deliveryMode=TTL_DISCARD`. +The endpoint never creates a Topic. Unknown Topics and `RELIABLE` Topics are +rejected. A client first calls Discovery `/lookup`, then connects directly to +every healthy Broker owner returned for that Topic; the producer proxy does +not multiplex this path. A client may union owner streams and deduplicate +migration overlap using the broker-scoped message ID in each frame. + +The stream starts at the current Topic tail. Messages published before the +connection, during disconnection, or after a failed write are not replayed. +There is no WebSocket ACK, reconnect resume, or WebSocket-layer retry. A +complete frame write means only that RustQueue finished the socket write; it is +not a business processing acknowledgement. + +### Ready boundary + +The HTTP `101 Switching Protocols` response establishes only the transport. +The live subscription begins when the server sends the text control frame +`{"type":"ready",...}`. No message capture is promised before `ready`, +including while the server has sent `auth_required` and is waiting for the +client's `auth` control. Clients must not treat HTTP upgrade or +`auth_required` as the capture boundary. + +The `ready` payload identifies the Topic and subscription, and uses +`{"start":"tail","ack":"auto"}` to make the current-tail and automatic +post-write completion semantics explicit. + +Before `ready`, text controls are used for authentication. After `ready`, the +server sends complete binary data frames and may send terminal error controls. +The server may also send WebSocket Ping frames; clients should answer using +the standard WebSocket Pong behavior. + +### Binary data frame + +Each message is exactly one complete binary WebSocket frame. The first 24 bytes +are big-endian metadata followed by the original message body: + +| Offset | Size | Field | +| --- | ---: | --- | +| `0..4` | 4 | ASCII magic `RQW1` | +| `4..6` | 2 | Header length, fixed at `24` | +| `6..8` | 2 | Delivery attempts, unsigned `u16` | +| `8..16` | 8 | Broker timestamp, signed `i64` nanoseconds | +| `16..24` | 8 | Broker-scoped message ID, unsigned `u64` | +| `24..` | — | Original message body | + +### AUTH, Origin and errors + +WebSocket AUTH reuses the existing subscription permission with the virtual +Channel `websocket#ephemeral`. Service clients may send +`Authorization: Bearer ` during the HTTP upgrade. Browser clients can +complete the `auth_required` / `auth` text-control exchange. Credentials must +never be placed in a query parameter, and query parameters are rejected. + +Browser requests with an `Origin` header must exactly match +`websocket.allowed_origins`; a missing Origin is accepted for a non-browser +client but still requires normal authentication. Discovery's read-only +`/lookup` response uses the same allowed-origin policy. + +Handshake failures reuse the HTTP `ApiError` shape: the stable code is in the +JSON `message` field and the explanation is in `detail`. After upgrade, +terminal errors use a text control such as +`{"type":"error","code":"E_TOPIC_CLOSED","detail":"..."}` before the +connection closes: + +| Code | Meaning | +| --- | --- | +| `E_WS_QUERY_FORBIDDEN` | Query parameters are forbidden; use the path and headers. | +| `E_WS_ORIGIN` | Browser Origin is not an exact configured allowlist entry. | +| `E_WS_SUBPROTOCOL` | `rustqueue.live.v1` was not negotiated. | +| `E_BAD_TOPIC` | Topic does not exist on this Broker. | +| `E_WS_TTL_REQUIRED` | Existing Topic is not `TTL_DISCARD` or changed away from it. | +| `E_WS_CAPACITY` | Connection, Channel or frame-memory capacity is exhausted; back off. | +| `E_WS_CONTROL` | Unsupported client text/binary data frame was received. | +| `E_WS_HEARTBEAT` | Peer missed the heartbeat liveness window; reconnect. | +| `E_AUTH_TIMEOUT` | `auth` was not received before the AUTH deadline. | +| `E_BAD_AUTH` | AUTH control was non-text, malformed, or missing a valid `auth`/`secret`. | +| `E_AUTH_OVERLOADED` | AUTH service or capacity gate is exhausted; retry later. | +| `E_AUTH_FAILED` | Authentication service or credential validation failed. | +| `E_UNAUTHORIZED` | AUTH does not permit the Topic subscription. | +| `E_TOPIC_CLOSED` | Topic was deleted, tombstoned, or closed. | +| `E_OWNER_CHANGED` | Topic owner/fence changed; perform a fresh lookup. | +| `E_DRAINING` | Broker is draining and closes the live subscription. | +| `E_SHUTDOWN` | Broker is shutting down; reconnect after discovery. | + +### Capacity, TTL and close semantics + +The feature is default-off and uses bounded runtime defaults: + +```toml +[websocket] +enabled = false +max_connections = 1024 +max_connections_per_topic = 256 +frame_inflight_bytes = 134217728 +allowed_origins = [] +``` + +Each connection creates one ephemeral Channel and reserves at most one message. +If the global frame budget, total connection limit, per-Topic limit or Channel +limit is exhausted, admission fails closed with `E_WS_CAPACITY`. A failed, +cancelled or timed-out write releases the reservation and closes the +connection; RustQueue does not retry the frame over WebSocket. + +The write deadline is the smaller of the existing socket timeout and the +message's remaining absolute TTL. If a complete frame cannot be written before +that deadline, the connection closes rather than leaving a partial frame. TTL +changes take effect for subsequent deadline checks. Paused Topics stop taking +messages. Topic deletion, owner migration, a change to `RELIABLE`, Broker +draining, shutdown and heartbeat timeout close affected connections so the +client can discover a healthy owner again. + +The aggregate metrics include WebSocket connections, completed messages and +bytes, authentication failures, capacity rejections, slow-consumer closes and +non-TTL Topic rejections. `/v1/capabilities` advertises +`websocket_ttl_live_v1` for binaries that implement this protocol. + +## Kubernetes configuration + +The runtime `[websocket]` settings are exposed equivalently through Helm +`queue.websocket` values and the RustQueue Operator/CRD `websocket` spec: + +```sh +helm upgrade rustqueue deploy/helm/rustqueue \ + --namespace rustqueue --reuse-values \ + --set queue.websocket.enabled=true \ + --set queue.websocket.maxConnections=1024 \ + --set queue.websocket.maxConnectionsPerTopic=256 \ + --set queue.websocket.frameInflightBytes=134217728 \ + --set-string 'queue.websocket.allowedOrigins[0]=https://app.example' +``` + +The equivalent Operator/CRD fragment is: + +```yaml +spec: + websocket: + enabled: true + maxConnections: 1024 + maxConnectionsPerTopic: 256 + frameInflightBytes: 134217728 + allowedOrigins: + - https://app.example +``` + +The existing Broker HTTP Service and NetworkPolicy are reused. The chart adds +no port and no default Ingress for this feature. Keep the Origin allowlist +narrow and put external browser access behind the deployment's normal TLS and +authentication boundary. Apply the checked-in CRDs before using new WebSocket +fields on a cluster where the CRDs are already installed; Helm does not +upgrade an existing CRD automatically. + +## Upgrade from 0.9.0 + +No Topic data migration is required. Existing Topics retain their persisted +policy, and WebSocket support remains disabled unless explicitly enabled. The +NSQ TCP/HTTP endpoints, publish acknowledgement modes, Topic TTL semantics and +v7 storage compatibility remain unchanged. + +For a Kubernetes rollout, update the CRDs first, then roll the 0.10.0 image: + +```sh +kubectl apply -f deploy/helm/rustqueue/crds +kubectl wait --for=condition=Established \ + crd/rustqueuetopics.rustqueue.io --timeout=60s +helm upgrade rustqueue deploy/helm/rustqueue \ + --namespace rustqueue \ + --set queue.image=registry.example/rustqueue:0.10.0 +``` + +After the Broker fleet is Ready, enable the `queue.websocket` settings in a +separate change if live subscriptions are needed. Consumers should continue +to use `/lookup` and direct owner connections; they must not assume a single +Broker or a replay window during rollout. A 0.9.0 binary that does not +advertise `websocket_ttl_live_v1` simply keeps the optional endpoint +unavailable while the existing NSQ and TTL paths continue to work. + +## Release assets + +Every 0.10.0 release contains the following five assets: + +- `rustqueue-0.10.0-linux-x86_64.tar.gz`: Linux x86_64 binaries, Console UI and + example configuration +- `rustqueue-0.10.0-linux-aarch64.tar.gz`: Linux ARM64 binaries, Console UI and + example configuration +- `rustqueue-0.10.0-source.tar.gz`: source archive for the tagged commit +- `rustqueue-0.10.0.tgz`: Helm Chart +- `SHA256SUMS-0.10.0`: SHA-256 checksums for all downloadable assets diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock index c2acc47..2a4e745 100644 --- a/fuzz/Cargo.lock +++ b/fuzz/Cargo.lock @@ -191,6 +191,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", + "base64", "bytes", "form_urlencoded", "futures-util", @@ -209,8 +210,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1", "sync_wrapper", "tokio", + "tokio-tungstenite", "tower", "tower-layer", "tower-service", @@ -490,6 +493,12 @@ dependencies = [ "syn", ] +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + [[package]] name = "defmt" version = "1.1.1" @@ -970,6 +979,12 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "http-range-header" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9171a2ea8a68358193d15dd5d70c1c10a2afc3e7e4c5bc92bc9f025cebd7359c" + [[package]] name = "httparse" version = "1.10.1" @@ -1473,6 +1488,16 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "mime_guess" +version = "2.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7c44f8e672c00fe5308fa235f821cb4198414e1c77935c1ab6948d3fd78550e" +dependencies = [ + "mime", + "unicase", +] + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -1790,10 +1815,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" dependencies = [ "libc", - "rand_chacha", + "rand_chacha 0.3.1", "rand_core 0.6.4", ] +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha 0.9.0", + "rand_core 0.9.5", +] + [[package]] name = "rand" version = "0.10.2" @@ -1815,6 +1850,16 @@ dependencies = [ "rand_core 0.6.4", ] +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + [[package]] name = "rand_core" version = "0.6.4" @@ -1824,6 +1869,15 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "rand_core" version = "0.10.1" @@ -2017,7 +2071,7 @@ dependencies = [ [[package]] name = "rustqueue-discovery" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "axum", @@ -2033,6 +2087,7 @@ dependencies = [ "serde_json", "sha2", "tokio", + "tower-http", "tracing", "tracing-subscriber", ] @@ -2055,7 +2110,7 @@ dependencies = [ [[package]] name = "rustqueue-protocol" -version = "0.9.0" +version = "0.10.0" dependencies = [ "bytes", "serde", @@ -2065,7 +2120,7 @@ dependencies = [ [[package]] name = "rustqueue-proxy" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "axum", @@ -2085,7 +2140,7 @@ dependencies = [ [[package]] name = "rustqueue-queue" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "bytes", @@ -2105,7 +2160,7 @@ dependencies = [ [[package]] name = "rustqueue-storage" -version = "0.9.0" +version = "0.10.0" dependencies = [ "anyhow", "crc32c", @@ -2117,7 +2172,7 @@ dependencies = [ [[package]] name = "rustqueue-telemetry" -version = "0.9.0" +version = "0.10.0" dependencies = [ "serde", ] @@ -2318,6 +2373,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha1" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest", +] + [[package]] name = "sha2" version = "0.10.9" @@ -2529,6 +2595,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -2569,11 +2647,19 @@ dependencies = [ "base64", "bitflags 2.13.1", "bytes", + "futures-core", "futures-util", "http", "http-body", + "http-body-util", + "http-range-header", + "httpdate", "mime", + "mime_guess", + "percent-encoding", "pin-project-lite", + "tokio", + "tokio-util", "tower", "tower-layer", "tower-service", @@ -2674,6 +2760,22 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.5", + "sha1", + "thiserror", +] + [[package]] name = "typenum" version = "1.20.1" @@ -2686,6 +2788,12 @@ version = "0.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" +[[package]] +name = "unicase" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" + [[package]] name = "unicode-ident" version = "1.0.24" diff --git a/rustqueue.example.toml b/rustqueue.example.toml index ef81feb..789a2ae 100644 --- a/rustqueue.example.toml +++ b/rustqueue.example.toml @@ -13,6 +13,15 @@ snappy_enabled = true deflate_enabled = true max_deflate_level = 6 +[websocket] +# Best-effort live subscriptions are opt-in and only accept TTL_DISCARD Topics. +enabled = false +max_connections = 1024 +max_connections_per_topic = 256 +frame_inflight_bytes = 134217728 +# Browser Origin values are exact matches; service clients without Origin are allowed. +allowed_origins = [] + [storage] data_path = "/data" feature_level = 1 diff --git a/scripts/acceptance-multi-broker-k8s.sh b/scripts/acceptance-multi-broker-k8s.sh index 6473fda..1aa3efe 100755 --- a/scripts/acceptance-multi-broker-k8s.sh +++ b/scripts/acceptance-multi-broker-k8s.sh @@ -343,9 +343,9 @@ require jq } if [[ "$BUILD_IMAGES" == "1" ]]; then - BUILD_VERSION=0.9.0-e2e-a MAX_STORAGE_FEATURE_LEVEL=1 make image + BUILD_VERSION=0.10.0-e2e-a MAX_STORAGE_FEATURE_LEVEL=1 make image docker tag rustqueue:dev "$BROKER_IMAGE_A" - BUILD_VERSION=0.9.0-e2e-b MAX_STORAGE_FEATURE_LEVEL=2 make image-from-dist + BUILD_VERSION=0.10.0-e2e-b MAX_STORAGE_FEATURE_LEVEL=2 make image-from-dist docker tag rustqueue:dev "$BROKER_IMAGE_B" [[ "$(docker image inspect "$BROKER_IMAGE_A" -f '{{.Id}}')" != \ "$(docker image inspect "$BROKER_IMAGE_B" -f '{{.Id}}')" ]] || { diff --git a/site/index.html b/site/index.html index 64efba8..67a4e14 100644 --- a/site/index.html +++ b/site/index.html @@ -183,6 +183,10 @@

Every broker owns its queue.

Consumers discover every broker that owns a Topic, then connect directly over the NSQ protocol.

+

+ TTL_DISCARD Topics also support an optional rustqueue.live.v1 WebSocket + live feed: best effort, current-tail only, and direct to each discovered owner. +

One Helm command. 
helm upgrade --install rustqueue deploy/helm/rustqueue \
   --namespace rustqueue --create-namespace \
-  --set queue.image=registry.example/rustqueue:0.9.0 \
+  --set queue.image=registry.example/rustqueue:0.10.0 \
   --set queue.storageClassName=ssd-rwo