diff --git a/docs/plan/message-bus.md b/docs/plan/message-bus.md index c58250ae..c28c96b7 100644 --- a/docs/plan/message-bus.md +++ b/docs/plan/message-bus.md @@ -16,6 +16,78 @@ editing is one consumer on top; folder-live updates, notifications, job progress, presence, and sync-client push invalidation follow with almost no extra scaffolding. +## Status โ€” 2026-09-11 + +The `feat/message-bus` branch delivers **D + F + follow-ups shipped +end-to-end** on the FE and BE, verified by S1โ€“S11 in the api-test +smoke suite plus manual multi-user E2E. Live today: + +- **Bus core** โ€” `MessageBus` port + `InProcessMessageBus` + + `NoopReplicator`. `๐Ÿ“ค bus publish` trace under + `RUST_LOG=oxicloud::message_bus=debug`. +- **WS handler** (`/api/rt/ws`) โ€” JSON-RPC 2.0, `rt.subscribe / + unsubscribe / event / revoked / ping / pong`, server-initiated RFC + 6455 keepalive Ping. +- **Auth for the WS upgrade** โ€” **F ticket flow shipped**. + `POST /api/rt/ticket` mints a one-shot 30 s ticket under the full + auth+DPoP+CSRF chain; the browser passes it via + `Sec-WebSocket-Protocol: oxi.ticket.`. Also accepts + `Authorization: Bearer ` for programmatic clients + (`rt-hurl-helper`). Route mounted OUTSIDE `protected_api` so the + standard DPoP-required middleware doesn't 401 browsers that can't + attach a `DPoP:` header to `new WebSocket()`. See + `handlers/rt_ws.rs` module doc. +- **Events firing end-to-end** โ€” every `MessageBusEvent` variant + except CRDT-flavoured ones: + - `FileCreated / Renamed / Moved / Deleted` (via + `FileUploadService` + `FileManagementService`) + - `FolderCreated / Renamed / Moved / Deleted` (via `FolderService` + for direct paths; `TrashService` publishes `FolderDeleted` on the + trash-first path โ€” the FE hits that path via + `DELETE /api/folders/{id}`) + - `AuthzChanged` (via `ShareService::revoke_grant`) drives the + grant-revocation eviction cascade. +- **Grant-revocation eviction (Slice C)** โ€” WS handler + auto-subscribes each session to `user:{caller}:authz`; on + `AuthzChanged` the reader translates to `SessionOut::EvictFolders` + and emits `rt.revoked` per evicted topic. Scope is per-topic; the + session itself and unrelated subscriptions survive. +- **FE composables** โ€” `useTopic` (generic), `useFolderTopic` + (folder-view sugar with per-verb + `onRevoked` + `onReconnect` + handlers), `useReconnect` (session-level, fires after 2nd+ open). + Types generated from AsyncAPI via `@asyncapi/modelina` in + `frontend/src/lib/generated/message-bus/`; `check-message-bus-spec` + CI + `pre-pull-request` block on drift. +- **Folder-view live refresh** โ€” `+page.svelte` wires every + `onFile*`/`onFolder*` handler to `scheduleLiveReload`, 100 ms + coalesce. Revocation โ†’ toast + `goto('/files')`. Reconnect โ†’ + refetch via `onReconnect` (bridges the "events lost during outage + window" gap; see `project_message_bus_reconnect_gap` memory). + Actor-echo skip was REMOVED for multi-tab correctness โ€” refetch is + idempotent, ~30 ms per self-mutation. +- **Client-side resilience** โ€” jittered exponential backoff + (250 ms โ†’ 30 s cap), circuit breaker at 20 consecutive failures + (~5 minutes of retry โ€” covers a cargo-release restart), `untrack` + in every mutation entry point so `$state` reads don't leak into + caller `$effect` deps. +- **AuthZ tested** โ€” S3 (folder no_read), S4 (nonexistent folder = + anti-enum parity), S9 (cross-user identity topic โ†’ `topic_forbidden`). +- **Ticket tested** โ€” S10 (happy path), S11 (single-use replay + rejected). + +Deferred and still open โ€” see the Roadmap section and the +`project_message_bus_reconnect_gap` memory: + +- **Notifications table + bell** (E) โ€” topic + producer + auto-sub + land here. Same pattern as `:authz`. +- **Presence** (Phase B) โ€” `folder:{id}:presence` topic + awareness + frames. +- **Yjs collab** โ€” `docs/plan/markdown-collab.md`, depends on the + binary-frame routing this plan sketches but doesn't ship. +- **Broker replicator** (Postgres LISTEN/NOTIFY or Redis) โ€” for + multi-instance and durable event log. `BusReplicator` port + declared, `NoopReplicator` wired today. + ## Non-goals - Persistent event log with "you missed these" replay. Durable state @@ -352,25 +424,61 @@ everywhere. ## Frontend components -### 1. Singleton client (`lib/stores/message-bus.svelte.ts`) +### 1. Singleton client (`lib/message-bus/client.svelte.ts`) -- Fetches a ticket via `POST /api/rt/ticket` (through `apiFetch`, so - DPoP is applied). -- Opens `wss:///api/rt/ws?ticket=โ€ฆ`. +Location is `lib/message-bus/` (subsystem dir, mirrors `lib/auth/` and +`lib/upload/` โ€” see `frontend/AGENTS.md`), NOT `lib/stores/` โ€” the +client is subsystem-scoped plumbing, not global reactive state that +routes read from. + +- **Fetches a ticket** via `POST /api/rt/ticket` (through `apiFetch`, + so DPoP is applied; `getCsrfHeaders()` merged in for the state- + changing POST). Ticket then passes on the WS upgrade via + `Sec-WebSocket-Protocol: oxi.ticket.` (NOT a query param โ€” + keeps the token off access logs and out of Referer / URL bar). +- Opens `wss:///api/rt/ws` with the subprotocol. - **Refcounted subscriptions**: - `subs: Map }>`. -- On subscribe by first component: send frame; on last unsubscribe: - send frame. -- On reconnect: reissue ticket, re-establish WS, re-send `subscribe` - for every live topic โ€” components don't care. -- Backoff: exponential (250 ms โ†’ 30 s), full-jitter. -- Health: `$state({ connected, latencyMs, subscribedTopics })` - exposed for a debug indicator. + `#subs: Map`. +- On first refcount of a topic: send `rt.subscribe`; on last drop: + send `rt.unsubscribe`. +- **On reconnect**: replay every already-known topic (client-side + state survives the disconnect); fire `onReconnect` handlers so + consumers refetch and catch up on events dropped during the + outage window. +- **Backoff**: exponential (250 ms โ†’ 30 s), full-jitter. Circuit + breaker at 20 consecutive failures (~5 minutes of retry โ€” + comfortably covers a cargo-release restart); trip logs one `error` + line and stops until `messageBus.reconnect()` is called or the + page reloads. +- **Reactive-safety rule**: every mutation entry point + (`subscribe`, `onReconnect`, `reconnect`, `close`, `#call`) wraps + its `$state` reads in `untrack(() => โ€ฆ)`. Without this a caller's + `$effect` inherits a hidden dep on `state`, and each transition + (idle โ†’ connecting โ†’ connected โ†’ disconnected โ†’ โ€ฆ) re-fires the + effect โ€” an observed 1000+/s loop on server-down. See the + `feedback-svelte5-untrack-mutation-methods` memory for the + general rule and the docstring on `subscribe` for the concrete + case. +- **Health**: `state = $state` + + `latencyMs = $state` exposed for a future debug + indicator (no UI consumes them yet โ€” silent MVP). -### 2. Composable (`lib/composables/useTopic.ts`) +### 2. Composables ```ts -useTopic(`folder:${folderId}`, (evt) => { /* mutate local $state */ }); +// Generic โ€” subscribe to any topic. +useTopic(topic, onEvent, onRevoked?) + +// Folder-view sugar โ€” per-verb handlers + reconnect hook. +useFolderTopic(() => folderId, { + onFileCreated, onFileRenamed, onFileMoved, onFileDeleted, + onFolderCreated, onFolderRenamed, onFolderMoved, onFolderDeleted, + onRevoked, // grant revoked, subscription evicted server-side + onReconnect, // WS came back; consumers refetch to catch up +}) + +// Session-level reconnect (fires on 2nd+ open, never initial). +useReconnect(cb) ``` Handles `$effect` lifecycle (subscribe on mount, unsubscribe on @@ -1047,28 +1155,44 @@ this baseline once the baseline is green. Ships the infrastructure and the two most visible consumers together. -- Bus port + `InProcessMessageBus` + `NoopReplicator` + WS handler - + ticket endpoint. -- Frontend singleton + `useTopic` composable. -- Topics live: `folder:{id}`, `user:{u}:notifications`, `job:{id}`, - `collab:{file_id}`, `collab:{file_id}:awareness`. -- **Folder-live updates**: `FolderService` / `FileManagementService` - publish `file.created` / `file.deleted` / `file.renamed` / - `file.moved` after commit; FE folder view subscribes and mutates - local state โ€” no manual refresh. -- **Job dashboard live**: `JobRegistry` publishes step progress and - terminal state; FE job dashboard subscribes and replaces the - current polling. -- **Notifications table + bell**: new `notifications` table + - `NotificationService` port; initial ingesters for `share-granted`, - `new-login-from-new-device`, `job-completed-for-you`, - `storage-quota-threshold`. FE bell with unread count, slide-out - panel, toast pop on receive. -- **MD collab editor**: see companion plan - `docs/plan/markdown-collab.md` โ€” depends on this phase's WS - handler + binary frame routing. +- **โœ… Bus port** + `InProcessMessageBus` + `NoopReplicator` + WS + handler + **ticket endpoint (F)**. +- **โœ… Frontend singleton** + `useTopic` + `useFolderTopic` + + `useReconnect` composables. `oxi:message-bus` logger namespace. +- **Topics live today**: `folder:{id}`, `user:{u}:authz`. +- **Topics reserved but not producing**: `user:{u}:notifications`, + `job:{id}`, `collab:{file_id}`, `collab:{file_id}:awareness` โ€” + land with their consumers below. +- **โœ… Folder-live updates**: `FolderService` / `FileUploadService` / + `FileManagementService` / `TrashService` publish `file_created / + renamed / moved / deleted` and `folder_created / renamed / moved / + deleted` after commit; FE folder view refetches on receipt (100 ms + coalesce, idempotent). Multi-user + multi-tab verified. +- **โœ… Grant-revocation eviction** (Slice C): `AuthzChanged` โ†’ + per-topic `rt.revoked`; folder view toasts + navigates to + `/files`. Session survives; unrelated subs unaffected. +- **โœ… Refetch-on-reconnect**: `messageBus.onReconnect(cb)` โ†’ + `useReconnect` composable โ†’ folder view refetches after WS comes + back. Bridges the in-memory-bus "events lost during outage" gap + (see `project_message_bus_reconnect_gap` memory). +- **Job dashboard live** โ€” TODO. `JobRegistry` publishes step + progress and terminal state; FE job dashboard subscribes and + replaces polling. +- **Notifications table + bell** โ€” TODO (Slice E). New + `notifications` table + `NotificationService` port; initial + ingesters for `share-granted`, `new-login-from-new-device`, + `job-completed-for-you`, `storage-quota-threshold`. FE bell with + unread count, slide-out panel, toast pop on receive. Auto-subscribe + to `user:{u}:notifications` server-side, same pattern as + `user:{u}:authz` today. +- **MD collab editor** โ€” TODO. See companion plan + `docs/plan/markdown-collab.md`. Depends on binary-frame routing + which this plan sketches but doesn't ship (`rt_ws.rs` today drops + binary frames with a debug log). -Deliverables sized ~4 weeks end-to-end. +Deliverables sized ~4 weeks end-to-end. Slice D (folder-live) and +Slice F (ticket flow) landed 2026-09-11. Slices E + collab are the +open work in Phase A. ### Phase B โ€” Presence + comments diff --git a/frontend/src/lib/message-bus/client.svelte.ts b/frontend/src/lib/message-bus/client.svelte.ts index fcc8982a..d3343487 100644 --- a/frontend/src/lib/message-bus/client.svelte.ts +++ b/frontend/src/lib/message-bus/client.svelte.ts @@ -104,10 +104,11 @@ const RECONNECT_MAX_MS = 30_000; * give up and stay `disconnected` until the caller explicitly asks * to `reconnect()`. Prevents an unrecoverable auth state (revoked * session, wrong CSRF cookie, missing DPoP nonce) from flooding - * logs. Ten attempts ร— exponential-backoff-with-jitter is roughly a - * minute of trying โ€” long enough for a transient blip, short enough - * to stop before it's noise. */ -const MAX_CONSECUTIVE_FAILURES = 10; + * logs. Twenty attempts ร— exponential-backoff-with-jitter caps + * around 5 minutes of retrying โ€” comfortably covers a cargo-release + * server restart on a hot machine while still short-circuiting a + * genuine permanent failure before it becomes noise. */ +const MAX_CONSECUTIVE_FAILURES = 20; interface SubEntry { count: number; diff --git a/frontend/src/routes/files/[...path]/+page.svelte b/frontend/src/routes/files/[...path]/+page.svelte index 9db4cf26..b7e11afa 100644 --- a/frontend/src/routes/files/[...path]/+page.svelte +++ b/frontend/src/routes/files/[...path]/+page.svelte @@ -438,23 +438,27 @@ } // โ”€โ”€ Live folder updates (message bus) โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ - // Subscribe to `folder:{currentId}` and refresh when another tab โ€” - // or another user with a share โ€” mutates something in this folder. - // The refresh call is coalesced through `#reloadScheduled` so a - // burst of events (e.g. a multi-file upload) collapses to a single - // fetch. Local mutations trigger `reload()` themselves, so events - // authored by this same user are dropped as echo (the `actor` on - // the event is the caller UUID from the server). + // Subscribe to `folder:{currentId}` and refresh when THIS session's + // tabs, another tab of the same user, or another user with a share + // mutates something in this folder. Refetch is coalesced through + // `reloadScheduled` so a burst of events (multi-file upload) collapses + // to a single fetch. + // + // Actor-echo skip was REMOVED: previously we skipped events whose + // `actor` equalled `session.user.id`, on the assumption "this tab + // already updated its state via the local mutation path". That is + // true for the ACTIVE tab, but it also silenced updates from OTHER + // TABS of the same user. Since `reload()` is idempotent (replaces + // `listing.files` with the same server state) the extra fetch on + // self-authored events costs one round-trip (~30 ms locally, never + // visible) and gains multi-tab correctness. The `reloadScheduled` + // coalescer already prevents redundant work when the local mutation + // path and the bus event race. // // See `docs/plan/message-bus.md ยง D` and the `useFolderTopic` // composable for the wiring. let reloadScheduled = false; - function scheduleLiveReload(actor: string): void { - // Actor echo: this same session's mutations already updated the - // listing through their own success path, so a re-fetch would - // only cost a round-trip. Other tabs of the same user still see - // the change (they render from their own state, not this one). - if (session.user?.id && actor === session.user.id) return; + function scheduleLiveReload(_actor: string): void { if (reloadScheduled) return; reloadScheduled = true; // Coalesce a burst; 100 ms is enough for the tail of a multi- @@ -493,12 +497,9 @@ // through the same `scheduleLiveReload` coalescer as event- // driven refreshes so a burst of reconnects (rare, but the // circuit breaker can produce one) collapses to a single - // fetch. Passing an actor of `null`-equivalent โ€” use an - // empty string so the echo-skip's `actor === user.id` - // check never matches. See - // `project_message_bus_reconnect_gap` memory. + // fetch. See `project_message_bus_reconnect_gap` memory. busLog.warn('reconnected โ€” refetching folder'); - scheduleLiveReload(''); + scheduleLiveReload('reconnect'); } });