diff --git a/dev/docs/yjs.md b/dev/docs/yjs.md index 90e635165d4..00b0daa6aac 100644 --- a/dev/docs/yjs.md +++ b/dev/docs/yjs.md @@ -191,7 +191,7 @@ considered. Client joining afterwards: -1. `hasContent(ydoc)` is `true` → return early, no hydration, no election, no 150 ms wait. (`useYjsSession.ts`) +1. `hasContent(ydoc)` is `true` → return early, no hydration, no seed request. (`useYjsSession.ts`) 2. Y.Doc lives locally, and syncs to the Yjs server if a provider exists ### Y.Doc @@ -299,27 +299,31 @@ state recovery may re-run hydration if the room turns out to be stale, i.e. `_oc the joining client just fetched. Two clients can arrive into an empty room at the same moment, and only one may seed it - otherwise the content lands -twice. The Yjs server decides which: a client asks over a stateless message and gets a grant or a refusal (see -`services/yjs/src/lib/seedGrant.ts`). The winner sets `_oc_meta.hydrated` before it seeds, so peers can react to the -incoming content rather than merge with it. +twice. The Yjs server decides which: a client asks for the `seed` grant over a stateless message and gets a grant or a +refusal (see `services/yjs/src/lib/grants.ts`). The winner sets `_oc_meta.hydrated` before it seeds, so peers can react +to the incoming content rather than merge with it. ### Stale recovery When a joining client finds that `_oc_meta.etag` no longer matches the etag it just fetched, the file changed outside -the room. It stamps `nativeEtag`, claims the job via `recoveryClientId` and raises `isStale`. Every peer's meta observer -fires, but only the elected one gets through: recovery wipes the room and re-seeds it, and the only peer holding the -body behind `nativeEtag` is the one that just fetched the file. Any other peer would publish the copy it opened with - -or its own last serialization - and then stamp the fresh etag onto it, so the next save would overwrite the external -writer with a matching `If-Match` and no conflict. +the room. Recovery wipes the room and re-seeds it, and only a peer holding the body behind the fresh etag may do that. +Any other peer would publish the copy it opened with - or its own last serialization. -The elected peer captures that body at detection time rather than reading `currentContent` when recovery runs. By then -the room has synced its own state into the Y.Doc and the debounced serialize has reported it straight back into -`currentContent`. +The detecting client captures that body at detection time rather than reading `currentContent` when recovery runs. -`recoveryClientId` is last-write-wins, so if several clients detect the same drift at once, exactly one of them survives -convergence. A peer that joins while `isStale` is already up offers itself the same way, provided its etag matches -`nativeEtag` - the observer only fires on change, so without that the room would stay stuck if the elected peer -navigated away mid-recovery. +Several clients can detect the same write at once, and only one may rewrite the room, or the body lands twice. Like +seeding, the Yjs server decides: each client asks for the `recover:` grant, and only the one granted goes on. It +raises `isStale` with `nativeEtag`, then resets, re-hydrates and commits, all in one go but as separate updates, so +peers see the flag before the rewrite. The others wrote nothing and simply receive the rewrite. A joining client waits +for the answer, and the rewrite, before its editor mounts. The grant lasts as long as the holder's connection and is not +passed on when it leaves, since nobody else holds its body. A peer that joins while `isStale` is still up recovers to +the etag it fetched itself, even if the flag names an older one - so the room does not stay stuck if the peer that +raised the flag left, or the file moved on since. + +A joining writer that does not rewrite the room itself - refused, or holding the room's etag - takes the room's etag +instead of the one it fetched. It still shows the room's old content, and with the fresh etag its next save +would overwrite the external write without a conflict. With the room's etag that save conflicts instead, unless the +rewrite lands first and brings the fresh etag along. Peers receive the rewrite as ordinary CRDT updates, and a peer holding unsaved work would lose it without a word. So a peer that is dirty when a _remote_ `isStale` goes up does not follow: it marks itself conflicted, stops its provider @@ -359,11 +363,12 @@ sequenceDiagram AW->>S: applyExternalUpdate {content, etag} Note over S: flush pending content report,
then: unsaved work? alt clean - S->>P: claim + isStale (one transaction) - S->>P: reset + hydrate, then commit with etag + S->>S: recover:etag grant from the Yjs server + Note over S: refused - another peer recovers, skip + S->>P: isStale, then reset + hydrate, then commit with etag S->>AW: onContentChange, onServerContentChange, onEtagChange else dirty - S->>P: isStale + nativeEtag, no claim + S->>P: isStale + nativeEtag, no recovery S->>S: markConflicted end ``` @@ -372,12 +377,13 @@ The wrapper does the cheap checks first, because most of these events are peer s The room's etag stamp settles those without a request, with a grace period (`EXTERNAL_UPDATE_ETAG_GRACE_MS`) for a stamp that is still on the way. Only a body the room cannot account for starts the recovery. -- A **clean** detector claims and flags the room in one transaction and runs the recovery from the body it fetched. - Clean peers follow the rewrite, dirty peers conflict, exactly as after a join-time detection. The recovering peer - then reports content, server content and etag to its own wrapper: the rewrite runs with the content reporter - suppressed and its commit carries an own origin, so nothing else would. -- A **dirty** detector flags the room _without_ claiming: it holds nothing safe to re-seed with. All peers are marked - dirty and the room stays flagged until the next join recovers it. +- A **clean** detector asks for the recovery grant and, if granted, runs the recovery from the body it fetched. Every + clean writer with the file open gets the same event, so this is where the grant matters most. Clean peers follow the + rewrite, dirty peers conflict, exactly as after a join-time detection. The recovering peer then reports content, + server content and etag to its own wrapper: the rewrite runs with the content reporter suppressed and the meta + observer only reacts to remote writes, so nothing else would. +- A **dirty** detector flags the room _without_ recovering: it holds nothing safe to re-seed with. All peers are marked + dirty, and a clean peer that fetched the file itself recovers it, otherwise the next join does. Before deciding, the session flushes its pending content report. The caller's dirty state lags the doc by `SERIALIZE_DEBOUNCE_MS`, and a keystroke from inside that window is unsaved work all the same. Peers do the same when @@ -407,7 +413,7 @@ A `Y.Map` alongside the editor content, used for coordination the editor never s | `savedStateVector` | last save, recovery | what that peer's doc held when it wrote | | `isStale` | any writer | the file changed outside this room; rehydrate | | `nativeEtag` | writer that noticed | the etag recovery should settle on | -| `recoveryClientId` | writer that noticed | which peer is elected to re-seed the room | +| `recoveryEpoch` | writer that noticed | bumped with every `isStale`, never cleared | | `hydrated` | the seeding peer | this room has been seeded | Every key lives in the shared Y.Doc, so a read-only peer's writes to it are @@ -499,7 +505,7 @@ hold the same content and duplicate the document for everyone. | `packages/web-pkg/src/editor/composables/useTextEditor.ts` | binds Tiptap to a Y.Doc and renders peer carets | | `packages/web-app-text-editor/src/yjs.ts` | the text editor's adapter and content-type detection | | `services/yjs/src/server.ts` | the Yjs server (Hocuspocus) | -| `services/yjs/src/lib/seedGrant.ts` | who may seed an empty room | +| `services/yjs/src/lib/grants.ts` | who may seed an empty room or recover a stale one | --- @@ -551,10 +557,14 @@ grace period turns into a spurious conflict dialog. The SSE path is less exposed stamp `EXTERNAL_UPDATE_ETAG_GRACE_MS`. **Stale recovery needs someone who holds the fresh body.** Only a peer that fetched the file after the write may re-seed -the room. With SSE that is every writer in the space who has the file open; a peer that conflicts flags the room so -the others fetch. Two gaps remain: when no tab of a space member is open, nobody hears the event and the room stays -flagged until a client opens the file; and when the claimant leaves mid-recovery, the other peers only retry once -something else makes them look (a later event, a join, a 412). +the room. With SSE that is every writer in the space who has the file open; each fetches on its own event, and the +server lets one of them rewrite. Gaps remain: + +- When nobody in the room hears the event, the room stays stale until a writer joins or a save hits the 412. +- When the grant holder leaves after it was granted but before it rewrote, the refused peers only retry once something + else makes them look (a later event, a join, a 412). +- The grant is keyed by the fresh etag. Two peers that fetched different etags after back-to-back writes can both be + granted, and if their rewrites cross, the room holds the body twice. **SSE is best effort.** File events reach the members of the space only, so a share recipient of a personal-space file depends on the owner's tab to flag the room. A restored file version produces no SSE event at all today. diff --git a/packages/web-pkg/src/composables/yjs/types.ts b/packages/web-pkg/src/composables/yjs/types.ts index 9beb2618484..73216ee2f6c 100644 --- a/packages/web-pkg/src/composables/yjs/types.ts +++ b/packages/web-pkg/src/composables/yjs/types.ts @@ -17,7 +17,7 @@ export interface YjsAdapter { * * **Must be synchronous.** Stale recovery wipes the shared document and * re-seeds it, and it relies on that running in one go: yielding partway - * through lets a remote update move the recovery claim. + * through lets a remote update land in the half-rewritten document. */ hydrate(ydoc: Y.Doc, content: string): void diff --git a/packages/web-pkg/src/composables/yjs/useYjsSession.ts b/packages/web-pkg/src/composables/yjs/useYjsSession.ts index ac1b20b3874..d8debc38d70 100644 --- a/packages/web-pkg/src/composables/yjs/useYjsSession.ts +++ b/packages/web-pkg/src/composables/yjs/useYjsSession.ts @@ -152,25 +152,6 @@ const META_KEY = '_oc_meta' const SERIALIZE_DEBOUNCE_MS = 300 /** How long to wait for the Yjs server before running the session locally. */ const CONNECT_TIMEOUT_MS = 10_000 -/** - * Transaction origins this session puts on its own meta writes. The meta - * observer uses them to tell a peer save (remote update, no string origin) - * apart from writes the caller already knows about or must not be told about. - */ -const LOCAL_SAVE_ORIGIN = 'local-save' -const STALE_RECOVERY_RESET_ORIGIN = 'stale-recovery-reset' -const STALE_RECOVERY_COMMIT_ORIGIN = 'stale-recovery-commit' -const HYDRATE_ORIGIN = 'hydrate' -const EXTERNAL_UPDATE_ORIGIN = 'external-update' -const ETAG_ADOPT_ORIGIN = 'etag-adopt' -const OWN_META_ORIGINS: unknown[] = [ - LOCAL_SAVE_ORIGIN, - STALE_RECOVERY_RESET_ORIGIN, - STALE_RECOVERY_COMMIT_ORIGIN, - HYDRATE_ORIGIN, - EXTERNAL_UPDATE_ORIGIN, - ETAG_ADOPT_ORIGIN -] /** * How long `wasWrittenByRoom` waits for a peer's etag stamp that may still be * in flight at conflict time. Only ever paid in full for a genuine external @@ -178,12 +159,53 @@ const OWN_META_ORIGINS: unknown[] = [ */ const ROOM_ETAG_GRACE_MS = 1_000 /** - * Stateless messages that arbitrate who seeds an empty room. Must match - * `SeedMessage` in `services/yjs/src/lib/seedGrant.ts`, keep the two in sync. + * Stateless messages through which the Yjs server grants a room-wide job to + * exactly one client, each followed by the grant key. Must match + * `GrantMessage` in `services/yjs/src/lib/grants.ts`, keep the two in sync. + */ +const GRANT_REQUEST = '_oc_grant_request:' +const GRANT_GRANTED = '_oc_grant_granted:' +const GRANT_DENIED = '_oc_grant_denied:' +const SEED_GRANT = 'seed' +/** + * The seed grant's payloads, unchanged since before recovery grants, so every + * client and server version can seed. Must match `SeedMessage`. + * + * TODO: switch to `GRANT_*` and drop these, see https://github.com/opencloud-eu/web/issues/3436 */ const SEED_REQUEST = '_oc_seed_request' const SEED_GRANTED = '_oc_seed_granted' const SEED_DENIED = '_oc_seed_denied' +/** + * @deprecated How long a recovery grant request waits for an answer. A server + * from before the recovery grant never answers, see `RECOVERY_ELECTION_MS`. + */ +const RECOVERY_GRANT_TIMEOUT_MS = 5_000 +/** + * @deprecated Fallback for a Yjs server without recovery grants, drop once + * none is around (v1.0.0). The clients that noticed the write elect one among + * themselves instead: the last write to `recoveryClientId` wins after this + * wait. Unlike the grant it can let two recoveries through. + */ +const RECOVERY_ELECTION_MS = 150 +function grantRequest(key: string) { + return key === SEED_GRANT ? SEED_REQUEST : GRANT_REQUEST + key +} +/** Key and verdict of a grant answer, or null for any other payload. */ +function parseGrantAnswer(payload: string): { key: string; granted: boolean } | null { + if (payload === SEED_GRANTED) return { key: SEED_GRANT, granted: true } + if (payload === SEED_DENIED) return { key: SEED_GRANT, granted: false } + if (payload.startsWith(GRANT_GRANTED)) { + return { key: payload.slice(GRANT_GRANTED.length), granted: true } + } + if (payload.startsWith(GRANT_DENIED)) { + return { key: payload.slice(GRANT_DENIED.length), granted: false } + } + return null +} +function recoveryGrant(etag: string) { + return `recover:${etag}` +} /** * How long an SSE-triggered check waits for a peer's etag stamp before it * treats the write as external. @@ -221,7 +243,7 @@ interface SessionMeta { isStale: boolean /** Etag of the fresh file body recovery must settle on. */ nativeEtag: string - /** The client elected to run the stale-state recovery. */ + /** @deprecated The client elected to recover, see `RECOVERY_ELECTION_MS`. */ recoveryClientId: number /** * Bumped with every `isStale`. Unlike the flag it is never cleared, so a @@ -349,12 +371,6 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { etag: string } - /** - * The native file body and its etag captured at the moment etag drift was detected. - * Recovery must publish exactly this body. - */ - let staleRecovery: RecoveryPayload | null = null - /** * The doc state behind the last content handed to the caller, i.e. what its * next PUT writes. Stamped into `_oc_meta.savedStateVector` after that PUT. @@ -426,62 +442,15 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { } /** - * Claim the stale-state recovery: capture the body it must publish (before - * the room's own state syncs in and gets reported back into - * `currentContent`) and name ourselves. Last write wins, so concurrent - * claims elect exactly one client. - */ - function claimRecovery( - doc: Y.Doc, - meta: SessionMetaMap, - payload: RecoveryPayload = { - content: toValue(currentContent), - etag: toValue(resource)?.etag ?? '' - }, - origin?: string - ) { - staleRecovery = payload - doc.transact(() => meta.set('recoveryClientId', doc.clientID), origin) - } - - /** - * Raise the staleness flag and, unless read-only, claim the recovery. The - * flag alone would strand the room - later joiners take the `isStale` early - * return in `runInitialHydration`, and without a claim and a `nativeEtag` - * to claim against, nothing ever clears it. A read-only client flags - * without claiming; the `nativeEtag` it leaves lets the first writer - * holding the same body pick the claim up. + * Raise the staleness flag for the file body behind `nativeEtag`. Peers + * holding unsaved work leave the room on it, see the meta observer. */ - function flagStale( - doc: Y.Doc, - meta: SessionMetaMap, - { - nativeEtag = toValue(resource)?.etag, - claim = !unref(effectiveReadOnly), - payload, - origin - }: { - nativeEtag?: string - claim?: boolean - payload?: RecoveryPayload - origin?: string - } = {} - ) { - // The claim must be complete before `isStale` goes up: raising the flag - // fires our own meta observer, whose recovery run needs the claim. - if (claim) claimRecovery(doc, meta, payload, origin) + function flagStale(doc: Y.Doc, meta: SessionMetaMap, nativeEtag: string) { doc.transact(() => { - if (nativeEtag) meta.set('nativeEtag', nativeEtag) + meta.set('nativeEtag', nativeEtag) meta.set('recoveryEpoch', (meta.get('recoveryEpoch') ?? 0) + 1) meta.set('isStale', true) - }, origin) - } - - /** Whether our freshly fetched etag is the one recovery must settle on. */ - function canSupplyRecoveryContent(meta: SessionMetaMap): boolean { - const target = meta.get('nativeEtag') - const ours = toValue(resource)?.etag - return Boolean(target && ours && target === ours) + }) } /** @@ -491,6 +460,8 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { */ function stopProvider(prov: HocuspocusProvider | null) { status.value = YjsStatus.Disconnected + // No answer can arrive any more. + abandonGrants() if (!prov) return try { prov.disconnect() @@ -572,43 +543,60 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { } /** - * The seed request this session is waiting on, if any. Also the marker for - * "a hydration is in flight": the request is the only thing hydration ever - * awaits. + * Grant requests waiting on the server's answer, by grant key. A pending + * request also marks a hydration or recovery in flight: they are what waits + * on it. */ - let pendingSeedGrant: ((granted: boolean) => void) | null = null + const pendingGrants = new Map void>() /** - * Abandon a request whose session is gone. Answered as a refusal so the - * hydration it belongs to unwinds without touching the new session's doc. + * Abandon requests that can no longer be answered. Answered as refusals so + * whatever waits on them unwinds without touching the doc. */ - function abandonSeedGrant() { - pendingSeedGrant?.(false) + function abandonGrants() { + for (const resolve of [...pendingGrants.values()]) resolve(false) } - /** Ask the Yjs server whether we may seed this room. */ - function requestSeedGrant(prov: HocuspocusProvider): Promise { - return new Promise((resolve) => { - pendingSeedGrant = (granted) => { - pendingSeedGrant = null + /** + * Ask the Yjs server whether we may do a job only one client in the room + * may do. A request already in flight for the same key is ours to finish, + * so a second one is refused right away. With `timeoutMs`, a request left + * unanswered resolves to null (deprecated, see `RECOVERY_GRANT_TIMEOUT_MS`). + */ + function requestGrant( + prov: HocuspocusProvider, + key: string, + timeoutMs?: number + ): Promise { + if (pendingGrants.has(key)) return Promise.resolve(false) + return new Promise((resolve) => { + const timer = + timeoutMs === undefined + ? undefined + : window.setTimeout(() => pendingGrants.get(key)?.(null), timeoutMs) + pendingGrants.set(key, (granted) => { + window.clearTimeout(timer) + pendingGrants.delete(key) resolve(granted) - } - prov.sendStateless(SEED_REQUEST) + }) + prov.sendStateless(grantRequest(key)) }) } /** - * The server's answer to a seed request, or an unsolicited grant when the - * previous holder left the room without seeding it. + * The server's answer to a grant request, or an unsolicited seed grant when + * the previous holder left the room without seeding it. */ - function onSeedMessage(doc: Y.Doc, prov: HocuspocusProvider, payload: string) { - if (payload !== SEED_GRANTED && payload !== SEED_DENIED) return - const granted = payload === SEED_GRANTED - if (pendingSeedGrant) { - pendingSeedGrant(granted) + function onGrantMessage(doc: Y.Doc, prov: HocuspocusProvider, payload: string) { + const answer = parseGrantAnswer(payload) + if (!answer) return + const { key, granted } = answer + const pending = pendingGrants.get(key) + if (pending) { + pending(granted) return } - if (granted) seedOnUnsolicitedGrant(doc, prov) + if (granted && key === SEED_GRANT) seedOnUnsolicitedGrant(doc, prov) } /** @@ -624,11 +612,11 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { doc.transact(() => { meta.set('hydrated', true) current.hydrate(doc, toValue(currentContent)) - }, HYDRATE_ORIGIN) + }) } catch (e) { // Withdraw the announce, or a read-only peer waits forever for a body // that is never coming. - doc.transact(() => meta.delete('hydrated'), HYDRATE_ORIGIN) + doc.transact(() => meta.delete('hydrated')) throw e } } @@ -670,36 +658,31 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { /** * Hydration: one client per room seeds the Y.Doc from native content, and - * the Yjs server picks which - see `requestSeedGrant`. In local mode there + * the Yjs server picks which - see `requestGrant`. In local mode there * is no room, so no permission is needed. */ async function runInitialHydration(doc: Y.Doc, prov: HocuspocusProvider | null) { const current = toValue(adapter) const meta = sessionMeta(doc) - // Already flagged stale: let the meta observer run the recovery, and skip - // the checks below so we don't race-lock a doc that is about to be - // rehydrated. The observer only fires on change, so a joiner that finds - // the flag already up offers itself as claimant - the elected peer may - // have navigated away before finishing. - if (meta.get('isStale') === true) { - if (!unref(effectiveReadOnly) && canSupplyRecoveryContent(meta)) { - claimRecovery(doc, meta) - void recoverFromStaleState(doc, prov) - } - return - } + // The body recovery must publish, captured now: once the room's own state + // syncs in, it gets reported back into `currentContent`. + const nativeEtag = toValue(resource)?.etag + const content = toValue(currentContent) - // Etag drift check. The Yjs server is relay-only and persists nothing, so - // the room's own `_oc_meta.etag` (seeded by whichever peer entered first) - // is compared against the etag the caller just fetched: the first peer - // seeds the baseline, a mismatch means the file on disk moved and flags - // recovery. `nativeEtag` lets recovery settle the final value without an - // extra fetch. + // Flagged stale, or etag drift. The Yjs server is relay-only and persists + // nothing, so the room's own `_oc_meta.etag` (seeded by whichever peer + // entered first) is compared against the etag the caller just fetched: a + // mismatch means the file on disk moved. A flag may be left by a peer + // that navigated away before finishing, so joiners retry it. const docEtag = meta.get('etag') - const nativeEtag = toValue(resource)?.etag - if (docEtag && nativeEtag && docEtag !== nativeEtag) { - flagStale(doc, meta) + if (meta.get('isStale') === true || (docEtag && nativeEtag && docEtag !== nativeEtag)) { + if (unref(effectiveReadOnly)) return + const recovered = + !!nativeEtag && + nativeEtag !== docEtag && + (await recoverFromStaleState(doc, prov, { content, etag: nativeEtag })) + if (!recovered) adoptRoomEtag(doc, meta) return } if (!docEtag && nativeEtag) { @@ -729,7 +712,7 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { // Let the server pick who seeds. Skipped in local mode. if (prov) { - if (!(await requestSeedGrant(prov))) return + if (!(await requestGrant(prov, SEED_GRANT))) return // Content may have landed while we waited. if (current.hasContent(doc)) return } @@ -737,45 +720,50 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { seedDoc(doc, current, meta) } - /** The recovery run in progress, so a second trigger queues behind it. */ - let recoveryInFlight: Promise | null = null + /** + * A writer joined a room that is behind the file on disk and did not + * rewrite it. The caller still holds the fresh etag, so its next save would + * put the room's old content over the external write with a matching + * `If-Match`. The room's etag makes that save conflict instead; the rewrite + * brings the fresh one along if it still comes. + */ + function adoptRoomEtag(doc: Y.Doc, meta: SessionMetaMap) { + if (doc.isDestroyed || unref(ydoc) !== doc) return + const roomEtag = meta.get('etag') + if (roomEtag) onEtagChange(roomEtag) + } /** - * Stale-state recovery, fired when `_oc_meta.isStale` goes up: the claimed - * client wipes the adapter content, clears the flags and re-hydrates from - * the body it captured at detection; peers receive the rewrite as ordinary - * CRDT updates. + * @deprecated Recovery arbitration for a Yjs server that never answers the + * recovery grant request, see `RECOVERY_ELECTION_MS`. + */ + async function winRecoveryElection(doc: Y.Doc, meta: SessionMetaMap) { + doc.transact(() => meta.set('recoveryClientId', doc.clientID)) + await new Promise((resolve) => window.setTimeout(resolve, RECOVERY_ELECTION_MS)) + return meta.get('recoveryClientId') === doc.clientID + } + + /** + * Stale-state recovery: rewrite the room from a fresh file body. The Yjs + * server grants it to one client per etag, so peers that noticed the same + * write at once never re-seed the room twice. The winner raises `isStale`, + * wipes the adapter content, re-hydrates and clears the flags; peers + * receive all of it as ordinary CRDT updates, the flag first. * - * Only the peer holding the fresh body may run this. Re-seeding from - * anything else would publish a pre-drift body and stamp the fresh etag - * onto it, so the next save would overwrite the external writer with a - * matching `If-Match` and no warning. Unreachable in local mode, but coded - * provider-tolerant so the two modes share one path. + * Only a client holding the body behind `etag` may run this. Re-seeding + * from anything else would publish a pre-drift body and stamp the fresh + * etag onto it, so the next save would overwrite the external writer with + * a matching `If-Match` and no warning. * * True when this run rewrote the room. */ async function recoverFromStaleState( doc: Y.Doc, - prov: HocuspocusProvider | null + prov: HocuspocusProvider | null, + { content, etag }: RecoveryPayload ): Promise { - if (unref(effectiveReadOnly) || unref(isConflicted)) return false - if (staleRecovery === null) return false - // A failed run must not reject its waiters too. - while (recoveryInFlight) await recoveryInFlight.catch(() => false) - if (staleRecovery === null || doc.isDestroyed) return false - const run = runStaleRecovery(doc, prov) - recoveryInFlight = run - try { - return await run - } finally { - if (recoveryInFlight === run) recoveryInFlight = null - } - } - - async function runStaleRecovery(doc: Y.Doc, prov: HocuspocusProvider | null): Promise { const current = toValue(adapter) const meta = sessionMeta(doc) - if (staleRecovery === null) return false if (typeof current.reset !== 'function') { lockForReload( prov, @@ -786,21 +774,27 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { return false } - // Let concurrent claims converge, then check we came out on top. - await new Promise((resolve) => setTimeout(resolve, 150)) - if (doc.isDestroyed || unref(ydoc) !== doc) return false - if (meta.get('isStale') !== true) { - // Someone else handled it. Drop the payload, or a later remote flag - // would read as one we hold the body for. - staleRecovery = null - return false + if (prov) { + const granted = + (await requestGrant(prov, recoveryGrant(etag), RECOVERY_GRANT_TIMEOUT_MS)) ?? + (await winRecoveryElection(doc, meta)) + if (!granted) return false } - if (meta.get('recoveryClientId') !== doc.clientID) { - staleRecovery = null + // The rewrite would wipe what was typed while waiting for the grant. + if (unref(isReady)) await activeReporter?.flush() + if (doc.isDestroyed || unref(ydoc) !== doc) return false + if (unref(effectiveReadOnly) || unref(isConflicted)) return false + // The rewrite for this etag already landed. + if (meta.get('etag') === etag) return false + if (unref(isReady) && toValue(hasUnsavedChanges)) { + flagStale(doc, meta, etag) + markConflicted() return false } - const { content, etag: freshEtag } = staleRecovery + // Flag, reset + hydrate and commit run in one go, as separate updates: + // dirty peers leave the room on the flag before the rewrite reaches them. + flagStale(doc, meta, etag) // Reset and hydrate share one transaction: mounted editors must never // observe the emptied doc, or ProseMirror pads it with an empty paragraph @@ -813,20 +807,20 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { doc.transact(() => { current.reset?.(doc) current.hydrate(doc, content) - }, STALE_RECOVERY_RESET_ORIGIN) + }) doc.transact(() => { meta.delete('isStale') meta.delete('nativeEtag') + // Deprecated, see `RECOVERY_ELECTION_MS`. meta.delete('recoveryClientId') - if (freshEtag) meta.set('etag', freshEtag) + meta.set('etag', etag) // The recovered body is what is on disk now, so clean peers take the // ordinary peer-save fan-out and stop counting it as unsaved work. meta.set('savedStateVector', Y.encodeStateVector(doc)) meta.set('lastSavedAt', Date.now()) - }, STALE_RECOVERY_COMMIT_ORIGIN) + }) - staleRecovery = null return true } catch (e) { // The reset already emptied the shared doc for every peer; `isStale` @@ -989,17 +983,15 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { prov.setAwarenessField('user', identity) return } - onSeedMessage(doc, prov, payload) + onGrantMessage(doc, prov, payload) }, onSynced() { clearConnectTimer() - // Reconnected inside the seed wait. The answer to the old socket is - // lost, so ask again on this one instead of starting a second - // hydration next to the one still waiting. - if (pendingSeedGrant) { - prov.sendStateless(SEED_REQUEST) - return - } + // Answers to the old socket are lost, so ask again on this one. + for (const key of pendingGrants.keys()) prov.sendStateless(grantRequest(key)) + // Reconnected while a hydration or recovery waits on the server: + // don't start a second one next to it. + if (pendingGrants.size > 0) return void onProviderSynced(doc, prov) } }) @@ -1071,7 +1063,7 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { doc.transact(() => { current.reset?.(doc) current.hydrate(doc, content) - }, STALE_RECOVERY_RESET_ORIGIN) + }) isRewritingDoc = false } catch (e) { // `reset` already committed, so the doc may be empty. Keep the rewrite @@ -1098,13 +1090,11 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { ) { const meta = sessionMeta(doc) return function metaObserver(event: Y.YMapEvent, transaction: Y.Transaction) { - // The initial sync replays the room's whole meta map through this - // observer; everything in it belongs to `runInitialHydration`, which - // reads the same keys with the context to act on them. - const isMidSession = unref(isReady) - // Remote ops carry no string origin, so this means "a peer is acting - // right now" - not the initial sync, not one of our own writes. - const isRemoteMetaWrite = isMidSession && !OWN_META_ORIGINS.includes(transaction.origin) + // "A peer is acting right now". Our own writes are acted on where they + // are made. The initial sync replays the room's whole meta map through + // this observer; everything in it belongs to `runInitialHydration`, + // which reads the same keys with the context to act on them. + const isRemoteMetaWrite = unref(isReady) && !transaction.local // Handled further down, but needed here to keep the fan-out out of it. const resyncWouldDropOurWork = event.keysChanged.has('recoveryEpoch') && isRemoteMetaWrite && toValue(hasUnsavedChanges) @@ -1171,32 +1161,19 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { return } - // Stale-state signal: every peer sees it, only the claimed one gets - // past the guards in `recoverFromStaleState`. An external update this - // client applies itself runs the recovery explicitly, so it can await it. - if (event.keysChanged.has('isStale') && meta.get('isStale') === true) { - if (transaction.origin === EXTERNAL_UPDATE_ORIGIN) return - if (isRemoteMetaWrite) { - void onRemoteStaleFlag(doc, meta, reporter) - return - } - void recoverFromStaleState(doc, prov) + if (isRemoteMetaWrite && event.keysChanged.has('isStale') && meta.get('isStale') === true) { + void onRemoteStaleFlag(doc, meta, reporter) } } } /** - * A peer raised `isStale` mid-session. Three cases: we claimed it earlier - * and run the recovery; a keystroke inside the debounce window made us - * dirty after all, so we conflict like the dirty peers did at flag time; or - * we are clean and follow the claimant's rewrite (`onExternalUpdate` fires - * when it lands). + * A peer raised `isStale` mid-session. Either a keystroke inside the + * debounce window made us dirty after all, so we conflict like the dirty + * peers did at flag time, or we are clean and follow the rewrite + * (`onExternalUpdate` fires when it lands). */ async function onRemoteStaleFlag(doc: Y.Doc, meta: SessionMetaMap, reporter: ContentReporter) { - if (staleRecovery !== null) { - void recoverFromStaleState(doc, unref(provider)) - return - } await reporter.flush() if (doc.isDestroyed || unref(ydoc) !== doc) return if (meta.get('isStale') !== true) return @@ -1204,14 +1181,6 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { if (toValue(hasUnsavedChanges)) markConflicted() } - /** Whether the room's recovery claimant is still connected. */ - function isClaimantPresent(doc: Y.Doc, meta: SessionMetaMap): boolean { - const claimant = meta.get('recoveryClientId') - if (typeof claimant !== 'number') return false - if (claimant === doc.clientID) return true - return unref(awareness)?.getStates().has(claimant) ?? false - } - /** See {@link YjsSession.isRoomWrite}. */ function isRoomWrite(etag: string) { const doc = unref(ydoc) @@ -1228,17 +1197,17 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { // state. The caller's etag mirror then finds the stamp in place and // does not claim one either. if (meta.get('etag') !== etag) { - doc.transact(() => meta.set('etag', etag), ETAG_ADOPT_ORIGIN) + doc.transact(() => meta.set('etag', etag)) } } onEtagChange(etag) } /** - * The rewrite suppressed the content reporter and its commit carries an own - * origin, so nothing else tells the caller that its content, server content - * and etag all moved. Reports the serialized form: the next debounced report - * produces the same string, so the file stays clean. + * The rewrite suppressed the content reporter and the meta observer ignores + * our own writes, so nothing else tells the caller that its content, server + * content and etag all moved. Reports the serialized form: the next + * debounced report produces the same string, so the file stays clean. */ async function reportRecovered(doc: Y.Doc, etag: string) { const value = await serializeDoc(doc) @@ -1279,26 +1248,18 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { onConflict() return 'conflicted' } - // Flag without claiming: dirty peers conflict on the flag, clean peers - // recover from their own fetch, late joiners recover from theirs. - flagStale(doc, meta, { nativeEtag: etag, claim: false, origin: EXTERNAL_UPDATE_ORIGIN }) + // Flag only, our doc is nothing safe to re-seed with: dirty peers + // conflict on the flag, clean peers recover from their own fetch, late + // joiners from theirs. + flagStale(doc, meta, etag) markConflicted() return 'conflicted' } - // A live claimant is already on it; without one, this is the same as a - // joiner finding the flag up. Concurrent claimants converge through - // `recoveryClientId` in `runStaleRecovery`, the losers just skip. - if (meta.get('isStale') === true && isClaimantPresent(doc, meta)) return 'skipped' - flagStale(doc, meta, { - nativeEtag: etag, - claim: true, - payload: { content, etag }, - origin: EXTERNAL_UPDATE_ORIGIN - }) - - const recovered = await recoverFromStaleState(doc, prov) - if (!recovered) return 'skipped' + // Every clean writer that got the event gets here, the server lets one + // of them through. + const recovered = await recoverFromStaleState(doc, prov, { content, etag }) + if (!recovered) return unref(isConflicted) ? 'conflicted' : 'skipped' await reportRecovered(doc, etag) onExternalUpdate?.() return 'recovered' @@ -1322,7 +1283,6 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { hasLocalOnlyContent = false // `undoResyncWipe` leaves this up for good when it cannot restore. isRewritingDoc = false - staleRecovery = null lastReportedStateVector = null lastReportedContent = null pendingSaveStateVector = null @@ -1369,7 +1329,7 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { reporter.cancel() if (activeReporter === reporter) activeReporter = null clearConnectTimer() - abandonSeedGrant() + abandonGrants() meta.map.unobserve(metaObserver) doc.off('update', reporter.onDocUpdate) if (prov) { @@ -1407,7 +1367,7 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { pendingSaveStateVector ?? lastReportedStateVector ?? Y.encodeStateVector(doc) ) meta.set('lastSavedAt', Date.now()) - }, LOCAL_SAVE_ORIGIN) + }) // Consumed. An etag change the caller did not announce falls back to // the last reported state again. pendingSaveStateVector = null diff --git a/packages/web-pkg/tests/unit/composables/yjs/useYjsSession.spec.ts b/packages/web-pkg/tests/unit/composables/yjs/useYjsSession.spec.ts index 3caebec0b05..dc94cfc2db1 100644 --- a/packages/web-pkg/tests/unit/composables/yjs/useYjsSession.spec.ts +++ b/packages/web-pkg/tests/unit/composables/yjs/useYjsSession.spec.ts @@ -49,13 +49,16 @@ interface MockProvider { const { providerInstances, relay } = vi.hoisted(() => { return { providerInstances: [] as MockProvider[], - // `seedHolders` mirrors the Yjs server's seed registry (see - // `services/yjs/src/lib/seedGrant.ts`): one grant per room. + // `grantHolders` mirrors the Yjs server's grant registry (see + // `services/yjs/src/lib/grants.ts`): one holder per room and key, stored + // under `grantSlot`. `seedGrantOnly` plays a server from before the + // generic grants, which ignores every other request. relay: { enabled: false, docDelayMs: 0, awarenessDelayMs: 0, - seedHolders: new Map() + grantHolders: new Map(), + seedGrantOnly: false } } }) @@ -90,26 +93,28 @@ vi.mock('@hocuspocus/provider', async () => { for (const peer of this.peers()) peer.awareness.states.set(clientId, state) }, relay.awarenessDelayMs) }) - // The client asks the server who may seed. The answer comes from the + // The client asks the server for a grant. The answer comes from the // server, not from a peer, so it does not depend on `relay.enabled`. It // travels the same socket as a doc update, so it carries the same delay; a // delay of 0 answers synchronously, which keeps every latency-agnostic // test on a single `flushPromises`. sendStateless = vi.fn((payload: string) => { + const prefix = '_oc_grant_request:' + const isSeed = payload === '_oc_seed_request' + if (!isSeed && (relay.seedGrantOnly || !payload.startsWith(prefix))) return + const key = isSeed ? 'seed' : payload.slice(prefix.length) + const slot = `${this.name}|${key}` const socketId = String(this.document.clientID) - let answer: (() => void) | null = null - if (payload === '_oc_seed_request') { - answer = () => { - if (this.stopped) return - const holder = relay.seedHolders.get(this.name) - const granted = holder === undefined || holder === socketId - if (granted) relay.seedHolders.set(this.name, socketId) - this._opts.onStateless?.({ - payload: granted ? '_oc_seed_granted' : '_oc_seed_denied' - }) - } + const answer = () => { + if (this.stopped) return + const holder = relay.grantHolders.get(slot) + const granted = holder === undefined || holder === socketId + if (granted) relay.grantHolders.set(slot, socketId) + const answerPayload = isSeed + ? `_oc_seed_${granted ? 'granted' : 'denied'}` + : `${granted ? '_oc_grant_granted:' : '_oc_grant_denied:'}${key}` + this._opts.onStateless?.({ payload: answerPayload }) } - if (!answer) return if (relay.docDelayMs === 0) { answer() return @@ -279,6 +284,16 @@ function setupSession({ } } +/** Key of `relay.grantHolders`. */ +function grantSlot(provider: MockProvider, key: string) { + return `${provider.name}|${key}` +} + +/** Someone else holds the grant, as far as the mock server is concerned. */ +function grantElsewhere(provider: MockProvider, key: string) { + relay.grantHolders.set(grantSlot(provider, key), 'a-peer') +} + function silenceConsoleError() { vi.spyOn(console, 'error').mockImplementation(() => undefined) } @@ -288,7 +303,8 @@ beforeEach(() => { relay.enabled = false relay.docDelayMs = 0 relay.awarenessDelayMs = 0 - relay.seedHolders.clear() + relay.grantHolders.clear() + relay.seedGrantOnly = false }) afterEach(() => { @@ -576,6 +592,21 @@ describe('useYjsSession — remote mode (yjsServerUrl set)', () => { expect(s.ydoc!.getText(SHARED_TEXT_KEY).toString()).toBe('the real file body') }) + it('seeds against a server from before the generic grants', async () => { + relay.seedGrantOnly = true + const s = setupSession({ + yjsServerUrl: 'wss://example.test/yjs', + currentContent: 'the real file body' + }) + await flushPromises() + + providerInstances[0].triggerSynced() + await flushPromises() + + expect(s.ydoc!.getText(SHARED_TEXT_KEY).toString()).toBe('the real file body') + expect(unref(s.session.isReady)).toBe(true) + }) + // The peer that holds the grant is seeding, and its body is on the way. it('does not seed when the server denies the grant', async () => { const s = setupSession({ @@ -583,7 +614,7 @@ describe('useYjsSession — remote mode (yjsServerUrl set)', () => { currentContent: 'the real file body' }) await flushPromises() - relay.seedHolders.set(providerInstances[0].name, 'a-peer') + grantElsewhere(providerInstances[0], 'seed') providerInstances[0].triggerSynced() await flushPromises() @@ -766,9 +797,8 @@ describe('useYjsSession — remote mode (yjsServerUrl set)', () => { }) // A token expiring mid-session leaves a live, populated document. Re-entering - // the hydration path there would re-run the etag-drift check and could plant - // `isStale` plus a `recoveryClientId` claim this now read-only client will - // never act on - leaving the room flagged stale with a dead claim. + // the hydration path there would re-run the etag-drift check and could start + // a recovery against a room this client has just left. it('does not re-run hydration when the token expires mid-session', async () => { silenceConsoleError() vi.useFakeTimers({ shouldAdvanceTime: true }) @@ -789,7 +819,7 @@ describe('useYjsSession — remote mode (yjsServerUrl set)', () => { const meta = s.ydoc!.getMap('_oc_meta') expect(meta.get('isStale')).toBeUndefined() - expect(meta.get('recoveryClientId')).toBeUndefined() + expect(meta.get('etag')).toBe('etag-old') }) }) @@ -922,9 +952,9 @@ describe('useYjsSession — read-only clients', () => { const privateDoc = s.ydoc expect(privateDoc!.getText(SHARED_TEXT_KEY).toString()).toBe('read me') - // A remote transaction has no string origin, which is what tells the - // observer this came from a peer. - privateDoc!.transact(() => privateDoc!.getMap('_oc_meta').set('hydrated', true)) + const peer = new Y.Doc() + peer.getMap('_oc_meta').set('hydrated', true) + Y.applyUpdate(privateDoc!, Y.encodeStateAsUpdate(peer)) await flushPromises() expect(s.ydoc).not.toBe(privateDoc) @@ -1038,7 +1068,7 @@ describe('useYjsSession — stale-state recovery', () => { /** * Brings a session up in remote mode against a room whose `_oc_meta.etag` * predates the file on disk, which is what makes the joining client detect - * drift and claim the recovery. + * drift and ask for the recovery. */ async function syncIntoStaleRoom({ currentContent = 'fresh body', @@ -1071,15 +1101,121 @@ describe('useYjsSession — stale-state recovery', () => { beforeEach(() => { vi.useFakeTimers({ shouldAdvanceTime: true }) + // The grant answer takes a moment, so tests can look inside the wait. + relay.docDelayMs = 100 }) - it('flags the room stale and claims the recovery when the etag drifted', async () => { + it('asks the server to recover from the etag it fetched when the etag drifted', async () => { const s = await syncIntoStaleRoom() + + expect(providerInstances[0].sendStateless).toHaveBeenCalledWith( + '_oc_grant_request:recover:etag-new' + ) + // Nothing is written before the answer, and the editor waits for it. + expect(s.ydoc!.getMap(META_KEY).get('isStale')).toBeUndefined() + expect(s.ydoc!.getText(SHARED_TEXT_KEY).toString()).toBe('stale room content') + expect(unref(s.session.isReady)).toBe(false) + + await vi.advanceTimersByTimeAsync(200) + + expect(s.ydoc!.getText(SHARED_TEXT_KEY).toString()).toBe('fresh body') + expect(unref(s.session.isReady)).toBe(true) + }) + + // #3429: the client holding the grant is rewriting the room already, a + // second rewrite would put the body in twice. + it('leaves the room alone while another client holds the recovery grant', async () => { + const s = setupSession({ + yjsServerUrl, + currentContent: 'fresh body', + resource: makeResource({ etag: 'etag-new' }) + }) + await flushPromises() + grantElsewhere(providerInstances[0], 'recover:etag-new') + const doc = s.ydoc! + doc.getText(SHARED_TEXT_KEY).insert(0, 'stale room content') + doc.getMap(META_KEY).set('etag', 'etag-old') + providerInstances[0].triggerSynced() + vi.advanceTimersByTime(200) + await flushPromises() + + expect(doc.getText(SHARED_TEXT_KEY).toString()).toBe('stale room content') + expect(doc.getMap(META_KEY).get('isStale')).toBeUndefined() + }) + + // The room still shows the old content. With the fresh etag, the first + // autosave would write it over the external change without a conflict. + it('takes the room etag when another client holds the recovery grant', async () => { + const s = setupSession({ + yjsServerUrl, + currentContent: 'fresh body', + resource: makeResource({ etag: 'etag-new' }) + }) + await flushPromises() + grantElsewhere(providerInstances[0], 'recover:etag-new') + const doc = s.ydoc! + doc.getText(SHARED_TEXT_KEY).insert(0, 'stale room content') + doc.getMap(META_KEY).set('etag', 'etag-old') + providerInstances[0].triggerSynced() + await vi.advanceTimersByTimeAsync(200) + + expect(s.onEtagChange).toHaveBeenCalledWith('etag-old') + expect(unref(s.session.isReady)).toBe(true) + }) + + it('recovers a room flagged for an older body than the one it holds', async () => { + const s = setupSession({ + yjsServerUrl, + currentContent: 'even newer body', + resource: makeResource({ etag: 'etag-newest' }) + }) + await flushPromises() + const doc = s.ydoc! + const meta = doc.getMap(META_KEY) + doc.getText(SHARED_TEXT_KEY).insert(0, 'stale room content') + doc.transact(() => { + meta.set('etag', 'etag-old') + meta.set('nativeEtag', 'etag-new') + meta.set('isStale', true) + }) + providerInstances[0].triggerSynced() + await vi.advanceTimersByTimeAsync(200) + + expect(doc.getText(SHARED_TEXT_KEY).toString()).toBe('even newer body') + expect(meta.get('etag')).toBe('etag-newest') + expect(meta.get('isStale')).toBeUndefined() + }) + + // A server from before recovery grants ignores the request. The clients + // then elect one among themselves, as before the grants. + it('falls back to an election when the server never answers the recovery request', async () => { + relay.seedGrantOnly = true + const s = await syncIntoStaleRoom() + + await vi.advanceTimersByTimeAsync(4_900) + expect(s.ydoc!.getText(SHARED_TEXT_KEY).toString()).toBe('stale room content') + expect(unref(s.session.isReady)).toBe(false) + + await vi.advanceTimersByTimeAsync(300) + const meta = s.ydoc!.getMap(META_KEY) + expect(s.ydoc!.getText(SHARED_TEXT_KEY).toString()).toBe('fresh body') + expect(meta.get('etag')).toBe('etag-new') + expect(meta.get('recoveryClientId')).toBeUndefined() + expect(unref(s.session.isReady)).toBe(true) + }) - expect(meta.get('isStale')).toBe(true) - expect(meta.get('nativeEtag')).toBe('etag-new') - expect(meta.get('recoveryClientId')).toBe(s.ydoc!.clientID) + it('takes the room etag when it loses the election', async () => { + relay.seedGrantOnly = true + const s = await syncIntoStaleRoom() + await vi.advanceTimersByTimeAsync(5_000) + + // A peer that noticed the same write claims after us, so it wins. + s.ydoc!.getMap(META_KEY).set('recoveryClientId', s.ydoc!.clientID + 1) + await vi.advanceTimersByTimeAsync(200) + + expect(s.ydoc!.getText(SHARED_TEXT_KEY).toString()).toBe('stale room content') + expect(s.onEtagChange).toHaveBeenCalledWith('etag-old') }) // Regression: recovery used to re-seed from whatever the elected peer held @@ -1101,7 +1237,6 @@ describe('useYjsSession — stale-state recovery', () => { expect(s.ydoc!.getText(SHARED_TEXT_KEY).toString()).toBe('fresh body') expect(meta.get('isStale')).toBeUndefined() expect(meta.get('nativeEtag')).toBeUndefined() - expect(meta.get('recoveryClientId')).toBeUndefined() expect(meta.get('etag')).toBe('etag-new') }) @@ -1137,11 +1272,15 @@ describe('useYjsSession — stale-state recovery', () => { // A peer elsewhere in the room notices the drift and flags it. const meta = s.ydoc!.getMap(META_KEY) - s.ydoc!.transact(() => { - meta.set('nativeEtag', 'etag-new') - meta.set('recoveryClientId', s.ydoc!.clientID + 1) - meta.set('isStale', true) + const peer = new Y.Doc() + Y.applyUpdate(peer, Y.encodeStateAsUpdate(s.ydoc!)) + peer.transact(() => { + const peerMeta = peer.getMap(META_KEY) + peerMeta.set('nativeEtag', 'etag-new') + peerMeta.set('recoveryEpoch', 1) + peerMeta.set('isStale', true) }) + Y.applyUpdate(s.ydoc!, Y.encodeStateAsUpdate(peer, Y.encodeStateVector(s.ydoc!))) vi.advanceTimersByTime(200) await flushPromises() @@ -1167,6 +1306,9 @@ describe('useYjsSession — stale-state recovery', () => { await flushPromises() expect(doc.getText(SHARED_TEXT_KEY).toString()).toBe('stale room content') + // Its writes never reach the room, so it neither flags nor asks. + expect(doc.getMap(META_KEY).get('isStale')).toBeUndefined() + expect(providerInstances[0].sendStateless).not.toHaveBeenCalled() }) // The reset lands before the hydrate, so a throw in between leaves every peer @@ -1193,11 +1335,10 @@ describe('useYjsSession — stale-state recovery', () => { expect(unref(s.session.error)).toBeTruthy() }) - // The claim is deliberately stealable: whoever was elected may have navigated - // away mid-recovery, and then nobody would finish. Safe because `hydrate` is - // synchronous by contract - reset, hydrate and commit run in one go, so a - // claim cannot move underneath a client that is already working. - it('takes the claim over from a peer that has left', async () => { + // Whoever raised the flag may have navigated away mid-recovery, and then + // nobody would finish. The server releases a leaver's grant, so a joiner + // holding the body asks for it again. + it('recovers a room it finds flagged for the body it holds', async () => { const s = setupSession({ yjsServerUrl, currentContent: 'fresh body', @@ -1207,10 +1348,9 @@ describe('useYjsSession — stale-state recovery', () => { const doc = s.ydoc! const meta = doc.getMap(META_KEY) - // Claimed by a client that is no longer in awareness: nobody would finish. + doc.getText(SHARED_TEXT_KEY).insert(0, 'stale room content') doc.transact(() => { meta.set('nativeEtag', 'etag-new') - meta.set('recoveryClientId', doc.clientID + 99) meta.set('isStale', true) }) @@ -1246,20 +1386,6 @@ describe('useYjsSession — stale-state recovery', () => { expect(meta.get('isStale')).toBeUndefined() }) - // A read-only client can spot the drift but cannot act on it. It still flags - // the room, and the `nativeEtag` it leaves behind is what lets a writer - // holding that same body take the claim. - it('flags without claiming on a read-only client', async () => { - const s = await syncIntoStaleRoom({ isReadOnly: true }) - vi.advanceTimersByTime(200) - await flushPromises() - - const meta = s.ydoc!.getMap(META_KEY) - expect(meta.get('isStale')).toBe(true) - expect(meta.get('nativeEtag')).toBe('etag-new') - expect(meta.get('recoveryClientId')).toBeUndefined() - }) - // A peer that holds unsaved work must not follow the reset: it arrives as // plain CRDT updates and would wipe that work with no warning. Leaving the // room keeps the local doc, and the caller surfaces its conflict message. @@ -1276,13 +1402,12 @@ describe('useYjsSession — stale-state recovery', () => { await flushPromises() const doc = s.ydoc! - // A peer joined, found the file changed on disk and claimed the recovery. + // A peer joined, found the file changed on disk and won the recovery. const peer = new Y.Doc() Y.applyUpdate(peer, Y.encodeStateAsUpdate(doc)) peer.transact(() => { const peerMeta = peer.getMap(META_KEY) peerMeta.set('nativeEtag', 'etag-external') - peerMeta.set('recoveryClientId', peer.clientID) peerMeta.set('recoveryEpoch', 1) peerMeta.set('isStale', true) }) @@ -1319,7 +1444,6 @@ describe('useYjsSession — stale-state recovery', () => { peer.transact(() => { const peerMeta = peer.getMap(META_KEY) peerMeta.set('nativeEtag', 'etag-external') - peerMeta.set('recoveryClientId', peer.clientID) peerMeta.set('recoveryEpoch', 1) peerMeta.set('isStale', true) }) @@ -1880,7 +2004,7 @@ describe('useYjsSession — initial sync is not a peer save', () => { // The point of the gate. With the write-back wired up as AppWrapper wires it, // the room's etag used to land in `resource.etag` before the drift check ran, // so the check compared the room etag against itself and a late joiner could - // never claim the recovery. + // never recover the room. it('still detects drift on a late joiner whose caller mirrors etags back', async () => { const s = setupSession({ yjsServerUrl, @@ -1894,9 +2018,8 @@ describe('useYjsSession — initial sync is not a peer save', () => { providerInstances[0].triggerSynced() await flushPromises() - expect(s.ydoc!.getMap(META_KEY).get('isStale')).toBe(true) - expect(s.ydoc!.getMap(META_KEY).get('nativeEtag')).toBe('etag-new') - expect(s.ydoc!.getMap(META_KEY).get('recoveryClientId')).toBe(s.ydoc!.clientID) + expect(s.ydoc!.getMap(META_KEY).get('etag')).toBe('etag-new') + expect(s.ydoc!.getText(SHARED_TEXT_KEY).toString()).toBe('our body') }) // Recovery rewrites `_oc_meta.etag` itself, after the gate has opened. That @@ -2063,7 +2186,8 @@ describe('useYjsSession — simultaneous hydration', () => { // The mock server's registry knows who was granted, the doc only that someone was. function seeder(a: ReturnType, b: ReturnType) { - return relay.seedHolders.get(providerInstances[0].name) === String(a.ydoc!.clientID) ? a : b + const holder = relay.grantHolders.get(grantSlot(providerInstances[0], 'seed')) + return holder === String(a.ydoc!.clientID) ? a : b } // A doubled body is not only wrong on screen: reported to the caller it is @@ -2109,7 +2233,7 @@ describe('useYjsSession — unsolicited seed grant', () => { it('seeds an empty room on a grant that arrives on its own', async () => { const s = setupSession({ yjsServerUrl: URL, currentContent: 'the file body' }) await flushPromises() - relay.seedHolders.set(providerInstances[0].name, 'a-peer') + grantElsewhere(providerInstances[0], 'seed') providerInstances[0].triggerSynced() await flushPromises() @@ -2143,7 +2267,7 @@ describe('useYjsSession — unsolicited seed grant', () => { vi.useFakeTimers() const s = setupSession({ yjsServerUrl: URL, currentContent: 'the file body' }) await flushPromises() - relay.seedHolders.set(providerInstances[0].name, 'a-peer') + grantElsewhere(providerInstances[0], 'seed') providerInstances[0].triggerSynced() await flushPromises() @@ -2171,7 +2295,7 @@ describe('useYjsSession — unsolicited seed grant', () => { adapter: throwingAdapter }) await flushPromises() - relay.seedHolders.set(providerInstances[0].name, 'a-peer') + grantElsewhere(providerInstances[0], 'seed') providerInstances[0].triggerSynced() await flushPromises() @@ -2217,7 +2341,7 @@ describe('useYjsSession — external updates', () => { return s } - /** Runs `applyExternalUpdate` through the 150 ms election wait. */ + /** Runs `applyExternalUpdate` through the grant round trip. */ async function apply(s: ReturnType, update = EXTERNAL) { const result = s.session.applyExternalUpdate(update) await vi.advanceTimersByTimeAsync(500) @@ -2274,7 +2398,6 @@ describe('useYjsSession — external updates', () => { expect(result).toBe('recovered') expect(text(s)).toBe('external body') expect(meta.get('etag')).toBe('etag-external') - expect(meta.get('recoveryClientId')).toBeUndefined() // The recovery commit is the only save stamp. The caller mirrors the // reported etag back into `resource.etag`, and that must not stamp a // second, fake local save on top. @@ -2292,8 +2415,10 @@ describe('useYjsSession — external updates', () => { expect(meta.get('isStale')).toBe(true) expect(meta.get('nativeEtag')).toBe('etag-external') expect(meta.get('recoveryEpoch')).toBe(1) - // Not claimed: this client has nothing safe to re-seed with. - expect(meta.get('recoveryClientId')).toBeUndefined() + // No recovery: this client has nothing safe to re-seed with. + expect(providerInstances[0].sendStateless).not.toHaveBeenCalledWith( + '_oc_grant_request:recover:etag-external' + ) expect(unref(s.session.isConflicted)).toBe(true) expect(providerInstances[0].disconnect).toHaveBeenCalled() expect(providerInstances[0].detach).toHaveBeenCalled() @@ -2342,34 +2467,40 @@ describe('useYjsSession — external updates', () => { expect(text(s)).toBe('seed') }) - it('skips when a connected peer already claimed the recovery', async () => { + it('skips when another client holds the recovery grant', async () => { const s = await remoteSession() - const meta = s.ydoc!.getMap(META_KEY) - const claimant = s.ydoc!.clientID + 99 - providerInstances[0].awareness.states.set(claimant, {}) - s.ydoc!.transact(() => { - meta.set('nativeEtag', 'etag-external') - meta.set('recoveryClientId', claimant) - meta.set('recoveryEpoch', 1) - meta.set('isStale', true) - }, 'test-peer') + grantElsewhere(providerInstances[0], 'recover:etag-external') const result = await apply(s) expect(result).toBe('skipped') - expect(meta.get('recoveryClientId')).toBe(claimant) expect(text(s)).toBe('seed') + expect(s.ydoc!.getMap(META_KEY).get('isStale')).toBeUndefined() + }) + + it('recovers through the election when the server never answers', async () => { + relay.seedGrantOnly = true + const s = await remoteSession() + + const result = s.session.applyExternalUpdate(EXTERNAL) + await vi.advanceTimersByTimeAsync(5_200) + + expect(await result).toBe('recovered') + expect(text(s)).toBe('external body') }) - it('takes over the recovery when the claimant has left', async () => { + it('recovers a room that is still flagged for an older write', async () => { const s = await remoteSession() const meta = s.ydoc!.getMap(META_KEY) - s.ydoc!.transact(() => { - meta.set('nativeEtag', 'etag-older') - meta.set('recoveryClientId', s.ydoc!.clientID + 99) - meta.set('recoveryEpoch', 1) - meta.set('isStale', true) - }, 'test-peer') + const peer = new Y.Doc() + Y.applyUpdate(peer, Y.encodeStateAsUpdate(s.ydoc!)) + peer.transact(() => { + const peerMeta = peer.getMap(META_KEY) + peerMeta.set('nativeEtag', 'etag-older') + peerMeta.set('recoveryEpoch', 1) + peerMeta.set('isStale', true) + }) + Y.applyUpdate(s.ydoc!, Y.encodeStateAsUpdate(peer, Y.encodeStateVector(s.ydoc!))) const result = await apply(s) @@ -2379,6 +2510,38 @@ describe('useYjsSession — external updates', () => { expect(meta.get('etag')).toBe('etag-external') }) + // No answer can arrive once the provider is stopped, so the wait must not hang. + it('gives up a recovery still waiting for the grant when the session leaves the room', async () => { + const s = await remoteSession() + relay.docDelayMs = 100 + + const result = s.session.applyExternalUpdate(EXTERNAL) + await flushPromises() + s.session.markConflicted() + await vi.advanceTimersByTimeAsync(500) + + expect(await result).toBe('conflicted') + expect(text(s)).toBe('seed') + }) + + it('conflicts instead of rewriting over an edit typed while waiting for the grant', async () => { + const s = await remoteSession() + relay.docDelayMs = 100 + s.onContentChange.mockImplementation(() => { + s.hasUnsavedChangesRef.value = true + }) + + const result = s.session.applyExternalUpdate(EXTERNAL) + await flushPromises() + s.ydoc!.getText(SHARED_TEXT_KEY).insert(0, 'typed ') + await vi.advanceTimersByTimeAsync(500) + + expect(await result).toBe('conflicted') + expect(text(s)).toBe('typed seed') + expect(unref(s.session.isConflicted)).toBe(true) + expect(s.ydoc!.getMap(META_KEY).get('isStale')).toBe(true) + }) + it('skips for a read-only, conflicted or locked session', async () => { const s = await remoteSession({ isReadOnly: true }) @@ -2387,10 +2550,10 @@ describe('useYjsSession — external updates', () => { }) describe('with peers', () => { - async function twoPeers({ bDirty = false } = {}) { + async function twoPeers({ bDirty = false, relayMs = 10 } = {}) { relay.enabled = true - relay.docDelayMs = 10 - relay.awarenessDelayMs = 10 + relay.docDelayMs = relayMs + relay.awarenessDelayMs = relayMs const a = setupSession({ yjsServerUrl: URL, currentContent: 'seed' }) const b = setupSession({ yjsServerUrl: URL, @@ -2400,7 +2563,7 @@ describe('useYjsSession — external updates', () => { await flushPromises() providerInstances[0].triggerSynced() providerInstances[1].triggerSynced() - await vi.advanceTimersByTimeAsync(1_000) + await vi.advanceTimersByTimeAsync(4 * relayMs + 1_000) await flushPromises() expect(text(a)).toBe('seed') expect(text(b)).toBe('seed') @@ -2409,40 +2572,67 @@ describe('useYjsSession — external updates', () => { return { a, b } } - it('lets exactly one of two clean peers rewrite, and the other follows', async () => { + // #3429: a relay slower than the old 150 ms election wait let both rewrite, + // and the room held the body twice. + it.each([10, 500])( + 'lets exactly one of two clean peers rewrite with a %ims relay', + async (relayMs) => { + const { a, b } = await twoPeers({ relayMs }) + + const results = Promise.all([ + a.session.applyExternalUpdate(EXTERNAL), + b.session.applyExternalUpdate(EXTERNAL) + ]) + await vi.advanceTimersByTimeAsync(4 * relayMs + 1_000) + await flushPromises() + + expect((await results).sort()).toEqual(['recovered', 'skipped']) + expect(text(a)).toBe('external body') + expect(text(b)).toBe('external body') + for (const s of [a, b]) { + expect(s.onEtagChange).toHaveBeenCalledWith('etag-external') + expect(s.onServerContentChange).toHaveBeenCalledWith('external body') + expect(s.onConflict).not.toHaveBeenCalled() + expect(s.onExternalUpdate).toHaveBeenCalledTimes(1) + // The client that was refused wrote nothing, so no flag outlives the rewrite. + expect(s.ydoc!.getMap(META_KEY).get('isStale')).toBeUndefined() + } + } + ) + + it('lets exactly one of two clean peers rewrite when the server never answers', async () => { const { a, b } = await twoPeers() + relay.seedGrantOnly = true - const [resultA, resultB] = await Promise.all([ + const results = Promise.all([ a.session.applyExternalUpdate(EXTERNAL), b.session.applyExternalUpdate(EXTERNAL) ]) - await vi.advanceTimersByTimeAsync(1_000) + await vi.advanceTimersByTimeAsync(6_000) await flushPromises() - expect([resultA, resultB].sort()).toEqual(['recovered', 'skipped']) + expect((await results).sort()).toEqual(['recovered', 'skipped']) expect(text(a)).toBe('external body') expect(text(b)).toBe('external body') - for (const s of [a, b]) { - expect(s.onEtagChange).toHaveBeenCalledWith('etag-external') - expect(s.onServerContentChange).toHaveBeenCalledWith('external body') - expect(s.onConflict).not.toHaveBeenCalled() - expect(s.onExternalUpdate).toHaveBeenCalledTimes(1) - } }) // The toast says the editor shows the new version, so a follower must not - // hear it on the flag: the claimant may still lose the election or fail. + // hear it on the flag: the rewrite may still fail. it('tells a follower about the update only once the rewrite landed', async () => { const { a, b } = await twoPeers() + const callsAtFlag: number[] = [] + const bMeta = b.ydoc!.getMap(META_KEY) + bMeta.observe((event) => { + if (event.keysChanged.has('isStale') && bMeta.get('isStale') === true) { + callsAtFlag.push(b.onExternalUpdate.mock.calls.length) + } + }) const result = a.session.applyExternalUpdate(EXTERNAL) - // Flag is out, the 150 ms election wait is not over. - await vi.advanceTimersByTimeAsync(50) - expect(b.ydoc!.getMap(META_KEY).get('isStale')).toBe(true) - expect(b.onExternalUpdate).not.toHaveBeenCalled() - await vi.advanceTimersByTimeAsync(1_000) + expect(await result).toBe('recovered') + expect(callsAtFlag).toEqual([0]) expect(text(b)).toBe('external body') expect(b.onExternalUpdate).toHaveBeenCalledTimes(1) }) @@ -2516,11 +2706,12 @@ describe('useYjsSession — external updates', () => { // save must cover the followers: its state vector was taken after the rewrite. it('stamps a save after recovery with a vector that covers a follower', async () => { const { a, b } = await twoPeers() - await Promise.all([ + const results = Promise.all([ a.session.applyExternalUpdate(EXTERNAL), b.session.applyExternalUpdate(EXTERNAL) ]) await vi.advanceTimersByTimeAsync(1_000) + await results b.onServerContentChange.mockClear() a.resourceRef.value = { ...unref(a.resourceRef), etag: 'etag-saved' } as Resource diff --git a/services/yjs/README.md b/services/yjs/README.md index 88277a28a22..f37d8bf92c2 100644 --- a/services/yjs/README.md +++ b/services/yjs/README.md @@ -17,17 +17,19 @@ Every connection is authenticated and authorized against OpenCloud: - on connect, each client receives its own stamped identity over a stateless message, because Yjs never echoes a client's own awareness back to it -## Seeding an empty room - -The first client in a room writes the file body into the Y.Doc. Exactly one client may do that, or -the room ends up holding the body twice. The yjs server decides which, because it is the only party -that sees every connection to a room at once. A client asks over a stateless message, and gets a -grant or a refusal. Read-only connections are always refused. - -Grants live in process memory and last as long as the holder's connection. If the holder leaves -before it seeds, the grant passes to another writer in the room. A grant for a room that already has -content is harmless: it is permission, not an instruction, and the client checks its own document -first. +## Seeding and recovering a room + +Two jobs may only be done by exactly one client per room, or the room ends up holding the file body +twice: seeding an empty room (grant `seed`), and rewriting a stale room after the file changed +outside it (grant `recover:`). The yjs server decides who, because it is the only party that +sees every connection to a room at once. A client asks over a stateless message, and gets a grant +or a refusal. The first writer to ask for a key gets it, read-only connections are always refused. + +Grants live in process memory and last as long as the holder's connection. If the holder of `seed` +leaves before it seeds, the grant passes to another writer in the room. A recovery grant is not +passed on, because only the requester holds the body to recover from; the next client that asks +gets it. A grant for a job that is already done is harmless: it is permission, not an instruction, +and the client checks its own document first. ## Configuration diff --git a/services/yjs/src/lib/grants.ts b/services/yjs/src/lib/grants.ts new file mode 100644 index 00000000000..fa802bce943 --- /dev/null +++ b/services/yjs/src/lib/grants.ts @@ -0,0 +1,129 @@ +/** + * Stateless protocol for room-wide jobs that exactly one connection may do: + * seeding an empty room (`seed`) and rewriting a stale one from the file body + * behind an etag (`recover:`). A payload is a prefix plus the key. + * + * The payloads have to match the `GRANT_*` and `SEED_*` constants in the + * client's `useYjsSession.ts`, keep the two in sync. + */ +export const GrantMessage = { + Request: '_oc_grant_request:', + Granted: '_oc_grant_granted:', + Denied: '_oc_grant_denied:' +} as const + +export const SEED_GRANT = 'seed' +const RECOVER_PREFIX = 'recover:' + +/** + * @deprecated The seed grant's payloads from before `GrantMessage`, drop once + * no client sends `SeedMessage.Request` anymore. Until then an unsolicited + * seed grant uses them too, because every client version understands them. + */ +export const SeedMessage = { + Request: '_oc_seed_request', + Granted: '_oc_seed_granted', + Denied: '_oc_seed_denied' +} as const + +/** Generous for `recover:`, but bounds what a client can make us store. */ +const MAX_GRANT_KEY_LENGTH = 256 + +export interface GrantRequest { + key: string + /** Asked with `SeedMessage`, so it has to be answered with it. */ + legacy: boolean +} + +/** The grant a request asks for, or null for any other payload. */ +export function parseGrantRequest(payload: string): GrantRequest | null { + if (payload === SeedMessage.Request) { + return { key: SEED_GRANT, legacy: true } + } + if (!payload.startsWith(GrantMessage.Request)) { + return null + } + const key = payload.slice(GrantMessage.Request.length) + if (key !== SEED_GRANT && !key.startsWith(RECOVER_PREFIX)) { + return null + } + if (key.length > MAX_GRANT_KEY_LENGTH) { + return null + } + return { key, legacy: false } +} + +/** The payload that answers a grant request for `key`. */ +export function grantAnswer({ key, legacy }: GrantRequest, granted: boolean): string { + if (legacy) { + return granted ? SeedMessage.Granted : SeedMessage.Denied + } + return (granted ? GrantMessage.Granted : GrantMessage.Denied) + key +} + +/** + * Which connection holds which grant in each room. In memory on purpose: the + * question is only ever about connections that are live right now, so the + * answer has to die with the room. + * + * A grant lasts as long as the holder's socket. A writer that is granted but + * never does the job blocks it until it disconnects. + */ +export function createGrantRegistry() { + const rooms = new Map>() + + /** Answer a grant request. Read-only connections are refused, no write permissions. */ + function request(documentName: string, key: string, socketId: string, readOnly: boolean) { + if (readOnly) { + return false + } + const holders = rooms.get(documentName) + const holder = holders?.get(key) + if (holder !== undefined && holder !== socketId) { + return false + } + // A newer recovery supersedes the older one, so a writer holds at most one. + if (holders && key.startsWith(RECOVER_PREFIX)) { + for (const [held, heldBy] of holders) { + if (heldBy === socketId && held.startsWith(RECOVER_PREFIX)) { + holders.delete(held) + } + } + } + grantTo(documentName, key, socketId) + return true + } + + /** Hand a grant to a connection that did not ask, see `release`. */ + function grantTo(documentName: string, key: string, socketId: string): void { + const holders = rooms.get(documentName) ?? new Map() + holders.set(key, socketId) + rooms.set(documentName, holders) + } + + /** + * Give up a leaving connection's grants and return their keys, so the + * caller can move on what must not die with the leaver. + */ + function release(documentName: string, socketId: string): string[] { + const holders = rooms.get(documentName) + if (!holders) { + return [] + } + const released = [...holders].filter(([, holder]) => holder === socketId).map(([key]) => key) + for (const key of released) { + holders.delete(key) + } + if (!holders.size) { + rooms.delete(documentName) + } + return released + } + + /** Forget a room that no longer exists. */ + function forget(documentName: string): void { + rooms.delete(documentName) + } + + return { request, grantTo, release, forget } +} diff --git a/services/yjs/src/lib/hooks.ts b/services/yjs/src/lib/hooks.ts index a6cd3ea0964..d0ca3cf71cb 100644 --- a/services/yjs/src/lib/hooks.ts +++ b/services/yjs/src/lib/hooks.ts @@ -14,7 +14,7 @@ import { probeFileAccess, validateTokenAgainstOpenCloud } from './graph.ts' -import { createSeedRegistry, SeedMessage } from './seedGrant.ts' +import { createGrantRegistry, grantAnswer, parseGrantRequest, SEED_GRANT } from './grants.ts' export const HEALTH_ENDPOINT_PATH = '/healthz/ready' @@ -107,19 +107,20 @@ export async function authenticate( } export function createHooks({ opencloudUrl, lifecycle }: HookOptions) { - const seedRegistry = createSeedRegistry() + const grants = createGrantRegistry() /** * Hand the seed grant to another writer in the room. Called when the holder - * leaves, which may have happened before it seeded. + * leaves, which may have happened before it seeded. The writer may never + * have asked, so it gets the payload every client version understands. */ function passSeedGrantOn(document: Document, documentName: string): void { const writer = document.getConnections().find((connection) => !connection.readOnly) if (!writer) { return } - seedRegistry.grantTo(documentName, writer.socketId) - writer.sendStateless(SeedMessage.Granted) + grants.grantTo(documentName, SEED_GRANT, writer.socketId) + writer.sendStateless(grantAnswer({ key: SEED_GRANT, legacy: true }, true)) } return { @@ -193,30 +194,38 @@ export function createHooks({ opencloudUrl, lifecycle }: HookOptions) { console.log( `[onDisconnect] document=${JSON.stringify(documentName)} remaining=${clientsCount}` ) - if (seedRegistry.release(documentName, socketId)) { + // Only the seed grant moves on. A recovery needs the fresh body only its + // requester holds, so the next client that asks gets it instead. + if (grants.release(documentName, socketId).includes(SEED_GRANT)) { passSeedGrantOn(document, documentName) } }, /** * Safety net only: the holder's `onDisconnect` has normally released the - * grant by the time the room unloads. + * grants by the time the room unloads. */ async afterUnloadDocument({ documentName }: afterUnloadDocumentPayload): Promise { - seedRegistry.forget(documentName) + grants.forget(documentName) }, /** - * Seeding arbitration, see `seedGrant.ts`. The answer is permission, not - * an instruction: the client still checks its own document first, so a - * grant for a room that already has content costs nothing. + * Seeding and recovery arbitration, see `grants.ts`. The answer is + * permission, not an instruction: the client still checks its own + * document first, so a grant for a job that is already done costs nothing. */ async onStateless({ connection, documentName, payload }: onStatelessPayload): Promise { - if (payload !== SeedMessage.Request) { + const grant = parseGrantRequest(payload) + if (grant === null) { return } - const granted = seedRegistry.request(documentName, connection.socketId, connection.readOnly) - connection.sendStateless(granted ? SeedMessage.Granted : SeedMessage.Denied) + const granted = grants.request( + documentName, + grant.key, + connection.socketId, + connection.readOnly + ) + connection.sendStateless(grantAnswer(grant, granted)) }, /** diff --git a/services/yjs/src/lib/seedGrant.ts b/services/yjs/src/lib/seedGrant.ts deleted file mode 100644 index 144094b0966..00000000000 --- a/services/yjs/src/lib/seedGrant.ts +++ /dev/null @@ -1,67 +0,0 @@ -/** - * Stateless protocol for seeding an empty room: exactly one connection per - * room is allowed to write the initial body. - * - * The payload strings have to match the `SEED_*` constants in the client's - * `useYjsSession.ts`, keep the two in sync. - */ -export const SeedMessage = { - Request: '_oc_seed_request', - Granted: '_oc_seed_granted', - Denied: '_oc_seed_denied' -} as const - -/** - * Which connection currently holds the right to seed each room. In memory on - * purpose: the question is only ever about connections that are live right - * now, so the answer has to die with the room. - * - * A grant lasts as long as the holder's socket. A writer that is granted but - * never seeds blocks the room until it disconnects. - */ -export function createSeedRegistry() { - const holders = new Map() - - /** Answer a seed request. Read-only connections are refused, no write permissions. */ - function request(documentName: string, socketId: string, readOnly: boolean): boolean { - if (readOnly) { - return false - } - const holder = holders.get(documentName) - if (holder !== undefined && holder !== socketId) { - return false - } - holders.set(documentName, socketId) - return true - } - - /** Hand the grant to a connection that did not ask, see `release`. */ - function grantTo(documentName: string, socketId: string): void { - holders.set(documentName, socketId) - } - - /** - * Give up a leaving connection's grant. True when the leaver held it, which - * means the grant has to move on: a client that took it and left without - * seeding would otherwise leave the room empty for everyone. - * - * Passing the grant to a room that already has content is harmless, because - * the client checks its own document before using a grant. - */ - function release(documentName: string, socketId: string): boolean { - if (holders.get(documentName) !== socketId) { - return false - } - holders.delete(documentName) - return true - } - - /** Forget a room that no longer exists. */ - function forget(documentName: string): void { - holders.delete(documentName) - } - - return { request, grantTo, release, forget } -} - -export type SeedRegistry = ReturnType diff --git a/services/yjs/tests/unit/lib/hooks.spec.ts b/services/yjs/tests/unit/lib/hooks.spec.ts index 2cd4868daa8..b768beb1500 100644 --- a/services/yjs/tests/unit/lib/hooks.spec.ts +++ b/services/yjs/tests/unit/lib/hooks.spec.ts @@ -1,6 +1,6 @@ import type { MockInstance } from 'vitest' import { createHooks, HEALTH_ENDPOINT_PATH } from '../../../src/lib/hooks.ts' -import { SeedMessage } from '../../../src/lib/seedGrant.ts' +import { GrantMessage } from '../../../src/lib/grants.ts' import { DeniedReason, refuse } from '../../../src/lib/errors.ts' import * as graph from '../../../src/lib/graph.ts' @@ -36,10 +36,15 @@ function connection(socketId: string, readOnly = false): FakeConnection { return { socketId, readOnly, sendStateless: vi.fn() } } +// Spelled out: every client version seeds with these. +const SEED_REQUEST = '_oc_seed_request' +const SEED_GRANTED = '_oc_seed_granted' +const SEED_DENIED = '_oc_seed_denied' + function statelessPayload( conn: FakeConnection, documentName = 'doc', - payload: string = SeedMessage.Request + payload: string = SEED_REQUEST ) { return { connection: conn, documentName, payload } as any } @@ -423,8 +428,8 @@ describe('seeding arbitration', () => { await hooks.onStateless(statelessPayload(first)) await hooks.onStateless(statelessPayload(second)) - expect(first.sendStateless).toHaveBeenCalledWith(SeedMessage.Granted) - expect(second.sendStateless).toHaveBeenCalledWith(SeedMessage.Denied) + expect(first.sendStateless).toHaveBeenCalledWith(SEED_GRANTED) + expect(second.sendStateless).toHaveBeenCalledWith(SEED_DENIED) }) // The server rejects their writes, so a read-only grantee would leave the @@ -434,7 +439,7 @@ describe('seeding arbitration', () => { await getHooks().onStateless(statelessPayload(conn)) - expect(conn.sendStateless).toHaveBeenCalledWith(SeedMessage.Denied) + expect(conn.sendStateless).toHaveBeenCalledWith(SEED_DENIED) }) it('answers the holder again with a grant', async () => { @@ -444,7 +449,7 @@ describe('seeding arbitration', () => { await hooks.onStateless(statelessPayload(conn)) await hooks.onStateless(statelessPayload(conn)) - expect(conn.sendStateless).toHaveBeenNthCalledWith(2, SeedMessage.Granted) + expect(conn.sendStateless).toHaveBeenNthCalledWith(2, SEED_GRANTED) }) it('grants each room separately', async () => { @@ -455,8 +460,8 @@ describe('seeding arbitration', () => { await hooks.onStateless(statelessPayload(first, 'doc-1')) await hooks.onStateless(statelessPayload(second, 'doc-2')) - expect(first.sendStateless).toHaveBeenCalledWith(SeedMessage.Granted) - expect(second.sendStateless).toHaveBeenCalledWith(SeedMessage.Granted) + expect(first.sendStateless).toHaveBeenCalledWith(SEED_GRANTED) + expect(second.sendStateless).toHaveBeenCalledWith(SEED_GRANTED) }) it('ignores an unrelated stateless payload', async () => { @@ -476,7 +481,7 @@ describe('seeding arbitration', () => { await hooks.onStateless(statelessPayload(holder)) await hooks.onDisconnect(disconnectPayload({ socketId: 'a', remaining: [peer] })) - expect(peer.sendStateless).toHaveBeenCalledWith(SeedMessage.Granted) + expect(peer.sendStateless).toHaveBeenCalledWith(SEED_GRANTED) }) it('skips read-only peers when passing the grant on', async () => { @@ -489,7 +494,7 @@ describe('seeding arbitration', () => { await hooks.onDisconnect(disconnectPayload({ socketId: 'a', remaining: [viewer, writer] })) expect(viewer.sendStateless).not.toHaveBeenCalled() - expect(writer.sendStateless).toHaveBeenCalledWith(SeedMessage.Granted) + expect(writer.sendStateless).toHaveBeenCalledWith(SEED_GRANTED) }) it('passes the grant on to exactly one writer', async () => { @@ -501,11 +506,11 @@ describe('seeding arbitration', () => { await hooks.onStateless(statelessPayload(holder)) await hooks.onDisconnect(disconnectPayload({ socketId: 'a', remaining: [first, second] })) - expect(first.sendStateless).toHaveBeenCalledWith(SeedMessage.Granted) + expect(first.sendStateless).toHaveBeenCalledWith(SEED_GRANTED) expect(second.sendStateless).not.toHaveBeenCalled() // The new holder is on record: a later request from the other writer is refused. await hooks.onStateless(statelessPayload(second)) - expect(second.sendStateless).toHaveBeenCalledWith(SeedMessage.Denied) + expect(second.sendStateless).toHaveBeenCalledWith(SEED_DENIED) }) it('keeps the grant when someone other than the holder leaves', async () => { @@ -528,6 +533,129 @@ describe('seeding arbitration', () => { await hooks.afterUnloadDocument({ documentName: 'doc' } as any) await hooks.onStateless(statelessPayload(second)) - expect(second.sendStateless).toHaveBeenCalledWith(SeedMessage.Granted) + expect(second.sendStateless).toHaveBeenCalledWith(SEED_GRANTED) + }) + + it('answers a seed request in the format it was asked in', async () => { + const hooks = getHooks() + const first = connection('a') + const second = connection('b') + + await hooks.onStateless(statelessPayload(first, 'doc', GrantMessage.Request + 'seed')) + await hooks.onStateless(statelessPayload(second)) + + expect(first.sendStateless).toHaveBeenCalledWith(GrantMessage.Granted + 'seed') + expect(second.sendStateless).toHaveBeenCalledWith(SEED_DENIED) + }) + + it('denies a seed request in the format it was asked in', async () => { + const hooks = getHooks() + const first = connection('a') + const second = connection('b') + + await hooks.onStateless(statelessPayload(first)) + await hooks.onStateless(statelessPayload(second, 'doc', GrantMessage.Request + 'seed')) + + expect(second.sendStateless).toHaveBeenCalledWith(GrantMessage.Denied + 'seed') + }) + + // The next writer may never have asked, so it may only know the old format. + it('passes a grant asked for in the new format on in the old one', async () => { + const hooks = getHooks() + const holder = connection('a') + const peer = connection('b') + + await hooks.onStateless(statelessPayload(holder, 'doc', GrantMessage.Request + 'seed')) + await hooks.onDisconnect(disconnectPayload({ socketId: 'a', remaining: [peer] })) + + expect(peer.sendStateless).toHaveBeenCalledWith(SEED_GRANTED) + }) + + it.each(['', 'other', 'recover:' + 'x'.repeat(249)])( + 'ignores a request for an unknown or oversized key', + async (key) => { + const conn = connection('a') + + await getHooks().onStateless(statelessPayload(conn, 'doc', GrantMessage.Request + key)) + + expect(conn.sendStateless).not.toHaveBeenCalled() + } + ) +}) + +// Every clean writer hears about an external write and asks to rewrite the +// room from the body it fetched. Exactly one may, or the body lands twice. +describe('recovery arbitration', () => { + const RECOVER = 'recover:etag-1' + + function request(conn: FakeConnection, key = RECOVER) { + return statelessPayload(conn, 'doc', GrantMessage.Request + key) + } + + it('grants the first writer and denies the next', async () => { + const hooks = getHooks() + const first = connection('a') + const second = connection('b') + + await hooks.onStateless(request(first)) + await hooks.onStateless(request(second)) + + expect(first.sendStateless).toHaveBeenCalledWith(GrantMessage.Granted + RECOVER) + expect(second.sendStateless).toHaveBeenCalledWith(GrantMessage.Denied + RECOVER) + }) + + it('grants each etag and the seed separately', async () => { + const hooks = getHooks() + const seeder = connection('a') + const first = connection('b') + const second = connection('c') + + await hooks.onStateless(statelessPayload(seeder)) + await hooks.onStateless(request(first)) + await hooks.onStateless(request(second, 'recover:etag-2')) + + expect(seeder.sendStateless).toHaveBeenCalledWith(SEED_GRANTED) + expect(first.sendStateless).toHaveBeenCalledWith(GrantMessage.Granted + RECOVER) + expect(second.sendStateless).toHaveBeenCalledWith(GrantMessage.Granted + 'recover:etag-2') + }) + + // Only the requester holds the body to recover from, so a grant handed to + // anyone else would be one they cannot use. + it('does not pass the grant on when the holder leaves, but grants the next request', async () => { + const hooks = getHooks() + const holder = connection('a') + const peer = connection('b') + + await hooks.onStateless(request(holder)) + await hooks.onDisconnect(disconnectPayload({ socketId: 'a', remaining: [peer] })) + expect(peer.sendStateless).not.toHaveBeenCalled() + + await hooks.onStateless(request(peer)) + expect(peer.sendStateless).toHaveBeenCalledWith(GrantMessage.Granted + RECOVER) + }) + + it("drops a writer's older recovery grant when it asks for a newer one", async () => { + const hooks = getHooks() + const holder = connection('a') + const peer = connection('b') + + await hooks.onStateless(request(holder)) + await hooks.onStateless(request(holder, 'recover:etag-2')) + await hooks.onStateless(request(peer)) + + expect(peer.sendStateless).toHaveBeenCalledWith(GrantMessage.Granted + RECOVER) + }) + + it('passes only the seed grant on when the holder leaves with both', async () => { + const hooks = getHooks() + const holder = connection('a') + const peer = connection('b') + + await hooks.onStateless(statelessPayload(holder)) + await hooks.onStateless(request(holder)) + await hooks.onDisconnect(disconnectPayload({ socketId: 'a', remaining: [peer] })) + + expect(peer.sendStateless).toHaveBeenCalledOnce() + expect(peer.sendStateless).toHaveBeenCalledWith(SEED_GRANTED) }) })