From 2f98bdeddc6c91fc9a117115e540ac5257329ef1 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 00:57:41 -0700 Subject: [PATCH] Surface presence connection loss in the canvas pane (CL-7228) The presence SSE stream stays open even when join or heartbeat fails, so the canvas looked healthy while the client was unjoined. The hook now subscribes to onError and onRecovered, reports failures with room and principal context, and shows a quiet reconnecting caption after more than one retry. --- apps/web/src/app.css | 8 ++ apps/web/src/presence/use-presence-room.ts | 52 ++++++++- apps/web/src/shell/app-shell.tsx | 26 +++-- apps/web/src/shell/canvas-column.tsx | 19 ++++ apps/web/test/canvas-column.test.tsx | 32 ++++++ apps/web/test/use-presence-room.test.tsx | 119 ++++++++++++++++++++- docs/presence.md | 25 +++-- packages/presence/src/client.test.ts | 87 +++++++++++++++ packages/presence/src/client.ts | 23 ++++ 9 files changed, 368 insertions(+), 23 deletions(-) diff --git a/apps/web/src/app.css b/apps/web/src/app.css index 83428f4f3..0b7d386f6 100644 --- a/apps/web/src/app.css +++ b/apps/web/src/app.css @@ -1121,9 +1121,17 @@ select:disabled, .shell-canvas-pane-actions { display: flex; flex-shrink: 0; + align-items: center; gap: 0.25rem; } +.shell-artifact-presence-status { + padding: 0 0.35rem; + font-size: 0.75rem; + color: var(--muted-foreground); + white-space: nowrap; +} + .shell-profile-pane-hint { margin: 0; font-size: 0.75rem; diff --git a/apps/web/src/presence/use-presence-room.ts b/apps/web/src/presence/use-presence-room.ts index 58ba7871d..f72a17d87 100644 --- a/apps/web/src/presence/use-presence-room.ts +++ b/apps/web/src/presence/use-presence-room.ts @@ -5,7 +5,12 @@ // disconnect lifecycle to reason about per (tenant, surface) pair. import { useCallback, useEffect, useRef, useState } from "react"; import type * as Y from "yjs"; -import { connectPresence, type PresenceHandle } from "@corbits/presence/client"; +import { reportError } from "@corbits/error-sink"; +import { + connectPresence, + type PresenceError, + type PresenceHandle, +} from "@corbits/presence/client"; export interface PresenceRoomMember { readonly principalId: string; @@ -19,8 +24,12 @@ export interface PresenceRoomMember { readonly typing?: boolean; } +/** Healthy until two consecutive transport failures; a single retry stays quiet. */ +export type PresenceConnection = "ok" | "degraded"; + export interface PresenceRoom { readonly members: readonly PresenceRoomMember[]; + readonly connection: PresenceConnection; readonly publishCursor: ( x: number, y: number, @@ -40,6 +49,16 @@ export interface UsePresenceRoomOptions { readonly doc?: Y.Doc; /** Called for every confirmed server-side snapshot — see `formatSaveStateLine` in `@corbits/artifact-ui`. */ readonly onSaved?: (info: { version: number; savedAt: number }) => void; + /** The signed-in principal, carried on `reportError` so a failed join is scoped. */ + readonly principalId?: string; +} + +const TRANSIENT_FAILURE_BUDGET = 1; + +function presenceErrorMessage(error: PresenceError): string { + return error.status === undefined + ? `Presence ${error.operation} failed` + : `Presence ${error.operation} failed (${String(error.status)})`; } /** @@ -56,7 +75,9 @@ export function usePresenceRoom( options?: UsePresenceRoomOptions, ): PresenceRoom { const [members, setMembers] = useState([]); + const [connection, setConnection] = useState("ok"); const handleRef = useRef(null); + const consecutiveFailuresRef = useRef(0); // `onSaved` is read through a ref, not a `connectPresence` dependency: // a caller re-rendering with a fresh inline callback must never tear // down and reconnect the stream (it would re-fetch the whole doc state @@ -64,10 +85,14 @@ export function usePresenceRoom( // different room to sync. const onSavedRef = useRef(options?.onSaved); onSavedRef.current = options?.onSaved; + const principalIdRef = useRef(options?.principalId); + principalIdRef.current = options?.principalId; const doc = options?.doc; useEffect(() => { setMembers([]); + setConnection("ok"); + consecutiveFailuresRef.current = 0; if (tenantId === null || surface === null) { handleRef.current = null; return; @@ -88,8 +113,31 @@ export function usePresenceRoom( const unsubscribe = handle.subscribe((snapshot) => setMembers(snapshot as readonly PresenceRoomMember[]), ); + const unsubscribeErrors = handle.onError((error) => { + const extra: Record = {}; + if (principalIdRef.current !== undefined) { + extra.principalId = principalIdRef.current; + } + if (error.status !== undefined) extra.status = error.status; + reportError(new Error(presenceErrorMessage(error)), { + operation: `presence.${error.operation}`, + tenantId, + roomId: surface, + ...(Object.keys(extra).length > 0 ? { extra } : {}), + }); + consecutiveFailuresRef.current += 1; + if (consecutiveFailuresRef.current > TRANSIENT_FAILURE_BUDGET) { + setConnection("degraded"); + } + }); + const unsubscribeRecovered = handle.onRecovered(() => { + consecutiveFailuresRef.current = 0; + setConnection("ok"); + }); return () => { unsubscribe(); + unsubscribeErrors(); + unsubscribeRecovered(); handle.disconnect(); handleRef.current = null; }; @@ -108,5 +156,5 @@ export function usePresenceRoom( handleRef.current?.publishTyping(typing); }, []); - return { members, publishCursor, publishTyping }; + return { members, connection, publishCursor, publishTyping }; } diff --git a/apps/web/src/shell/app-shell.tsx b/apps/web/src/shell/app-shell.tsx index 7401589ae..05756d4ab 100644 --- a/apps/web/src/shell/app-shell.tsx +++ b/apps/web/src/shell/app-shell.tsx @@ -123,17 +123,20 @@ export function AppShell({ tenantId, canvasArtifact === null ? null : `artifact:${canvasArtifact.id}`, undefined, - artifactDoc === null - ? undefined - : { - doc: artifactDoc, - onSaved: (info) => - setArtifactSaveState({ - kind: "saved", - version: info.version, - savedAt: info.savedAt, - }), - }, + { + ...(viewerPrincipalId === null ? {} : { principalId: viewerPrincipalId }), + ...(artifactDoc === null + ? {} + : { + doc: artifactDoc, + onSaved: (info: { version: number; savedAt: number }) => + setArtifactSaveState({ + kind: "saved", + version: info.version, + savedAt: info.savedAt, + }), + }), + }, ); const editingCoworkers = artifactPresence.members .filter( @@ -213,6 +216,7 @@ export function AppShell({ {...(artifactDoc !== null ? { artifactDoc } : {})} artifactSaveState={artifactSaveStateWithEditors} onArtifactTyping={artifactPresence.publishTyping} + presenceConnection={artifactPresence.connection} /> ) : null} diff --git a/apps/web/src/shell/canvas-column.tsx b/apps/web/src/shell/canvas-column.tsx index 26478a5ce..7733d26cb 100644 --- a/apps/web/src/shell/canvas-column.tsx +++ b/apps/web/src/shell/canvas-column.tsx @@ -52,6 +52,7 @@ import type { ReactNode } from "react"; import type * as Y from "yjs"; import { useBench } from "../bench-context"; +import type { PresenceConnection } from "../presence/use-presence-room"; import { workbenchPath } from "../workbench-path"; import { ensureProfileDm, loadSharedWorkbenches } from "../profile-relations"; import type { @@ -90,6 +91,7 @@ export function CanvasColumn({ artifactDoc, artifactSaveState, onArtifactTyping, + presenceConnection = "ok", }: { readonly open: boolean; readonly profile: ProfileSubject | null; @@ -117,6 +119,8 @@ export function CanvasColumn({ readonly artifactSaveState?: ArtifactSaveState; /** Fired on local typing start/stop in the text editor, for the host to publish through presence's `typing` awareness field. */ readonly onArtifactTyping?: (typing: boolean) => void; + /** Quiet reconnecting caption when presence has failed more than once. */ + readonly presenceConnection?: PresenceConnection; }) { // `inert` rather than `aria-hidden`: a collapsed column has to be out of // both the accessibility tree and the tab order, and `aria-hidden` alone @@ -150,6 +154,7 @@ export function CanvasColumn({ {...(artifactDoc !== undefined ? { artifactDoc } : {})} {...(artifactSaveState !== undefined ? { artifactSaveState } : {})} {...(onArtifactTyping !== undefined ? { onArtifactTyping } : {})} + presenceConnection={presenceConnection} /> ) : routine !== null ? ( @@ -511,6 +516,7 @@ function ArtifactCanvasPane({ artifactDoc, artifactSaveState = { kind: "read-only" }, onArtifactTyping, + presenceConnection = "ok", }: { readonly artifact: CanvasArtifactContent; readonly focus: boolean; @@ -521,6 +527,7 @@ function ArtifactCanvasPane({ readonly artifactDoc?: Y.Doc; readonly artifactSaveState?: ArtifactSaveState; readonly onArtifactTyping?: (typing: boolean) => void; + readonly presenceConnection?: PresenceConnection; }) { const showEditor = showsTextEditor(artifact, artifactDoc); return ( @@ -533,6 +540,18 @@ function ArtifactCanvasPane({ {...(artifact.previewSrc !== undefined ? { previewSrc: artifact.previewSrc } : {})} + {...(presenceConnection === "degraded" + ? { + trailing: ( + + Reconnecting… + + ), + } + : {})} />
{ readonly canEdit?: boolean; }; readonly artifactDoc?: import("yjs").Doc; + readonly presenceConnection?: "ok" | "degraded"; }): Promise { await act(async () => { root.render( @@ -247,6 +248,9 @@ describe("canvas artifact pane: co-editing (CL-5958 phase 2)", () => { {...(props.artifactDoc !== undefined ? { artifactDoc: props.artifactDoc } : {})} + {...(props.presenceConnection !== undefined + ? { presenceConnection: props.presenceConnection } + : {})} /> @@ -359,4 +363,32 @@ describe("canvas artifact pane: co-editing (CL-5958 phase 2)", () => { expect(container.querySelector("textarea")).toBeNull(); expect(container.textContent).toContain("fetched content"); }); + + test("a degraded presence connection shows a quiet reconnecting caption", async () => { + await renderArtifact({ + artifact: { + id: "art_6", + title: "Notes", + rendererKind: "doc", + content: "fetched content", + }, + presenceConnection: "degraded", + }); + + expect(container.textContent).toContain("Reconnecting…"); + }); + + test("a healthy presence connection shows no reconnecting caption", async () => { + await renderArtifact({ + artifact: { + id: "art_7", + title: "Notes", + rendererKind: "doc", + content: "fetched content", + }, + presenceConnection: "ok", + }); + + expect(container.textContent).not.toContain("Reconnecting…"); + }); }); diff --git a/apps/web/test/use-presence-room.test.tsx b/apps/web/test/use-presence-room.test.tsx index dafe9d01e..63cf66b5d 100644 --- a/apps/web/test/use-presence-room.test.tsx +++ b/apps/web/test/use-presence-room.test.tsx @@ -3,21 +3,35 @@ // either changes, and exposes publishCursor/publishTyping as stable // callbacks. Mocks `@corbits/presence/client` directly rather than a real // EventSource/fetch — that transport is `@corbits/presence`'s own test -// responsibility, not apps/web's. +// responsibility, not apps/web's. CL-7228: also wires `onError`/`onRecovered` +// so a degraded connection is visible after more than one retry. -import { afterEach, describe, expect, mock, test } from "bun:test"; +import { afterAll, afterEach, describe, expect, mock, test } from "bun:test"; import { act, createElement, useRef } from "react"; import { createRoot } from "react-dom/client"; import type { Root } from "react-dom/client"; +import type { PresenceError } from "@corbits/presence/client"; +import type { UsePresenceRoomOptions } from "../src/presence/use-presence-room"; interface FakeHandle { roomUrl: string; disconnected: boolean; listener: ((members: unknown[]) => void) | null; + errorListeners: Set<(error: PresenceError) => void>; + recoveredListeners: Set<() => void>; cursorCalls: unknown[]; } const created: FakeHandle[] = []; +const reportErrorCalls: { error: unknown; context: Record }[] = + []; + +mock.module("@corbits/error-sink", () => ({ + reportError: (error: unknown, context: Record) => { + reportErrorCalls.push({ error, context }); + return "presence-error"; + }, +})); mock.module("@corbits/presence/client", () => ({ connectPresence: (options: { roomUrl: string }) => { @@ -25,6 +39,8 @@ mock.module("@corbits/presence/client", () => ({ roomUrl: options.roomUrl, disconnected: false, listener: null, + errorListeners: new Set(), + recoveredListeners: new Set(), cursorCalls: [], }; created.push(handle); @@ -36,6 +52,18 @@ mock.module("@corbits/presence/client", () => ({ handle.listener = null; }; }, + onError: (listener: (error: PresenceError) => void) => { + handle.errorListeners.add(listener); + return () => { + handle.errorListeners.delete(listener); + }; + }, + onRecovered: (listener: () => void) => { + handle.recoveredListeners.add(listener); + return () => { + handle.recoveredListeners.delete(listener); + }; + }, publishCursor: (cursor: unknown) => handle.cursorCalls.push(cursor), publishTyping: () => undefined, disconnect: () => { @@ -49,11 +77,33 @@ const { usePresenceRoom } = await import("../src/presence/use-presence-room"); afterEach(() => { created.length = 0; + reportErrorCalls.length = 0; +}); + +afterAll(() => { + mock.restore(); }); +function fireError(error: PresenceError): void { + const handle = created[created.length - 1]; + if (handle === undefined) throw new Error("no presence handle"); + act(() => { + for (const listener of handle.errorListeners) listener(error); + }); +} + +function fireRecovered(): void { + const handle = created[created.length - 1]; + if (handle === undefined) throw new Error("no presence handle"); + act(() => { + for (const listener of handle.recoveredListeners) listener(); + }); +} + function mountHook( initialTenantId: string | null, initialSurface: string | null, + options?: UsePresenceRoomOptions, ) { const container = document.createElement("div"); document.body.appendChild(container); @@ -71,7 +121,7 @@ function mountHook( tenantId: string | null; surface: string | null; }) { - latest = usePresenceRoom(tenantId, surface); + latest = usePresenceRoom(tenantId, surface, undefined, options); return null; } @@ -141,4 +191,67 @@ describe("usePresenceRoom", () => { harness.unmount(); expect(created[0]?.disconnected).toBe(true); }); + + test("stays ok through a single transient error", () => { + const harness = mountHook("tnt_1", "artifact:art_1", { + principalId: "prn_1", + }); + fireError({ operation: "join", status: 500 }); + + expect(harness.result()?.connection).toBe("ok"); + expect(reportErrorCalls).toHaveLength(1); + harness.unmount(); + }); + + test("marks the connection degraded after a second consecutive error", () => { + const harness = mountHook("tnt_1", "artifact:art_1", { + principalId: "prn_1", + }); + fireError({ operation: "join", status: 500 }); + fireError({ operation: "join" }); + + expect(harness.result()?.connection).toBe("degraded"); + expect(reportErrorCalls).toHaveLength(2); + expect(reportErrorCalls[1]?.context).toEqual({ + operation: "presence.join", + tenantId: "tnt_1", + roomId: "artifact:art_1", + extra: { principalId: "prn_1" }, + }); + harness.unmount(); + }); + + test("reports every error with room and principal context", () => { + const harness = mountHook("tnt_1", "artifact:art_1", { + principalId: "prn_1", + }); + fireError({ operation: "heartbeat", status: 404 }); + + expect(reportErrorCalls).toHaveLength(1); + const call = reportErrorCalls[0]; + expect(call?.error).toBeInstanceOf(Error); + expect((call?.error as Error).message).toBe( + "Presence heartbeat failed (404)", + ); + expect(call?.context).toEqual({ + operation: "presence.heartbeat", + tenantId: "tnt_1", + roomId: "artifact:art_1", + extra: { principalId: "prn_1", status: 404 }, + }); + harness.unmount(); + }); + + test("clears degraded state on successful rejoin", () => { + const harness = mountHook("tnt_1", "artifact:art_1", { + principalId: "prn_1", + }); + fireError({ operation: "join", status: 500 }); + fireError({ operation: "join", status: 500 }); + expect(harness.result()?.connection).toBe("degraded"); + + fireRecovered(); + expect(harness.result()?.connection).toBe("ok"); + harness.unmount(); + }); }); diff --git a/docs/presence.md b/docs/presence.md index 2836b8f76..9b26e40d8 100644 --- a/docs/presence.md +++ b/docs/presence.md @@ -102,20 +102,31 @@ dependency-injectable for testing. `apps/web/src/presence/use-presence-room.ts` is the thin React hook that wraps it — the only place apps/web talks to the presence package directly. Two composition sites use it today: -- `apps/web/src/pages/chat-page.tsx` connects to `workbench:` and - hands the snapshot to `@corbits/chat-ui`'s `ChatWorkspace` as - `presenceMembers`, rendered as a live avatar stack beside the workbench - header's static participant list. +- `apps/web/src/pages/chat-page.tsx` used to connect to `workbench:` + and hand the snapshot to `@corbits/chat-ui`'s `ChatWorkspace` as + `presenceMembers`. That roster now rides the workbench `/stream` + (`chat.presence` / `chat.presence.snapshot`, CL-6328); `@corbits/presence` + no longer backs the header stack. - `apps/web/src/shell/app-shell.tsx` connects to `artifact:` when the canvas has one open, publishes the pointer's fractional position over the artifact pane as `cursor`, and hands the room's snapshot to `CanvasColumn` as `presenceCursors` — colored, labeled dots overlaid on - the read-only artifact renderer. + the artifact renderer. + +The SSE stream opens and stays open regardless of whether join or a later +heartbeat actually succeeded, so a healthy-looking cursor overlay is not +proof the client is in the room. `PresenceHandle.onError` reports every +failed join/heartbeat/leave/update; `onRecovered` fires on a successful +join or heartbeat. `usePresenceRoom` subscribes to both: errors go through +`reportError` with tenant, room (`surface`), and principal context, and the +canvas pane shows a quiet "Reconnecting…" caption only after a second +consecutive failure. A single transient retry stays silent; a successful +rejoin clears the caption. Neither `@corbits/chat-ui` nor `apps/web/src/shell/canvas-column.tsx` imports `@corbits/presence` itself — both only take plain data -(`PresenceMember`, `PresenceCursor`) as props, the same way they take any -other host-supplied data. +(`PresenceMember`, `PresenceCursor`, and a `PresenceConnection` flag) as +props, the same way they take any other host-supplied data. ## Phase 2: co-editing diff --git a/packages/presence/src/client.test.ts b/packages/presence/src/client.test.ts index c864caf4e..2da35e5bb 100644 --- a/packages/presence/src/client.test.ts +++ b/packages/presence/src/client.test.ts @@ -490,6 +490,93 @@ describe("connectPresence: error reporting and rejoin (CL-7202)", () => { handle.disconnect(); }); + test("a successful join is reported through onRecovered", async () => { + let recovered = 0; + const handle = connectPresence({ + roomUrl: ROOM_URL, + fetchImpl: fakeFetch([]), + openEventSource: () => new FakeEventSource(), + }); + handle.onRecovered(() => { + recovered += 1; + }); + + await flushMicrotasks(); + + expect(recovered).toBe(1); + handle.disconnect(); + }); + + test("a failed join is not reported through onRecovered", async () => { + let recovered = 0; + const handle = connectPresence({ + roomUrl: ROOM_URL, + fetchImpl: scriptedFetch([], () => 500), + openEventSource: () => new FakeEventSource(), + }); + handle.onRecovered(() => { + recovered += 1; + }); + + await flushMicrotasks(); + + expect(recovered).toBe(0); + handle.disconnect(); + }); + + test("a successful heartbeat after a failed one is reported through onRecovered", async () => { + let heartbeatStatus = 500; + let recovered = 0; + const fetchImpl: PresenceFetch = (url) => { + const status = url.endsWith("/heartbeat") ? heartbeatStatus : 200; + return Promise.resolve({ + ok: status >= 200 && status < 300, + status, + json: () => Promise.resolve({}), + }); + }; + const handle = connectPresence({ + roomUrl: ROOM_URL, + fetchImpl, + openEventSource: () => new FakeEventSource(), + }); + handle.onRecovered(() => { + recovered += 1; + }); + await flushMicrotasks(); + expect(recovered).toBe(1); + + handle.publishTyping(true); + await flushMicrotasks(); + expect(recovered).toBe(1); + + heartbeatStatus = 200; + handle.publishTyping(false); + await flushMicrotasks(); + expect(recovered).toBe(2); + handle.disconnect(); + }); + + test("onRecovered's unsubscribe stops further delivery", async () => { + let recovered = 0; + const handle = connectPresence({ + roomUrl: ROOM_URL, + fetchImpl: fakeFetch([]), + openEventSource: () => new FakeEventSource(), + }); + const unsubscribe = handle.onRecovered(() => { + recovered += 1; + }); + await flushMicrotasks(); + expect(recovered).toBe(1); + unsubscribe(); + + handle.publishTyping(true); + await flushMicrotasks(); + expect(recovered).toBe(1); + handle.disconnect(); + }); + test("a client stuck unable to join backs off instead of re-posting on every publish call", async () => { let clock = 0; const calls: string[] = []; diff --git a/packages/presence/src/client.ts b/packages/presence/src/client.ts index 0dbb6ecde..b0026cc66 100644 --- a/packages/presence/src/client.ts +++ b/packages/presence/src/client.ts @@ -112,6 +112,14 @@ export interface PresenceHandle { subscribe(listener: (members: readonly PresenceState[]) => void): () => void; /** Subscribes to every failed join/heartbeat/leave/update request. */ onError(listener: (error: PresenceError) => void): () => void; + /** + * Subscribes to every successful join or heartbeat — the client's only + * signals that the connection is healthy again after failures. A consuming + * UI that showed a reconnecting/degraded caption from `onError` clears it + * here, not from the SSE snapshot (the stream stays open even while join + * is failing). + */ + onRecovered(listener: () => void): () => void; /** Leaves the room and tears down the stream/heartbeat timer. */ disconnect(): void; } @@ -220,6 +228,7 @@ export function connectPresence( const listeners = new Set<(members: readonly PresenceState[]) => void>(); const errorListeners = new Set<(error: PresenceError) => void>(); + const recoveredListeners = new Set<() => void>(); let latestMembers: readonly PresenceState[] = []; let disconnected = false; // Whether the last join attempt is known to have succeeded server-side. @@ -258,6 +267,10 @@ export function connectPresence( for (const listener of errorListeners) listener(error); }; + const notifyRecovered = () => { + for (const listener of recoveredListeners) listener(); + }; + const post = (operation: PresenceOperation, body: unknown) => fetchImpl(`${options.roomUrl}/${operation}`, { method: "POST", @@ -296,6 +309,7 @@ export function connectPresence( } joinFailureCount = 0; joined = true; + notifyRecovered(); flushPendingUpdates(); if (doc === undefined) return; void response.json().then((body) => { @@ -332,6 +346,7 @@ export function connectPresence( } return; } + notifyRecovered(); flushPendingUpdates(); }); } @@ -452,6 +467,13 @@ export function connectPresence( }; }, + onRecovered(listener) { + recoveredListeners.add(listener); + return () => { + recoveredListeners.delete(listener); + }; + }, + disconnect() { if (disconnected) return; disconnected = true; @@ -459,6 +481,7 @@ export function connectPresence( source.close(); listeners.clear(); errorListeners.clear(); + recoveredListeners.clear(); if (doc !== undefined && onLocalDocUpdate !== undefined) { doc.off("update", onLocalDocUpdate); }