From d20c79205610b563bad15fd586ce6ec03824f833 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Thu, 10 Sep 2026 01:33:37 +0200 Subject: [PATCH] feat(message-bus): add ping/keepalive on WS + root declaraiton on AsyncAPI - plan also eviction in case of permison revoked --- docs/config/env.md | 8 +++ docs/plan/message-bus.md | 44 ++++++++++++++-- example.env | 21 ++++++++ src/bin/generate-asyncapi.rs | 75 +++++++++++++++++++++++++- src/bin/rt-hurl-helper.rs | 19 +++++-- src/interfaces/api/handlers/rt_ws.rs | 78 +++++++++++++++++++++++++++- tests/api/rt_bus_check.sh | 51 +++++++++++++++++- tests/common/server.env | 7 +++ 8 files changed, 293 insertions(+), 10 deletions(-) diff --git a/docs/config/env.md b/docs/config/env.md index e261767e..7ca660da 100644 --- a/docs/config/env.md +++ b/docs/config/env.md @@ -355,6 +355,14 @@ Today's shipped locales: `ar, de, en, es, fa, fr, hi, it, ja, ko, nl, pl, pt, ru Example: `OXICLOUD_TRUST_PROXY_CIDR=127.0.0.1/32,10.0.0.0/8,172.16.0.0/12` +## Realtime WebSocket + +| Variable | Default | Description | +|---|---|---| +| `OXICLOUD_RT_WS_KEEPALIVE_SECONDS` | `30` | Server-initiated protocol Ping interval on `/api/rt/ws`. Prevents intermediate proxies (Traefik, nginx, Cloudflare) and NAT boxes from reaping the TCP session as idle. Read at each WS connect — a change takes effect on new connections, no restart needed. Set `0` or any non-positive value to fall back to the default. | + +Tuning: 30 s is comfortably under nginx's 60 s `proxy_read_timeout` default and Cloudflare's 100 s hard limit. Behind Traefik with `respondingTimeouts.idleTimeout` bumped to `3600s` (as documented in the reverse-proxy setup), you can leave this at 30 s or raise it — the interval should sit at most half the smallest hop's idle timeout so a single missed Ping doesn't reap the connection. + ## Allocator Tuning These variables are read directly by **mimalloc**, not by OxiCloud's config parser. diff --git a/docs/plan/message-bus.md b/docs/plan/message-bus.md index 56ed4845..56df8f13 100644 --- a/docs/plan/message-bus.md +++ b/docs/plan/message-bus.md @@ -559,12 +559,50 @@ Follows the same shape as `generate-openapi`: - Generator produces spec covering the Phase-A-MVP surface only (`rt.subscribe` / `rt.unsubscribe` / `rt.ping` methods, - `rt.event` / `rt.revoked` notifications, `Folder(id)` and - `UserAuthz(u)` topics, `FileCreated` / `FolderCreated` events, - the error-code table). + `rt.event` notification, `Folder(id)` and `UserAuthz(u)` topics, + `FileCreated` / `FolderCreated` events, the error-code table, + `defaultContentType`, `securitySchemes.bearerAuth`, and a `ping` + operation with the `rt.pong` reply shape). - Adding a new topic/event/method later is an enum variant + serde derive → regenerate → commit. Same discipline as OpenAPI. +### AsyncAPI follow-ups (deferred) + +Land with their producer PRs; each is a small addition to +`generate-asyncapi.rs` alongside the code that emits it. + +- **`rt.revoked` notification** on the Folder + File channels — the + server-initiated eviction frame fired when a grant is revoked + mid-session ([[project-message-bus]] AuthZ eviction section). Ships + with the `AuthzChanged` publish hook in `ShareService::revoke`. + Wire shape is already fixed by the plan; the AsyncAPI additions are + a `RtRevokedNotification` message + a `receive`-action operation on + every resource-scoped channel that supports eviction. +- **Yjs binary frames — prose section** at the doc level: AsyncAPI + schemas can't fully describe the `[kind][doc_id][payload]` framing + (it's out-of-band from the JSON envelope), so a plain-English + section on the `Collab` channel description referring to + `docs/plan/markdown-collab.md § Wire protocol` is the pragmatic + documentation. Ships with the Collab channel definition when the + editor PR lands. +- **Server variable expansion** — add a `port` variable so local dev + URLs (`ws://localhost:8086/api/rt/ws`) can be expressed in tooling + that reads the AsyncAPI URL template. Trivial addition; not + blocking. +- **`defaultMessages` per channel** — AsyncAPI convention for + reducing per-operation `$ref` boilerplate as the message count + grows. Worth introducing once we hit ~10 messages per channel; MVP + has 6 on Folder, still legible. +- **Bindings on messages** — declare `bindings.ws.headers` on the + subscribe messages so tools can render the auth header shape (the + spec knows about it via `securitySchemes`, but per-message bindings + make it explicit at the point of use). +- **Reply message discrimination** — the `receiveFolderEvent` + operation could split into per-event-kind messages + (`RtFileCreatedEvent`, `RtFolderCreatedEvent`) instead of one + polymorphic `RtFolderEventNotification` with `oneOf`. Better + codegen for typed clients. Refactor when we generate an FE SDK. + ## AuthZ model (audit rules per AGENTS.md) ### The subscribe gate diff --git a/example.env b/example.env index 649443e5..d67ecc9f 100644 --- a/example.env +++ b/example.env @@ -1093,6 +1093,27 @@ OXICLOUD_WOPI_ENABLED=false # DEPRECATED — use OXICLOUD_TRUST_PROXY_CIDR instead #OXICLOUD_TRUST_PROXY_HEADERS= +# ----------------------------------------------------------------------------- +# REALTIME MESSAGE BUS (/api/rt/ws) +# ----------------------------------------------------------------------------- +# WebSocket endpoint for folder-live updates, notifications, and the +# collaborative editor. See docs/plan/message-bus.md for the JSON-RPC 2.0 +# wire protocol. + +# Server-initiated protocol Ping interval (seconds). Prevents intermediate +# proxies (Traefik, nginx, Cloudflare) and NAT boxes from reaping the TCP +# session as idle. Read at each WS connect — a change takes effect on new +# connections without restart. Set 0 (or any non-positive value) to fall +# back to the default. +# +# Tuning: the interval should sit at most half the smallest hop's idle +# timeout, so a single missed Ping doesn't kill the connection. Common +# floors: +# * nginx `proxy_read_timeout` default 60s → ping ≤ 30s +# * Cloudflare hard limit 100s → ping ≤ 45s +# * Traefik with idleTimeout bumped to 3600s → 30s is safely under +#OXICLOUD_RT_WS_KEEPALIVE_SECONDS=30 + # ----------------------------------------------------------------------------- # MEMORY ALLOCATOR TUNING (IMPORTANT FOR RAM USAGE) # ----------------------------------------------------------------------------- diff --git a/src/bin/generate-asyncapi.rs b/src/bin/generate-asyncapi.rs index 9cc25056..93e93bbd 100644 --- a/src/bin/generate-asyncapi.rs +++ b/src/bin/generate-asyncapi.rs @@ -59,12 +59,16 @@ Phase C (sync-client push, album live) extend the same channels — see "#.trim(), "license": { "name": "AGPL-3.0-or-later" }, }, + // Applied to every message that doesn't set its own — the JSON-RPC + // control frames are all `application/json`. Binary Yjs frames + // stay out of AsyncAPI (see the Server description for pointers). + "defaultContentType": "application/json", "servers": { "default": { "host": "{host}", "pathname": "/api/rt/ws", "protocol": "wss", - "description": "OxiCloud realtime bus WebSocket endpoint", + "description": "OxiCloud realtime bus WebSocket endpoint. Text frames are JSON-RPC 2.0. Binary frames (out of AsyncAPI scope) are Yjs sync protocol for the collab editor — see `docs/plan/markdown-collab.md`.", "variables": { "host": { "description": "Server host — replace with the deployment domain", @@ -78,6 +82,16 @@ Phase C (sync-client push, album live) extend the same channels — see "bindings": { "ws": { "subProtocol": "oxi.rt.v1" } }, + // Every request MUST be authenticated. Programmatic + // clients set `Authorization: Bearer ` on the WS + // upgrade (same header the REST API uses); browser + // clients — which can't set headers on `new WebSocket()` + // — will use the deferred ticket flow (a plain HTTP + // POST issues a short-lived one-shot ticket bound to + // the WS URL, see the plan's DPoP-gap section). + "security": [ + { "$ref": "#/components/securitySchemes/bearerAuth" } + ], } }, "channels": channels(), @@ -98,6 +112,7 @@ fn channels() -> Value { "SubscribeRequest": { "$ref": "#/components/messages/RtSubscribeRequest" }, "UnsubscribeRequest": { "$ref": "#/components/messages/RtUnsubscribeRequest" }, "PingRequest": { "$ref": "#/components/messages/RtPingRequest" }, + "PongResponse": { "$ref": "#/components/messages/RtPongResponse" }, "SubscribedResponse": { "$ref": "#/components/messages/RtSubscribedResponse" }, "ErrorResponse": { "$ref": "#/components/messages/RtErrorResponse" }, "FolderEvent": { "$ref": "#/components/messages/RtFolderEventNotification" }, @@ -149,6 +164,26 @@ fn operations() -> Value { "messages": [ { "$ref": "#/channels/Folder/messages/FolderEvent" } ] + }, + // Application-layer keepalive. Separate from the RFC 6455 Ping + // control frame the server sends on `OXICLOUD_RT_WS_KEEPALIVE_SECONDS` + // (which is transport-level and not modelled in AsyncAPI). This + // operation lets a client actively confirm the socket is + // end-to-end alive when transport-level Pings alone can't rule + // out a proxy black-hole. + "ping": { + "action": "send", + "channel": { "$ref": "#/channels/Folder" }, + "summary": "Application-level keepalive; `rt.pong` reply confirms end-to-end liveness", + "messages": [ + { "$ref": "#/channels/Folder/messages/PingRequest" } + ], + "reply": { + "channel": { "$ref": "#/channels/Folder" }, + "messages": [ + { "$ref": "#/channels/Folder/messages/PongResponse" } + ] + } } }) } @@ -188,6 +223,12 @@ fn components() -> Value { "contentType": "application/json", "payload": { "$ref": "#/components/schemas/RtErrorResponseBody" }, }, + "RtPongResponse": { + "name": "rt.pong", + "title": "Reply to rt.ping — `result.pong == true`", + "contentType": "application/json", + "payload": { "$ref": "#/components/schemas/RtPongResponseBody" }, + }, // ── Notifications (server → client) ───────────────────── "RtFolderEventNotification": { "name": "rt.event", @@ -201,10 +242,22 @@ fn components() -> Value { "RtUnsubscribeRequestBody": rpc_request_schema("rt.unsubscribe", topic_params_schema()), "RtPingRequestBody": rpc_request_schema("rt.ping", json!({ "type": "null" })), "RtSuccessResponseBody": rpc_success_response_schema(), + "RtPongResponseBody": rpc_pong_response_schema(), "RtErrorResponseBody": rpc_error_response_schema(), "RtFolderEventBody": folder_event_notification_schema(), "FileCreatedData": file_created_schema(), "FolderCreatedData": folder_created_schema(), + }, + // How the client authenticates. Handler side is `auth_middleware` + // — the same middleware every `/api/*` request goes through, so + // any JWT valid for REST is valid for WS. + "securitySchemes": { + "bearerAuth": { + "type": "http", + "scheme": "bearer", + "bearerFormat": "JWT", + "description": "OxiCloud JWT — same access_token minted by `POST /api/auth/login` (or the OPAQUE handshake). Programmatic clients set `Authorization: Bearer ` on the WS upgrade request. Browsers, which cannot set headers on `new WebSocket()`, will use the deferred ticket flow (`POST /api/rt/ticket` → short-lived one-shot ticket in the WS URL); see the plan's DPoP-gap section.", + } } }) } @@ -250,6 +303,26 @@ fn rpc_success_response_schema() -> Value { }) } +/// Reply to `rt.ping` — the shape pins `result.pong == true` so +/// contract tests can assert on it directly. +fn rpc_pong_response_schema() -> Value { + json!({ + "type": "object", + "required": ["jsonrpc", "id", "result"], + "properties": { + "jsonrpc": { "type": "string", "const": "2.0" }, + "id": { "type": ["integer", "string", "null"] }, + "result": { + "type": "object", + "required": ["pong"], + "properties": { + "pong": { "type": "boolean", "const": true } + } + }, + } + }) +} + fn rpc_error_response_schema() -> Value { // The `code`/`message` catalog is the stable public vocabulary — // any change here IS a wire break. Every entry mirrors diff --git a/src/bin/rt-hurl-helper.rs b/src/bin/rt-hurl-helper.rs index 62cf43d4..5eac6e96 100644 --- a/src/bin/rt-hurl-helper.rs +++ b/src/bin/rt-hurl-helper.rs @@ -261,6 +261,11 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> { } let mut events: Vec = Vec::new(); + // Count server-initiated protocol Pings so scenarios can assert the + // keepalive fires. tokio-tungstenite queues an auto-Pong on the next + // write path, so we don't need to send one ourselves; we just observe + // the frame. + let mut pings_received: usize = 0; let mut timed_out = false; let deadline = tokio::time::Instant::now() + args.timeout; @@ -290,9 +295,16 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> { } }; - let Message::Text(text) = msg else { - // Ignore ping/pong/binary; server may send close later. - continue; + let text = match msg { + Message::Text(t) => t, + Message::Ping(_) => { + // Server-initiated keepalive — observable proof that the + // interval is firing. tokio-tungstenite queues an + // auto-Pong on the next flush; nothing to do here. + pings_received += 1; + continue; + } + _ => continue, // pong/binary/close — not asserted on }; let value: Value = serde_json::from_str(&text) .map_err(|e| HelperError::Protocol(format!("bad frame: {e}: {text}")))?; @@ -333,6 +345,7 @@ async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> { let summary = json!({ "subscribed": subscribed, "events": events, + "pings_received": pings_received, "timed_out": timed_out, }); std::fs::write(path, serde_json::to_vec_pretty(&summary).unwrap()) diff --git a/src/interfaces/api/handlers/rt_ws.rs b/src/interfaces/api/handlers/rt_ws.rs index 839ff66b..510bf88f 100644 --- a/src/interfaces/api/handlers/rt_ws.rs +++ b/src/interfaces/api/handlers/rt_ws.rs @@ -35,7 +35,9 @@ use std::collections::HashMap; use std::sync::Arc; +use std::time::Duration; +use axum::body::Bytes; use axum::extract::State; use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; use axum::response::Response; @@ -44,6 +46,7 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use tokio::sync::mpsc; use tokio::task::JoinHandle; +use tokio::time::MissedTickBehavior; use uuid::Uuid; use crate::application::ports::authorization_ports::AuthorizationEngine; @@ -64,6 +67,31 @@ const MAX_SUBSCRIPTIONS_PER_CONNECTION: usize = 128; /// socket layer doesn't back-pressure into the bus's broadcast ring. const OUTBOUND_CHANNEL_CAPACITY: usize = 512; +/// Default server-initiated protocol Ping interval. Keeps intermediate +/// proxies (Traefik, nginx, Cloudflare) and NAT boxes from reaping the +/// TCP session as idle. 30 s sits comfortably under nginx's 60 s +/// default and Cloudflare's 100 s hard limit; behind Traefik we +/// document a much longer `idleTimeout` anyway. +/// +/// Overridable at server start via `OXICLOUD_RT_WS_KEEPALIVE_SECONDS` +/// — test suites drop it to a low value to exercise the keepalive path +/// within a bounded wall-clock. +const DEFAULT_KEEPALIVE_SECONDS: u64 = 30; + +/// Read the keepalive interval from env at connection time. Kept as a +/// function rather than a `LazyLock` so a running server with the env +/// var flipped picks it up on the NEXT connection without a restart — +/// useful for smoke tests that toggle the value on the fly. +fn keepalive_interval() -> Duration { + Duration::from_secs( + std::env::var("OXICLOUD_RT_WS_KEEPALIVE_SECONDS") + .ok() + .and_then(|s| s.parse().ok()) + .filter(|&n: &u64| n > 0) + .unwrap_or(DEFAULT_KEEPALIVE_SECONDS), + ) +} + // ════════════════════════════════════════════════════════════════════════════ // JSON-RPC 2.0 envelope types // ════════════════════════════════════════════════════════════════════════════ @@ -166,6 +194,42 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc = HashMap::new(); + // Server-initiated protocol Ping ticker — prevents intermediate + // proxies (Traefik, nginx, Cloudflare) and NAT boxes from reaping + // the TCP session as idle. Browsers can't send Ping control frames + // (the JS `WebSocket` API doesn't expose them), so the server owns + // this responsibility; the client's WS layer auto-Pongs. A truly + // dead peer surfaces on the next `socket.send` and breaks out of + // the loop the same way any WS error does — no pong-timeout + // tracking needed for MVP. + // + // ─────────────────────── Scaling note ──────────────────────────── + // This is a `tokio::time::interval` PER connection — not a thread. + // The tokio timer wheel handles arbitrary N intervals in O(1) and + // each Sleep future is ~150 bytes of state. Per-session task + // memory dominates at any interesting N (~1 KB stack), which is + // still trivial: 10 000 clients ≈ 12 MB total + ~333 Pings/sec + // spread across the worker pool. + // + // If a deployment ever hits 100 000+ concurrent WS AND the + // per-connection interval becomes a measurable cost, the swap is: + // 1. one global `tokio::spawn(async { interval.tick().await; ... })` + // task that scans a `DashMap>` + // registry and pings each session's mailbox on tick, + // 2. session tasks receive the mailbox signal in their `select!` + // and send `Message::Ping` from there (still per-session, so + // one slow socket doesn't block the whole fleet). + // Neither pattern change would touch the wire; both are same-file + // refactors. Don't do this until N genuinely warrants it — until + // then, per-connection is the standard tokio idiom for a reason. + let mut keepalive = tokio::time::interval(keepalive_interval()); + // Coalesce backlog if the runtime pauses (e.g. under heavy load) + // rather than firing a burst of Pings when it recovers. + keepalive.set_missed_tick_behavior(MissedTickBehavior::Delay); + // Discard the immediate first tick — the socket just opened; a + // client sending its opening `rt.subscribe` shouldn't race a Ping. + keepalive.tick().await; + loop { tokio::select! { // biased: process outbound before inbound so an event burst @@ -183,6 +247,15 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc { + // RFC 6455 Ping control frame. 0-byte payload is + // spec-legal and the smallest wire footprint. Client + // auto-Pongs; nothing to observe here on that. + if socket.send(Message::Ping(Bytes::new())).await.is_err() { + break; + } + } + incoming = socket.recv() => { match incoming { Some(Ok(Message::Text(txt))) => { @@ -199,7 +272,10 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc { - // Handled by axum's WebSocket state machine. + // Client Ping → axum auto-Pongs. Client Pong is + // the response to OUR keepalive Ping — nothing + // to do at the app layer; TCP + WS keep the + // pipe warm regardless. } Some(Ok(Message::Close(_))) | Some(Err(_)) | None => break, } diff --git a/tests/api/rt_bus_check.sh b/tests/api/rt_bus_check.sh index 75a660c6..6a1a1a30 100755 --- a/tests/api/rt_bus_check.sh +++ b/tests/api/rt_bus_check.sh @@ -7,7 +7,7 @@ # This script orchestrates it against a live oxicloud server: bootstraps # state with curl, exercises the bus, asserts on the helper's JSON output. # -# Four scenarios: +# Five scenarios: # S1 Positive delivery — subscribe to folder A, upload into A, see event. # S2 Topic isolation — subscribe to folder A only, upload into B and # then A; must see A's event only. @@ -16,6 +16,10 @@ # S4 Anti-enumeration — subscribe to a folder that does not exist; # must return the SAME wire reason (`no_read`) # as S3, per the plan's anti-enum invariant. +# S5 Server keepalive — 3 s of idle surfaces multiple RFC 6455 Ping +# frames from the server (proves the interval +# fires), and the session still delivers an +# event on the same subscription afterwards. # # Exit non-zero on any failure — run.sh treats that as a suite failure. # ───────────────────────────────────────────────────────────────────────────── @@ -216,4 +220,47 @@ if ! "$HELPER_BIN" expect-denied \ fi log "S4 OK" -log "All four realtime-bus scenarios passed." +# ── Scenario 5 — Server-initiated keepalive ───────────────────────────────── +# Verifies the WS handler sends RFC 6455 Ping control frames on the +# `OXICLOUD_RT_WS_KEEPALIVE_SECONDS` cadence (1 s in tests/common/server.env). +# Two invariants: +# (a) idling on a live subscription surfaces multiple Ping frames — the +# keepalive interval genuinely fires, not just at connect and never again. +# (b) after 3 s of app-layer idle + keepalive traffic, the session is +# still healthy: an upload's event still delivers cleanly. +# If the keepalive impl were broken (missed-tick burst, dead select! arm, +# stalled write on the socket), either (a) trips (0-1 pings observed) or +# (b) trips (event never arrives after idle). +log "S5: server-initiated keepalive fires on idle; session still delivers." +out_s5="$(mktemp -t rtbus_s5.XXXXXX)" +"$HELPER_BIN" subscribe-and-collect \ + --url "$ws_url" \ + --token "$user1_token" \ + --subscribe "folder:$folder_a" \ + --expect-events 1 \ + --timeout 6s \ + --output "$out_s5" & +helper_pid=$! +# 3 s of pure idle — with 1 s keepalive on the server, that's ~3 Pings. +sleep 3 +mkfile_in "$folder_a" "s5.txt" "$user1_token" +if ! wait "$helper_pid"; then + cat "$out_s5" >&2 || true + die "S5: helper did not observe the expected event after idle" +fi +# (a) At least 2 Pings during the 3 s idle. Tolerant floor: with 1 s +# interval and a first-tick discard, 2 is the minimum credible observation +# before flakiness (missed tick, timer coalesce) becomes a concern. +pings=$(jq -r '.pings_received' "$out_s5") +if [[ "$pings" -lt 2 ]]; then + cat "$out_s5" >&2 || true + die "S5: expected >=2 keepalive pings during 3 s idle, got $pings" +fi +# (b) Exactly one event on the folder A subscription, from the post-idle upload. +[[ "$(jq -r '.events | length' "$out_s5")" == "1" ]] \ + || { cat "$out_s5"; die "S5: expected 1 event after idle, got $(jq -r '.events | length' "$out_s5")"; } +[[ "$(jq -r '.events[0].data.parent_id' "$out_s5")" == "$folder_a" ]] \ + || die "S5: parent_id mismatch after idle" +log "S5 OK ($pings pings observed)" + +log "All five realtime-bus scenarios passed." diff --git a/tests/common/server.env b/tests/common/server.env index 77c6760d..3adc50ac 100644 --- a/tests/common/server.env +++ b/tests/common/server.env @@ -35,6 +35,13 @@ OXICLOUD_OIDC_ENABLED=false OXICLOUD_NEXTCLOUD_ENABLED=true +# Realtime WS keepalive interval (server-initiated protocol Ping). +# Production default is 30 s; tests drop it to 1 s so the S5 scenario +# in `tests/api/rt_bus_check.sh` can observe multiple keepalive frames +# arriving within a bounded (few-seconds) wall-clock. Only observed +# by `rt_ws_handler`, which reads it at each WS connect time. +OXICLOUD_RT_WS_KEEPALIVE_SECONDS=1 + # Multi-entry storage config — see docs/plan/storage-multi-entry.md. # `local_main` is FIRST so the boot fallback picks it when no active # pointer is set in the DB yet (fresh test DB). Its root_dir falls