From 76bbc80a51f62a788e5329b7e79eb9b6dad34f27 Mon Sep 17 00:00:00 2001 From: Jannik Stehle Date: Thu, 17 Sep 2026 07:54:34 +0200 Subject: [PATCH] feat(yjs): immediately notify peers about external doc updates --- dev/docs/yjs.md | 101 ++++- packages/web-client/src/sse/index.ts | 14 + .../components/AppTemplates/AppWrapper.vue | 11 +- .../AppTemplates/useAppWrapperYjs.ts | 98 ++++- packages/web-pkg/src/composables/index.ts | 1 + packages/web-pkg/src/composables/sse/index.ts | 1 + .../composables/sse/useExternalFileUpdates.ts | 131 ++++++ .../src/composables/yjs/useYjsSession.ts | 343 ++++++++++++++-- .../AppTemplates/AppWrapper.spec.ts | 205 +++++++++- .../sse/useExternalFileUpdates.spec.ts | 228 +++++++++++ .../composables/yjs/useYjsSession.spec.ts | 379 +++++++++++++++++- .../web-runtime/src/container/sse/types.ts | 14 +- tests/e2e/features/yjs/conflicts.feature | 131 ++++-- tests/e2e/steps/ui/resources.ts | 8 + .../support/objects/app-files/utils/editor.ts | 4 + 15 files changed, 1548 insertions(+), 121 deletions(-) create mode 100644 packages/web-pkg/src/composables/sse/index.ts create mode 100644 packages/web-pkg/src/composables/sse/useExternalFileUpdates.ts create mode 100644 packages/web-pkg/tests/unit/composables/sse/useExternalFileUpdates.spec.ts diff --git a/dev/docs/yjs.md b/dev/docs/yjs.md index 451ee4ad88b..90e635165d4 100644 --- a/dev/docs/yjs.md +++ b/dev/docs/yjs.md @@ -299,9 +299,9 @@ 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 elected client is the one with the lowest Yjs `clientID`, after a 150 ms pause to let peers announce -themselves. 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 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. ### Stale recovery @@ -338,6 +338,64 @@ block saving - `isDirty` deliberately ignores it so a lock cannot silently disca What actually stops an empty document reaching disk is `serializeDoc` returning `null` when the adapter reports no content. +### External updates + +External updates to a document get also propagated to all peers through the SSE stream. + +```mermaid +sequenceDiagram + participant SSE as SSE stream + participant EU as useExternalFileUpdates + participant AW as useAppWrapperYjs + participant S as Session + participant P as Peers + + SSE->>EU: postprocessing-finished {itemid, initiatorid, etag} + Note over EU: own file? not our tab? etag new?
one at a time, deferred while loading / saving / disconnected + EU->>AW: handleExternalFileEvent + AW->>S: isRoomWrite? (room etag stamp) + Note over AW: GET file -> body, fresh etag + Note over AW: body == server content -> adoptEtag
body == room state -> record as save
else wasWrittenByRoom with a 2 s grace + 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->>AW: onContentChange, onServerContentChange, onEtagChange + else dirty + S->>P: isStale + nativeEtag, no claim + S->>S: markConflicted + end +``` + +The wrapper does the cheap checks first, because most of these events are peer saves the room already knows about. +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. + +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 +a remote flag arrives, so a keystroke that has not been reported yet still turns into a conflict rather than a silent +wipe. + +Two etag-only cases never reach the recovery. When the fetched body equals the last server content, only the etag +moved, and `adoptEtag` records it for the room and the caller without a save stamp, so peers keep their dirty state. +When the body equals the room's merged state, a peer saved and left before its stamp arrived, and the wrapper lands it +like a save. + +While the wrapper cannot act - loading, a PUT in flight, the room disconnected, the session conflicted, read-only - +events are dropped. Nothing is lost by that: the conflicted, locked and read-only states end in a reload that fetches +the file anew, a write during a PUT fails that PUT with a 412, and the load itself fetches the current etag. A +disconnect is the one gap: the SSE stream shares the network with the room, so a write during an outage is announced +to nobody, and the reconnect's drift check compares the room's etag against this client's, which the server never saw +change. Such a write surfaces on the next 412, as it did before SSE events were wired in. Accepted as an edge case. + ### `_oc_meta` A `Y.Map` alongside the editor content, used for coordination the editor never sees: @@ -429,16 +487,19 @@ hold the same content and duplicate the document for everyone. ### File map -| Path | Role | -| ------------------------------------------------------------- | --------------------------------------------------------------- | -| `packages/web-pkg/src/composables/yjs/useYjsSession.ts` | the session: Y.Doc, provider, hydration, staleness, etag mirror | -| `packages/web-pkg/src/composables/yjs/types.ts` | `YjsAdapter` | -| `packages/web-pkg/src/components/AppTemplates/AppWrapper.vue` | owns the session, the save loop and the loading gate | -| `packages/web-pkg/src/components/AppTemplates/types.ts` | `YjsOptions`, `YjsAdapterContext`, slot args | -| `packages/web-pkg/src/editor/yjsAdapter.ts` | `makeTiptapYjsAdapter` - any strategy to a Y.Doc | -| `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) | +| Path | Role | +| ------------------------------------------------------------------ | --------------------------------------------------------------- | +| `packages/web-pkg/src/composables/yjs/useYjsSession.ts` | the session: Y.Doc, provider, hydration, staleness, etag mirror | +| `packages/web-pkg/src/composables/sse/useExternalFileUpdates.ts` | SSE subscription for the open file, filtering and queueing | +| `packages/web-pkg/src/components/AppTemplates/useAppWrapperYjs.ts` | wrapper glue: save conflicts, external update classification | +| `packages/web-pkg/src/composables/yjs/types.ts` | `YjsAdapter` | +| `packages/web-pkg/src/components/AppTemplates/AppWrapper.vue` | owns the session, the save loop and the loading gate | +| `packages/web-pkg/src/components/AppTemplates/types.ts` | `YjsOptions`, `YjsAdapterContext`, slot args | +| `packages/web-pkg/src/editor/yjsAdapter.ts` | `makeTiptapYjsAdapter` - any strategy to a Y.Doc | +| `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 | --- @@ -486,8 +547,14 @@ do meaningfully here. account for the write: the fresh etag must match `_oc_meta.etag`, with a short grace period (`ROOM_ETAG_GRACE_MS`) for a peer's stamp that is still in flight. An external writer - a desktop sync client, say - fails that check and raises the conflict dialog instead of being overwritten. The residual gap is timing: a peer save whose stamp arrives after the -grace period turns into a spurious conflict dialog. +grace period turns into a spurious conflict dialog. The SSE path is less exposed: nobody waits on it, so it gives the +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). -**Stale recovery needs someone who holds the fresh body.** Only a peer whose fetched etag matches `nativeEtag` may -re-seed the room. If that peer leaves before finishing, the room stays flagged until another client opens the file and -picks the job up. Peers already in the room keep editing a document that no longer matches disk in the meantime. +**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-client/src/sse/index.ts b/packages/web-client/src/sse/index.ts index 2d72f38c65a..811e99bf8a2 100644 --- a/packages/web-client/src/sse/index.ts +++ b/packages/web-client/src/sse/index.ts @@ -1,4 +1,18 @@ import { fetchEventSource, FetchEventSourceInit } from '@microsoft/fetch-event-source' +import { z } from 'zod' + +/** Payload of every file and space related SSE event. */ +export const sseEventSchema = z.object({ + itemid: z.string().optional(), + parentitemid: z.string().optional(), + spaceid: z.string().optional(), + initiatorid: z.string().optional(), + etag: z.string().optional(), + affecteduserids: z.array(z.string()).optional().nullable(), + sessionid: z.string().optional() +}) + +export type SseEventData = z.infer export enum MESSAGE_TYPE { NOTIFICATION = 'userlog-notification', diff --git a/packages/web-pkg/src/components/AppTemplates/AppWrapper.vue b/packages/web-pkg/src/components/AppTemplates/AppWrapper.vue index caabeea6df3..a0790b4c9e4 100644 --- a/packages/web-pkg/src/components/AppTemplates/AppWrapper.vue +++ b/packages/web-pkg/src/components/AppTemplates/AppWrapper.vue @@ -255,7 +255,16 @@ const { putFileContents, applySavedResource, runSaveCallback, - onConflict: showExternalUpdateConflict + onConflict: showExternalUpdateConflict, + isSaving: () => saveFileTask.isRunning, + onExternalUpdateApplied: () => { + showMessage({ + title: $gettext('File updated'), + desc: $gettext( + 'This file was updated outside this window. The editor now shows the latest version.' + ) + }) + } }) // Keep the loading screen up until the Yjs session is synced and hydrated, diff --git a/packages/web-pkg/src/components/AppTemplates/useAppWrapperYjs.ts b/packages/web-pkg/src/components/AppTemplates/useAppWrapperYjs.ts index 77f3da800c4..4ecf84558eb 100644 --- a/packages/web-pkg/src/components/AppTemplates/useAppWrapperYjs.ts +++ b/packages/web-pkg/src/components/AppTemplates/useAppWrapperYjs.ts @@ -1,10 +1,21 @@ -import { computed, unref, watch } from 'vue' -import type { Ref } from 'vue' +import { computed, toValue, unref, watch } from 'vue' +import type { MaybeRefOrGetter, Ref } from 'vue' import { useGettext } from 'vue3-gettext' import { call } from '@opencloud-eu/web-client' import type { Resource } from '@opencloud-eu/web-client' -import { useMessages, useYjsSession, YjsStatus } from '../../composables' -import type { AppFileHandlingResult, FileContentOptions, FileContext } from '../../composables' +import { + EXTERNAL_UPDATE_ETAG_GRACE_MS, + useExternalFileUpdates, + useMessages, + useYjsSession, + YjsStatus +} from '../../composables' +import type { + AppFileHandlingResult, + ExternalFileEvent, + FileContentOptions, + FileContext +} from '../../composables' import type { YjsOptions } from './types' export interface AppWrapperYjsOptions { @@ -31,6 +42,10 @@ export interface AppWrapperYjsOptions { runSaveCallback: (content: unknown) => Promise /** The file changed outside this window and the session gave up the room. */ onConflict: () => void + /** Whether a PUT of this wrapper is in flight. External events wait for it. */ + isSaving: MaybeRefOrGetter + /** The file changed outside this window and the editor now shows that version. */ + onExternalUpdateApplied: () => void } /** @@ -57,7 +72,9 @@ export function useAppWrapperYjs(options: AppWrapperYjsOptions) { putFileContents, applySavedResource, runSaveCallback, - onConflict + onConflict, + isSaving, + onExternalUpdateApplied } = options const { $gettext } = useGettext() @@ -99,7 +116,8 @@ export function useAppWrapperYjs(options: AppWrapperYjsOptions) { resource.value = { ...unref(resource), etag: value } }, hasUnsavedChanges: () => unref(isDirty), - onConflict + onConflict, + onExternalUpdate: onExternalUpdateApplied }) : null @@ -181,5 +199,73 @@ export function useAppWrapperYjs(options: AppWrapperYjsOptions) { /** True once the session gave up the room over an external write. */ const isConflicted = computed(() => Boolean(unref(session?.isConflicted))) + /** Whether an SSE event may be acted on right now. */ + const canHandleExternalEvents = computed(() => { + if (!session) return false + if (unref(loading) || unref(loadingError) || !unref(session.isReady)) return false + if (unref(session.isConflicted) || unref(session.isLockedForReload)) return false + if (unref(isReadOnly) || toValue(isSaving)) return false + const status = unref(session.status) + return status === YjsStatus.Connected || status === YjsStatus.Local + }) + + /** + * The server announced a write to this file by someone else. Most of these + * are peer saves the room already knows about, so the cheap checks come + * first. + */ + async function handleExternalFileEvent({ etag }: ExternalFileEvent) { + if (!session) return + if (etag === unref(currentETag)) return + // A peer's stamp may still be on the way; resolves as soon as it lands. + if (await session.wasWrittenByRoom(etag, EXTERNAL_UPDATE_ETAG_GRACE_MS)) return + if (!unref(canHandleExternalEvents)) return + + let fresh: Awaited> + try { + fresh = await getFileContents(currentFileContext, { ...fileContentOptions }) + } catch (e) { + console.error('[yjs] fetching an externally updated file failed:', e) + return + } + // A save or a conflict may have started while we waited. Same after + // every await below; the save's own 412 handling covers a write we skip. + if (!unref(canHandleExternalEvents)) return + const freshEtag = fresh.headers['OC-ETag'] + const body = fresh.body + if (typeof body !== 'string') return + if (!freshEtag || freshEtag === unref(currentETag)) return + if (session.isRoomWrite(freshEtag)) return + + // Same body we last saw on disk: only the etag moved (a rename, a + // re-upload of the same bytes). Not a save, so peers keep their dirty state. + if (body === unref(serverContent)) { + session.adoptEtag(freshEtag) + return + } + + // The room's own state is what landed: a peer saved and left before its + // stamp reached us. Land it like the conflict retry does. + const merged = await session.serializeMerged() + if (!unref(canHandleExternalEvents)) return + if (merged !== null && body === merged) { + serverContent.value = currentContent.value = body + applySavedResource(freshEtag) + await runSaveCallback(body) + return + } + + await session.applyExternalUpdate({ content: body, etag: freshEtag }) + } + + if (session) { + useExternalFileUpdates({ + resource, + currentETag, + isActive: canHandleExternalEvents, + onExternalEvent: handleExternalFileEvent + }) + } + return { session, isSessionReady, effectiveReadOnly, isConflicted, reconcileConflict } } diff --git a/packages/web-pkg/src/composables/index.ts b/packages/web-pkg/src/composables/index.ts index 302464f9a2c..86f247c949b 100644 --- a/packages/web-pkg/src/composables/index.ts +++ b/packages/web-pkg/src/composables/index.ts @@ -35,6 +35,7 @@ export * from './selection' export * from './service' export * from './shares' export * from './sort' +export * from './sse' export * from './spaces' export * from './upload' export * from './viewMode' diff --git a/packages/web-pkg/src/composables/sse/index.ts b/packages/web-pkg/src/composables/sse/index.ts new file mode 100644 index 00000000000..8c20b1ab8c1 --- /dev/null +++ b/packages/web-pkg/src/composables/sse/index.ts @@ -0,0 +1 @@ +export * from './useExternalFileUpdates' diff --git a/packages/web-pkg/src/composables/sse/useExternalFileUpdates.ts b/packages/web-pkg/src/composables/sse/useExternalFileUpdates.ts new file mode 100644 index 00000000000..817584c7874 --- /dev/null +++ b/packages/web-pkg/src/composables/sse/useExternalFileUpdates.ts @@ -0,0 +1,131 @@ +import { computed, onBeforeUnmount, toValue, unref, watch } from 'vue' +import type { MaybeRefOrGetter } from 'vue' +import type { Resource } from '@opencloud-eu/web-client' +import { MESSAGE_TYPE, sseEventSchema } from '@opencloud-eu/web-client/sse' +import { useClientService } from '../clientService' +import { useAuthStore, useCapabilityStore } from '../piniaStores' + +/** A write to the open file that did not come from this tab. */ +export interface ExternalFileEvent { + /** The etag the file carries after the write, as announced by the server. */ + etag: string +} + +export interface ExternalFileUpdatesOptions { + resource: MaybeRefOrGetter + /** The etag this client believes is on disk. */ + currentETag: MaybeRefOrGetter + /** Whether an event may be handled right now. Events arriving while false are dropped. */ + isActive: MaybeRefOrGetter + /** + * Handles one event. Runs one at a time; an event arriving meanwhile runs + * afterwards, and a newer one replaces it. + */ + onExternalEvent: (event: ExternalFileEvent) => Promise +} + +/** SSE topics that mean the file body may have changed. */ +const CONTENT_TOPICS = [MESSAGE_TYPE.POSTPROCESSING_FINISHED, MESSAGE_TYPE.FILE_TOUCHED] + +function parseEventData(data: unknown) { + if (typeof data !== 'string') return null + try { + const parsed = sseEventSchema.safeParse(JSON.parse(data)) + return parsed.success ? parsed.data : null + } catch { + return null + } +} + +/** + * Listens for server-sent file events about one resource and hands the ones + * that mean "someone else wrote this file" to the caller, one at a time. + * + * Only a signed-in user has an SSE stream. + */ +export function useExternalFileUpdates({ + resource, + currentETag, + isActive, + onExternalEvent +}: ExternalFileUpdatesOptions) { + const clientService = useClientService() + const capabilityStore = useCapabilityStore() + const authStore = useAuthStore() + + const isEnabled = computed( + () => + Boolean(unref(capabilityStore.supportSSE)) && + Boolean(authStore.accessToken) && + !authStore.publicLinkContextReady + ) + + /** The event that arrived while a handler ran; only the newest is kept. */ + let pending: ExternalFileEvent | null = null + let inFlight = false + + async function handle(event: ExternalFileEvent) { + if (inFlight) { + pending = event + return + } + inFlight = true + try { + await onExternalEvent(event) + } catch (e) { + console.error('[sse] handling an external file update failed:', e) + } finally { + inFlight = false + } + const next = pending + pending = null + if (next && toValue(isActive)) void handle(next) + } + + function onMessage(msg: MessageEvent) { + const data = parseEventData(msg.data) + if (!data) return + const current = toValue(resource) + // The composite id all peers share; a share recipient's `id` differs. + const fileId = current?.fileId ?? current?.id + if (!fileId || data.itemid !== fileId) return + // Our own PUT: the save path already applied its etag. + if (data.initiatorid && data.initiatorid === clientService.initiatorId) return + // Both topics stat the file server-side, so the etag is always set. + if (!data.etag || data.etag === toValue(currentETag)) return + if (!toValue(isActive)) return + void handle({ etag: data.etag }) + } + + let stream: EventSource | null = null + function subscribe() { + if (stream) return + stream = clientService.sseAuthenticated + for (const topic of CONTENT_TOPICS) { + stream.addEventListener(topic, onMessage) + } + } + function unsubscribe() { + if (!stream) return + for (const topic of CONTENT_TOPICS) { + stream.removeEventListener(topic, onMessage) + } + stream = null + } + + watch( + isEnabled, + (enabled) => { + if (enabled) subscribe() + else unsubscribe() + }, + { immediate: true } + ) + watch( + () => toValue(resource)?.id, + () => { + pending = null + } + ) + onBeforeUnmount(unsubscribe) +} diff --git a/packages/web-pkg/src/composables/yjs/useYjsSession.ts b/packages/web-pkg/src/composables/yjs/useYjsSession.ts index db0319cf7c4..ac1b20b3874 100644 --- a/packages/web-pkg/src/composables/yjs/useYjsSession.ts +++ b/packages/web-pkg/src/composables/yjs/useYjsSession.ts @@ -72,8 +72,13 @@ export interface YjsSessionOptions { hasUnsavedChanges: MaybeRefOrGetter /** The session gave up the room because the file changed outside it. */ onConflict: () => void + /** The room has been updated with an external write. Fired once per update. */ + onExternalUpdate?: () => void } +/** Outcome of {@link YjsSession.applyExternalUpdate}. */ +export type ExternalUpdateResult = 'recovered' | 'conflicted' | 'skipped' + export interface YjsSession { ydoc: ShallowRef awareness: ShallowRef @@ -125,6 +130,22 @@ export interface YjsSession { * peer edits that caused the conflict. */ serializeMerged: () => Promise + /** + * Whether a write announced with this etag came from inside the room, i.e. + * the room's etag stamp already names it. Always false in local mode. + */ + isRoomWrite: (etag: string) => boolean + /** + * Stamp the given etag on the room without claiming a save, so peers keep + * their dirty state. + */ + adoptEtag: (etag: string) => void + /** + * The file changed outside the room and the caller fetched the new body. + * Recovers the room from it when nothing is unsaved, otherwise flags the + * room and conflicts. + */ + applyExternalUpdate: (update: { content: string; etag: string }) => Promise } const META_KEY = '_oc_meta' @@ -140,11 +161,15 @@ 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 + HYDRATE_ORIGIN, + EXTERNAL_UPDATE_ORIGIN, + ETAG_ADOPT_ORIGIN ] /** * How long `wasWrittenByRoom` waits for a peer's etag stamp that may still be @@ -159,6 +184,11 @@ const ROOM_ETAG_GRACE_MS = 1_000 const SEED_REQUEST = '_oc_seed_request' const SEED_GRANTED = '_oc_seed_granted' const SEED_DENIED = '_oc_seed_denied' +/** + * How long an SSE-triggered check waits for a peer's etag stamp before it + * treats the write as external. + */ +export const EXTERNAL_UPDATE_ETAG_GRACE_MS = 2_000 const FALLBACK_WEB_VERSION = '0.0.0' function resolveWebVersion(): string { @@ -248,7 +278,8 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { onServerContentChange, onEtagChange, hasUnsavedChanges, - onConflict + onConflict, + onExternalUpdate } = options const { $gettext } = useGettext() @@ -312,13 +343,17 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { */ let hasLocalOnlyContent = false + /** What a recovery publishes: a native file body and the etag it was read under. */ + interface RecoveryPayload { + content: string + etag: string + } + /** - * The native file body captured at the moment etag drift was detected. - * Recovery must publish exactly this body; reading `currentContent` at - * recovery time would return the room's own (older) state, which the - * debounced serialize has reported back by then. + * The native file body and its etag captured at the moment etag drift was detected. + * Recovery must publish exactly this body. */ - let staleRecoveryContent: string | null = null + let staleRecovery: RecoveryPayload | null = null /** * The doc state behind the last content handed to the caller, i.e. what its @@ -396,9 +431,17 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { * `currentContent`) and name ourselves. Last write wins, so concurrent * claims elect exactly one client. */ - function claimRecovery(doc: Y.Doc, meta: SessionMetaMap) { - staleRecoveryContent = toValue(currentContent) - doc.transact(() => meta.set('recoveryClientId', doc.clientID)) + 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) } /** @@ -409,16 +452,29 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { * without claiming; the `nativeEtag` it leaves lets the first writer * holding the same body pick the claim up. */ - function flagStale(doc: Y.Doc, meta: SessionMetaMap) { - const nativeEtag = toValue(resource)?.etag + 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 (!unref(effectiveReadOnly)) claimRecovery(doc, meta) + if (claim) claimRecovery(doc, meta, payload, origin) doc.transact(() => { if (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. */ @@ -453,7 +509,8 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { if (unref(isConflicted)) return // Local mode has no room to leave and no peer that could have written the // file. Conflicting would only stop its autosave, which is the single - // safety net there. + // safety net there. `applyExternalUpdate` is the one path that knows + // better, see there. if (unref(status) === YjsStatus.Local) return isConflicted.value = true const prov = unref(provider) @@ -680,6 +737,9 @@ 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 + /** * Stale-state recovery, fired when `_oc_meta.isStale` goes up: the claimed * client wipes the adapter content, clears the flags and re-hydrates from @@ -691,12 +751,31 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { * 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. + * + * True when this run rewrote the room. */ - async function recoverFromStaleState(doc: Y.Doc, prov: HocuspocusProvider | null) { + async function recoverFromStaleState( + doc: Y.Doc, + prov: HocuspocusProvider | null + ): 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 (unref(effectiveReadOnly)) return - if (staleRecoveryContent === null) return + if (staleRecovery === null) return false if (typeof current.reset !== 'function') { lockForReload( prov, @@ -704,16 +783,24 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { 'This file was changed externally and your editor cannot recover in-place. Please reload.' ) ) - return + return false } // Let concurrent claims converge, then check we came out on top. await new Promise((resolve) => setTimeout(resolve, 150)) - if (meta.get('isStale') !== true) return // someone else handled it - if (meta.get('recoveryClientId') !== doc.clientID) return + 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 (meta.get('recoveryClientId') !== doc.clientID) { + staleRecovery = null + return false + } - const content = staleRecoveryContent - const freshEtag = meta.get('nativeEtag') ?? toValue(resource)?.etag ?? '' + const { content, etag: freshEtag } = staleRecovery // Reset and hydrate share one transaction: mounted editors must never // observe the emptied doc, or ProseMirror pads it with an empty paragraph @@ -739,7 +826,8 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { meta.set('lastSavedAt', Date.now()) }, STALE_RECOVERY_COMMIT_ORIGIN) - staleRecoveryContent = null + staleRecovery = null + return true } catch (e) { // The reset already emptied the shared doc for every peer; `isStale` // stays up so a later joiner holding the fresh body retries. Lock so @@ -749,6 +837,7 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { prov, $gettext('This file was changed externally and recovering it failed. Please reload.') ) + return false } finally { isRewritingDoc = false } @@ -784,34 +873,64 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { */ function createContentReporter(doc: Y.Doc) { let timer: number | undefined + let inFlight: Promise | null = null + function serialize(): Promise { + if (doc.isDestroyed) return Promise.resolve() + // Vector taken before serializing: adapters may serialize + // asynchronously, and a peer update landing in between must not be + // counted as part of what we reported. + const vectorAtSerialize = Y.encodeStateVector(doc) + return serializeDoc(doc).then((value) => { + if (value === null) return + lastReportedStateVector = vectorAtSerialize + lastReportedContent = value + onContentChange(value) + }) + } + function report(): Promise { + // Re-checked: the debounce window can outlive the change that opened it. + if (!canReportContent()) return Promise.resolve() + // Chained onto a running report: two concurrent serializations could + // settle out of order, and the older one would then win. + const previous = inFlight + const run: Promise = (previous ? previous.then(serialize) : serialize()) + .catch((e) => console.error('[yjs] serialize for content update failed:', e)) + .finally(() => { + if (inFlight === run) inFlight = null + }) + inFlight = run + return run + } function onDocUpdate() { if (!canReportContent()) return if (timer !== undefined) window.clearTimeout(timer) timer = window.setTimeout(() => { timer = undefined - // Re-checked: the debounce window can outlive the change that opened it. - if (!canReportContent()) return - // Vector taken before serializing: adapters may serialize - // asynchronously, and a peer update landing in between must not be - // counted as part of what we reported. - const vectorAtSerialize = Y.encodeStateVector(doc) - serializeDoc(doc) - .then((value) => { - if (value === null) return - lastReportedStateVector = vectorAtSerialize - lastReportedContent = value - onContentChange(value) - }) - .catch((e) => console.error('[yjs] serialize for content update failed:', e)) + void report() }, SERIALIZE_DEBOUNCE_MS) } + /** + * Run a pending report now. The caller's dirty state lags the doc by the + * debounce, and a decision about unsaved work must not miss a keystroke + * from inside that window. A doc with no pending report stays untouched. + */ + function flush(): Promise { + if (timer === undefined) return inFlight ?? Promise.resolve() + window.clearTimeout(timer) + timer = undefined + return report() + } function cancel() { if (timer !== undefined) window.clearTimeout(timer) + timer = undefined } - return { onDocUpdate, cancel } + return { onDocUpdate, flush, cancel } } type ContentReporter = ReturnType + /** The reporter of the live session, for code outside the session watch. */ + let activeReporter: ContentReporter | null = null + /** The user-facing side of a refused handshake. */ function deniedMessage(reason: string): string { switch (reason) { @@ -1003,6 +1122,16 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { const newEtag = meta.get('etag') if (newEtag) onEtagChange(newEtag) } + // A peer's recovery commit: the flag goes down with the etag it + // settled on. Only now does this doc show the external write. + if ( + event.keysChanged.has('isStale') && + meta.get('isStale') !== true && + event.keysChanged.has('etag') && + !unref(isConflicted) + ) { + onExternalUpdate?.() + } if (event.keysChanged.has('lastSavedAt')) { const theirState = meta.get('savedStateVector') if (theirState instanceof Uint8Array) { @@ -1043,13 +1172,138 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { } // Stale-state signal: every peer sees it, only the claimed one gets - // past the guards in `recoverFromStaleState`. + // 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) } } } + /** + * 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). + */ + 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 + if (unref(isConflicted) || unref(isLockedForReload)) return + 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) + if (!doc || doc.isDestroyed || !unref(provider)) return false + return Boolean(etag) && sessionMeta(doc).get('etag') === etag + } + + /** See {@link YjsSession.adoptEtag}. */ + function adoptEtag(etag: string) { + const doc = unref(ydoc) + if (doc && !doc.isDestroyed) { + const meta = sessionMeta(doc) + // No `lastSavedAt`: this is not a save, peers must keep their dirty + // 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) + } + } + 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. + */ + async function reportRecovered(doc: Y.Doc, etag: string) { + const value = await serializeDoc(doc) + if (value === null || doc.isDestroyed || unref(ydoc) !== doc) return + lastReportedStateVector = Y.encodeStateVector(doc) + lastReportedContent = value + onContentChange(value) + onServerContentChange(value) + onEtagChange(etag) + } + + /** See {@link YjsSession.applyExternalUpdate}. */ + async function applyExternalUpdate({ + content, + etag + }: { + content: string + etag: string + }): Promise { + const doc = unref(ydoc) + const prov = unref(provider) + if (!doc || doc.isDestroyed || !unref(isReady) || !etag) return 'skipped' + if (unref(effectiveReadOnly) || unref(isConflicted) || unref(isLockedForReload)) { + return 'skipped' + } + const meta = sessionMeta(doc) + if (meta.get('etag') === etag) return 'skipped' + + await activeReporter?.flush() + if (doc.isDestroyed || unref(ydoc) !== doc) return 'skipped' + + if (toValue(hasUnsavedChanges)) { + if (!prov) { + // Local mode keeps autosaving through `markConflicted` on purpose, but + // here the autosave would only hit 412 over the same write every + // interval, so stop it. + isConflicted.value = true + 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 }) + 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' + await reportRecovered(doc, etag) + onExternalUpdate?.() + return 'recovered' + } + /** * Y.Doc + (optional) provider lifecycle, rebuilt whenever the session key * changes. Remote mode connects a Hocuspocus provider and hydrates on @@ -1068,7 +1322,7 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { hasLocalOnlyContent = false // `undoResyncWipe` leaves this up for good when it cannot restore. isRewritingDoc = false - staleRecoveryContent = null + staleRecovery = null lastReportedStateVector = null lastReportedContent = null pendingSaveStateVector = null @@ -1082,6 +1336,7 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { const doc = new Y.Doc() const reporter = createContentReporter(doc) + activeReporter = reporter doc.on('update', reporter.onDocUpdate) let prov: HocuspocusProvider | null = null @@ -1112,6 +1367,7 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { onCleanup(() => { reporter.cancel() + if (activeReporter === reporter) activeReporter = null clearConnectTimer() abandonSeedGrant() meta.map.unobserve(metaObserver) @@ -1170,6 +1426,9 @@ export function useYjsSession(options: YjsSessionOptions): YjsSession { error, wasWrittenByRoom, beginSave, - serializeMerged + serializeMerged, + isRoomWrite, + adoptEtag, + applyExternalUpdate } } diff --git a/packages/web-pkg/tests/unit/components/AppTemplates/AppWrapper.spec.ts b/packages/web-pkg/tests/unit/components/AppTemplates/AppWrapper.spec.ts index 9ab48291005..80afd287db2 100644 --- a/packages/web-pkg/tests/unit/components/AppTemplates/AppWrapper.spec.ts +++ b/packages/web-pkg/tests/unit/components/AppTemplates/AppWrapper.spec.ts @@ -56,6 +56,7 @@ afterEach(() => { const FILE_A = mock({ id: 'storage$space!file-a', + fileId: 'storage$space!file-a', name: 'a.md', etag: 'etag-a', permissions: 'RDNVW' @@ -79,11 +80,18 @@ function setup({ // session that cannot produce one, which falls back to the pre-conflict // snapshot. mergedContent = null as string | null, - putFileContents = vi.fn().mockResolvedValue(mock({ etag: 'etag-saved' })) + putFileContents = vi.fn().mockResolvedValue(mock({ etag: 'etag-saved' })), + // Whether the room recognises an announced write as one of its own. + roomWrite = false, + // Whether the runtime has an authenticated SSE stream for this user. + sse = true } = {}) { const wasWrittenByRoom = vi.fn().mockResolvedValue(writtenByRoom) const beginSave = vi.fn() const serializeMerged = vi.fn().mockResolvedValue(mergedContent) + const isRoomWrite = vi.fn().mockReturnValue(roomWrite) + const adoptEtag = vi.fn() + const applyExternalUpdate = vi.fn().mockResolvedValue('recovered') const isLockedForReload = ref(false) const isConflicted = ref(false) const markConflicted = vi.fn(() => { @@ -130,7 +138,10 @@ function setup({ error: ref(null) as any, wasWrittenByRoom, beginSave, - serializeMerged + serializeMerged, + isRoomWrite, + adoptEtag, + applyExternalUpdate }) }) @@ -140,6 +151,16 @@ function setup({ params: { driveAliasAndItem: 'personal/alan/a.md' } }) }) + // A deep mock hands back a function for every property; the wrapper + // compares this one against the initiator in SSE events. + ;(mocks.$clientService as any).initiatorId = 'tab-own' + const sseListeners = new Map void>() + vi.mocked(mocks.$clientService.sseAuthenticated.addEventListener).mockImplementation((( + topic: string, + listener: (msg: MessageEvent) => void + ) => { + sseListeners.set(topic, listener) + }) as any) let slotProps: any const wrapper = mount(AppWrapper, { @@ -158,7 +179,9 @@ function setup({ plugins: [ ...defaultPlugins({ piniaOptions: { - appsState: { apps: { 'test-app': { id: 'test-app', name: 'Test App' } } } + appsState: { apps: { 'test-app': { id: 'test-app', name: 'Test App' } } }, + authState: { accessToken: sse ? 'token' : undefined }, + capabilityState: { capabilities: { core: { 'support-sse': sse } as any } } } }), createRouter({ @@ -185,6 +208,23 @@ function setup({ wasWrittenByRoom, beginSave, serializeMerged, + isRoomWrite, + adoptEtag, + applyExternalUpdate, + sseListeners, + /** The server announces a write to the open file by another tab. */ + async emitFileEvent(data: Record = {}) { + const payload = { + itemid: FILE_A.id, + initiatorid: 'tab-other', + etag: 'etag-external', + ...data + } + sseListeners.get('postprocessing-finished')?.({ + data: JSON.stringify(payload) + } as MessageEvent) + await flushPromises() + }, currentContent: () => slotProps?.currentContent, yjsStatus: () => slotProps?.yjsStatus, statusRef, @@ -210,6 +250,9 @@ function setup({ ) await flushPromises() }, + resolveContentWith(body: string, etag: string) { + resolveContents(mock({ body, headers: { 'OC-ETag': etag } as any })) + }, rejectContent(error: Error) { rejectContents(error) } @@ -770,3 +813,159 @@ describe('AppWrapper — locked session', () => { expect(s.putFileContents).not.toHaveBeenCalled() }) }) + +describe('AppWrapper — external file updates via SSE', () => { + async function openFile(options: Parameters[0] = {}) { + const s = setup(options) + await nextTick() + await s.resolveResource(FILE_A) + await s.resolveContent('content of a') + return s + } + + it('listens for content events once the user has an SSE stream', async () => { + const s = await openFile() + + expect(s.sseListeners.has('postprocessing-finished')).toBe(true) + expect(s.sseListeners.has('file-touched')).toBe(true) + }) + + it('does not listen without an SSE stream', async () => { + const s = await openFile({ sse: false }) + + expect(s.sseListeners.size).toBe(0) + }) + + it('ignores a write the room can account for without fetching', async () => { + const s = await openFile({ writtenByRoom: true }) + s.getFileContents.mockClear() + + await s.emitFileEvent() + + expect(s.wasWrittenByRoom).toHaveBeenCalledWith('etag-external', expect.any(Number)) + expect(s.getFileContents).not.toHaveBeenCalled() + expect(s.applyExternalUpdate).not.toHaveBeenCalled() + }) + + it('waits for a late peer stamp before fetching', async () => { + let settle: (v: boolean) => void + const s = await openFile({ writtenByRoom: false }) + s.wasWrittenByRoom.mockReturnValueOnce(new Promise((r) => (settle = r))) + s.getFileContents.mockClear() + + const pending = s.emitFileEvent() + await flushPromises() + expect(s.getFileContents).not.toHaveBeenCalled() + + settle(true) + await pending + expect(s.getFileContents).not.toHaveBeenCalled() + expect(s.applyExternalUpdate).not.toHaveBeenCalled() + }) + + it('adopts the etag when the body on disk is the one we know', async () => { + const s = await openFile({ writtenByRoom: false }) + + await s.emitFileEvent() + s.resolveContentWith('content of a', 'etag-external') + await flushPromises() + + expect(s.adoptEtag).toHaveBeenCalledWith('etag-external') + expect(s.applyExternalUpdate).not.toHaveBeenCalled() + expect(s.putFileContents).not.toHaveBeenCalled() + }) + + // A peer saved and left before its stamp reached us: what is on disk is the + // room's own state, so it lands as a save. + it('records the room state as saved when that is what landed', async () => { + const callback = vi.fn() + const s = await openFile({ mergedContent: 'merged room state', writtenByRoom: false }) + await s.edit('merged room state') + s.registerSaveCallback(callback) + + await s.emitFileEvent() + s.resolveContentWith('merged room state', 'etag-external') + await flushPromises() + + expect(s.applyExternalUpdate).not.toHaveBeenCalled() + expect(unref(s.session().resource).etag).toBe('etag-external') + // The wrapped app never saves this body itself, so it hears about it here. + expect(callback).toHaveBeenCalledWith('merged room state') + // Nothing left to save. + await s.pressCtrlS() + expect(s.putFileContents).not.toHaveBeenCalled() + }) + + it('hands a foreign body to the session with the etag it was read under', async () => { + const s = await openFile({ writtenByRoom: false }) + + await s.emitFileEvent({ etag: 'etag-announced' }) + // The file moved again between the event and our GET. + s.resolveContentWith('desktop client wrote this', 'etag-fetched') + await flushPromises() + + expect(s.applyExternalUpdate).toHaveBeenCalledWith({ + content: 'desktop client wrote this', + etag: 'etag-fetched' + }) + }) + + it('shows the update message when the session reports an applied update', async () => { + const s = await openFile() + + s.session().onExternalUpdate() + await nextTick() + + const { showMessage } = useMessages() + expect(showMessage).toHaveBeenCalledTimes(1) + expect(vi.mocked(showMessage).mock.calls[0][0].desc).toContain('updated outside this window') + }) + + it('leaves the session alone when the fetch fails', async () => { + const s = await openFile({ writtenByRoom: false }) + + await s.emitFileEvent() + s.rejectContent(new Error('network')) + await flushPromises() + + expect(s.applyExternalUpdate).not.toHaveBeenCalled() + expect(consoleErrorSpy).toHaveBeenCalled() + }) + + // The PUT's `If-Match` catches that write as a 412. + it('ignores an event while a save is in flight', async () => { + let resolvePut: (r: Resource) => void + const putFileContents = vi.fn(() => new Promise((r) => (resolvePut = r))) + const s = await openFile({ putFileContents }) + await s.edit('edited content') + s.getFileContents.mockClear() + + const pressed = s.pressCtrlS() + await s.emitFileEvent() + resolvePut!(mock({ etag: 'etag-saved' })) + await pressed + await flushPromises() + + expect(s.getFileContents).not.toHaveBeenCalled() + }) + + it('ignores an event while the session is disconnected', async () => { + const s = await openFile({ status: 'disconnected' }) + s.getFileContents.mockClear() + + await s.emitFileEvent() + + expect(s.getFileContents).not.toHaveBeenCalled() + }) + + it('never fetches for a conflicted session', async () => { + const s = await openFile() + s.isConflicted.value = true + await nextTick() + s.getFileContents.mockClear() + + await s.emitFileEvent() + + expect(s.getFileContents).not.toHaveBeenCalled() + }) +}) diff --git a/packages/web-pkg/tests/unit/composables/sse/useExternalFileUpdates.spec.ts b/packages/web-pkg/tests/unit/composables/sse/useExternalFileUpdates.spec.ts new file mode 100644 index 00000000000..66f7f138468 --- /dev/null +++ b/packages/web-pkg/tests/unit/composables/sse/useExternalFileUpdates.spec.ts @@ -0,0 +1,228 @@ +import { nextTick, ref, unref } from 'vue' +import { flushPromises } from '@vue/test-utils' +import type { Resource } from '@opencloud-eu/web-client' +import { MESSAGE_TYPE } from '@opencloud-eu/web-client/sse' +import { getComposableWrapper } from '@opencloud-eu/web-test-helpers' +import { useAuthStore } from '../../../../src/composables/piniaStores' +import { useExternalFileUpdates } from '../../../../src/composables/sse' + +const FILE = { + id: 'storage$space!file-1', + fileId: 'storage$space!file-1', + etag: 'etag-1' +} as Resource + +function setup({ + supportSSE = true, + accessToken = 'token', + publicLinkContextReady = false, + isActive = true, + resource = FILE as Resource | undefined +} = {}) { + const listeners = new Map void>() + const sseAuthenticated = { + addEventListener: vi.fn((topic: string, listener: (msg: MessageEvent) => void) => { + listeners.set(topic, listener) + }), + removeEventListener: vi.fn((topic: string) => { + listeners.delete(topic) + }) + } + const clientService = { initiatorId: 'tab-own', sseAuthenticated } + const resourceRef = ref(resource) + const currentETag = ref('etag-1') + const isActiveRef = ref(isActive) + // Resolved by the test, so it can observe the queue while a handler runs. + const handlerDone: Array<() => void> = [] + const onExternalEvent = vi.fn(() => new Promise((resolve) => handlerDone.push(resolve))) + const wrapper = getComposableWrapper( + () => { + useExternalFileUpdates({ + resource: resourceRef, + currentETag, + isActive: isActiveRef, + onExternalEvent + }) + }, + { + provide: { $clientService: clientService }, + pluginOptions: { + piniaOptions: { + authState: { accessToken, publicLinkContextReady }, + capabilityState: { capabilities: { core: { 'support-sse': supportSSE } as any } } + } + } + } + ) + + return { + wrapper, + sseAuthenticated, + clientService, + resourceRef, + currentETag, + isActiveRef, + onExternalEvent, + emit(topic: string, data: unknown) { + const payload = typeof data === 'string' ? data : JSON.stringify(data) + listeners.get(topic)?.({ data: payload } as MessageEvent) + }, + async finishHandler() { + handlerDone.shift()?.() + await flushPromises() + } + } +} + +const event = (overrides: Record = {}) => ({ + itemid: FILE.id, + initiatorid: 'tab-other', + etag: 'etag-2', + ...overrides +}) + +describe('useExternalFileUpdates', () => { + it('subscribes to both content topics and unsubscribes on unmount', () => { + const s = setup() + + expect(s.sseAuthenticated.addEventListener).toHaveBeenCalledWith( + MESSAGE_TYPE.POSTPROCESSING_FINISHED, + expect.any(Function) + ) + expect(s.sseAuthenticated.addEventListener).toHaveBeenCalledWith( + MESSAGE_TYPE.FILE_TOUCHED, + expect.any(Function) + ) + + s.wrapper.unmount() + + expect(s.sseAuthenticated.removeEventListener).toHaveBeenCalledTimes(2) + }) + + it.each([ + ['SSE unsupported', { supportSSE: false }], + ['no access token', { accessToken: '' }], + ['public link context', { publicLinkContextReady: true }] + ])('does not subscribe with %s', (_, options) => { + const s = setup(options) + + expect(s.sseAuthenticated.addEventListener).not.toHaveBeenCalled() + }) + + it('hands a foreign write to the handler', async () => { + const s = setup() + + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event()) + await flushPromises() + + expect(s.onExternalEvent).toHaveBeenCalledWith({ etag: 'etag-2' }) + }) + + // `resetSSE` on logout nulls the singleton, and the getter would build a + // fresh, tokenless stream if unsubscribing went through it. + it('unsubscribes from the stream it subscribed to, not a fresh one', async () => { + const s = setup() + const fresh = { addEventListener: vi.fn(), removeEventListener: vi.fn() } + s.clientService.sseAuthenticated = fresh as any + + useAuthStore().accessToken = null + await nextTick() + + expect(s.sseAuthenticated.removeEventListener).toHaveBeenCalledTimes(2) + expect(fresh.removeEventListener).not.toHaveBeenCalled() + expect(fresh.addEventListener).not.toHaveBeenCalled() + }) + + it.each([ + ['another file', event({ itemid: 'storage$space!file-9' })], + ['our own write', event({ initiatorid: 'tab-own' })], + ['the etag we already have', event({ etag: 'etag-1' })], + ['no etag', event({ etag: undefined })], + ['malformed json', '{not json'] + ])('ignores %s', async (_, data) => { + const s = setup() + + s.emit(MESSAGE_TYPE.FILE_TOUCHED, data) + await flushPromises() + + expect(s.onExternalEvent).not.toHaveBeenCalled() + }) + + it('matches a share recipient by the composite file id', async () => { + const s = setup({ + resource: { id: 'share-id', fileId: FILE.id, etag: 'etag-1' } as Resource + }) + + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event()) + await flushPromises() + + expect(s.onExternalEvent).toHaveBeenCalledTimes(1) + }) + + it('drops events while inactive', async () => { + const s = setup({ isActive: false }) + + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event({ etag: 'etag-2' })) + await flushPromises() + s.isActiveRef.value = true + await nextTick() + await flushPromises() + + expect(s.onExternalEvent).not.toHaveBeenCalled() + }) + + it('runs one handler at a time and coalesces events that arrive meanwhile', async () => { + const s = setup() + + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event({ etag: 'etag-2' })) + await flushPromises() + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event({ etag: 'etag-3' })) + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event({ etag: 'etag-4' })) + await flushPromises() + expect(s.onExternalEvent).toHaveBeenCalledTimes(1) + + await s.finishHandler() + + expect(s.onExternalEvent).toHaveBeenCalledTimes(2) + expect(s.onExternalEvent).toHaveBeenLastCalledWith({ etag: 'etag-4' }) + }) + + it('drops a queued event once the wrapper turned inactive', async () => { + const s = setup() + + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event({ etag: 'etag-2' })) + await flushPromises() + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event({ etag: 'etag-3' })) + s.isActiveRef.value = false + await s.finishHandler() + + expect(s.onExternalEvent).toHaveBeenCalledTimes(1) + }) + + it('drops a queued event when the resource changes', async () => { + const s = setup() + + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event({ etag: 'etag-2' })) + await flushPromises() + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event({ etag: 'etag-3' })) + s.resourceRef.value = { ...unref(s.resourceRef), id: 'other', fileId: 'other' } as Resource + await nextTick() + await s.finishHandler() + + expect(s.onExternalEvent).toHaveBeenCalledTimes(1) + }) + + it('keeps going after a handler failed', async () => { + vi.spyOn(console, 'error').mockImplementation(() => undefined) + const s = setup() + s.onExternalEvent.mockRejectedValueOnce(new Error('boom')) + + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event({ etag: 'etag-2' })) + await flushPromises() + s.emit(MESSAGE_TYPE.POSTPROCESSING_FINISHED, event({ etag: 'etag-3' })) + await flushPromises() + + expect(s.onExternalEvent).toHaveBeenCalledTimes(2) + expect(console.error).toHaveBeenCalled() + }) +}) 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 5a3a20652ac..3caebec0b05 100644 --- a/packages/web-pkg/tests/unit/composables/yjs/useYjsSession.spec.ts +++ b/packages/web-pkg/tests/unit/composables/yjs/useYjsSession.spec.ts @@ -96,17 +96,20 @@ vi.mock('@hocuspocus/provider', async () => { // delay of 0 answers synchronously, which keeps every latency-agnostic // test on a single `flushPromises`. sendStateless = vi.fn((payload: string) => { - if (payload !== '_oc_seed_request') return const socketId = String(this.document.clientID) - const 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' - }) + 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' + }) + } } + if (!answer) return if (relay.docDelayMs === 0) { answer() return @@ -222,6 +225,7 @@ function setupSession({ if (!mirrorEtagToResource) return resourceRef.value = { ...unref(resourceRef), etag } as Resource }) + const onExternalUpdate = vi.fn() let session: YjsSession const wrapper = getComposableWrapper( @@ -237,7 +241,8 @@ function setupSession({ onServerContentChange, onEtagChange, hasUnsavedChanges: hasUnsavedChangesRef, - onConflict + onConflict, + onExternalUpdate }) }, { @@ -264,6 +269,7 @@ function setupSession({ onServerContentChange, onEtagChange, onConflict, + onExternalUpdate, get session() { return session }, @@ -2190,3 +2196,356 @@ describe('useYjsSession — unsolicited seed grant', () => { expect(s.ydoc!.getMap('_oc_meta').get('hydrated')).toBeUndefined() }) }) + +// An SSE event told the caller that the file changed on disk, it fetched the +// new body, and hands it in. What happens next depends on whether anything in +// the room is unsaved, and on who else is trying to do the same. +describe('useYjsSession — external updates', () => { + const URL = 'wss://example.test/yjs' + const META_KEY = '_oc_meta' + const EXTERNAL = { content: 'external body', etag: 'etag-external' } + + function text(s: ReturnType) { + return s.ydoc!.getText(SHARED_TEXT_KEY).toString() + } + + async function remoteSession(options: Parameters[0] = {}) { + const s = setupSession({ yjsServerUrl: URL, currentContent: 'seed', ...options }) + await flushPromises() + providerInstances.at(-1)!.triggerSynced() + await flushPromises() + return s + } + + /** Runs `applyExternalUpdate` through the 150 ms election wait. */ + async function apply(s: ReturnType, update = EXTERNAL) { + const result = s.session.applyExternalUpdate(update) + await vi.advanceTimersByTimeAsync(500) + await flushPromises() + return result + } + + beforeEach(() => { + vi.useFakeTimers({ shouldAdvanceTime: true }) + }) + + it('rewrites a clean doc in local mode and reports the result to the caller', async () => { + const s = setupSession({ currentContent: 'seed' }) + await flushPromises() + + const result = await apply(s) + + expect(result).toBe('recovered') + expect(text(s)).toBe('external body') + expect(s.onContentChange).toHaveBeenCalledWith('external body') + expect(s.onServerContentChange).toHaveBeenCalledWith('external body') + expect(s.onEtagChange).toHaveBeenCalledWith('etag-external') + expect(s.onExternalUpdate).toHaveBeenCalledTimes(1) + expect(s.onConflict).not.toHaveBeenCalled() + const meta = s.ydoc!.getMap(META_KEY) + expect(meta.get('etag')).toBe('etag-external') + expect(meta.get('isStale')).toBeUndefined() + }) + + // The rewrite suppresses the content reporter, so without an explicit + // report the caller would still hold the old body and read the recovered + // doc as unsaved work. The serialized form is reported, not the raw body: + // the next debounced report yields the same string, so nothing flips dirty. + it('reports the serialized form of the recovered body', async () => { + const s = setupSession({ currentContent: 'seed', adapter: normalizingAdapter }) + await flushPromises() + + await apply(s) + + expect(s.onContentChange).toHaveBeenLastCalledWith('external body\n') + expect(s.onServerContentChange).toHaveBeenLastCalledWith('external body\n') + }) + + it('recovers a clean remote room and stamps the given etag, not the resource etag', async () => { + const s = await remoteSession({ mirrorEtagToResource: true }) + const meta = s.ydoc!.getMap(META_KEY) + let saveStamps = 0 + meta.observe((event) => { + if (event.keysChanged.has('lastSavedAt')) saveStamps++ + }) + + const result = await apply(s) + + 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. + expect(saveStamps).toBe(1) + expect(unref(s.session.status)).not.toBe('disconnected') + }) + + it('flags the room and conflicts when this client holds unsaved work', async () => { + const s = await remoteSession({ hasUnsavedChanges: true }) + + const result = await apply(s) + + expect(result).toBe('conflicted') + const meta = s.ydoc!.getMap(META_KEY) + 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() + expect(unref(s.session.isConflicted)).toBe(true) + expect(providerInstances[0].disconnect).toHaveBeenCalled() + expect(providerInstances[0].detach).toHaveBeenCalled() + expect(text(s)).toBe('seed') + expect(s.onConflict).toHaveBeenCalledTimes(1) + expect(s.onExternalUpdate).not.toHaveBeenCalled() + }) + + it('conflicts without touching the doc in local mode when work is unsaved', async () => { + const s = setupSession({ currentContent: 'seed', hasUnsavedChanges: true }) + await flushPromises() + + const result = await apply(s) + + expect(result).toBe('conflicted') + expect(s.onConflict).toHaveBeenCalledTimes(1) + expect(unref(s.session.isConflicted)).toBe(true) + expect(unref(s.session.status)).toBe('local') + expect(text(s)).toBe('seed') + expect(s.ydoc!.getMap(META_KEY).get('isStale')).toBeUndefined() + }) + + // The caller's dirty state lags the doc by the serialize debounce. A + // keystroke from inside that window is unsaved work all the same. + it('flushes the pending content report before deciding', async () => { + const s = setupSession({ currentContent: 'seed' }) + await flushPromises() + s.onContentChange.mockImplementation(() => { + s.hasUnsavedChangesRef.value = true + }) + + s.ydoc!.getText(SHARED_TEXT_KEY).insert(0, 'typed ') + // Well inside SERIALIZE_DEBOUNCE_MS. + await vi.advanceTimersByTimeAsync(50) + const result = await apply(s) + + expect(result).toBe('conflicted') + expect(text(s)).toBe('typed seed') + }) + + it('skips when the room already carries the etag', async () => { + const s = await remoteSession() + s.ydoc!.getMap(META_KEY).set('etag', 'etag-external') + + expect(await apply(s)).toBe('skipped') + expect(text(s)).toBe('seed') + }) + + it('skips when a connected peer already claimed the recovery', 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') + + const result = await apply(s) + + expect(result).toBe('skipped') + expect(meta.get('recoveryClientId')).toBe(claimant) + expect(text(s)).toBe('seed') + }) + + it('takes over the recovery when the claimant has left', 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 result = await apply(s) + + expect(result).toBe('recovered') + expect(text(s)).toBe('external body') + expect(meta.get('isStale')).toBeUndefined() + expect(meta.get('etag')).toBe('etag-external') + }) + + it('skips for a read-only, conflicted or locked session', async () => { + const s = await remoteSession({ isReadOnly: true }) + + expect(await apply(s)).toBe('skipped') + expect(text(s)).toBe('seed') + }) + + describe('with peers', () => { + async function twoPeers({ bDirty = false } = {}) { + relay.enabled = true + relay.docDelayMs = 10 + relay.awarenessDelayMs = 10 + const a = setupSession({ yjsServerUrl: URL, currentContent: 'seed' }) + const b = setupSession({ + yjsServerUrl: URL, + currentContent: 'seed', + hasUnsavedChanges: bDirty + }) + await flushPromises() + providerInstances[0].triggerSynced() + providerInstances[1].triggerSynced() + await vi.advanceTimersByTimeAsync(1_000) + await flushPromises() + expect(text(a)).toBe('seed') + expect(text(b)).toBe('seed') + a.onContentChange.mockClear() + b.onContentChange.mockClear() + return { a, b } + } + + it('lets exactly one of two clean peers rewrite, and the other follows', async () => { + const { a, b } = await twoPeers() + + const [resultA, resultB] = await Promise.all([ + a.session.applyExternalUpdate(EXTERNAL), + b.session.applyExternalUpdate(EXTERNAL) + ]) + await vi.advanceTimersByTimeAsync(1_000) + await flushPromises() + + expect([resultA, resultB].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. + it('tells a follower about the update only once the rewrite landed', async () => { + const { a, b } = await twoPeers() + + 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(text(b)).toBe('external body') + expect(b.onExternalUpdate).toHaveBeenCalledTimes(1) + }) + + it('conflicts a dirty peer and leaves its doc intact', async () => { + const { a, b } = await twoPeers({ bDirty: true }) + + const result = await apply(a) + await vi.advanceTimersByTimeAsync(1_000) + + expect(result).toBe('recovered') + expect(text(a)).toBe('external body') + expect(unref(b.session.isConflicted)).toBe(true) + expect(b.onConflict).toHaveBeenCalledTimes(1) + expect(text(b)).toBe('seed') + }) + + // A dirty detector flags without claiming. A clean peer that holds a + // fresh body of its own (it got the same event) recovers from that. + it('lets a clean peer recover after a dirty detector flagged the room', async () => { + const { a, b } = await twoPeers() + a.hasUnsavedChangesRef.value = true + + const result = await apply(a) + await vi.advanceTimersByTimeAsync(1_000) + + expect(result).toBe('conflicted') + expect(b.onConflict).not.toHaveBeenCalled() + expect(unref(b.session.isConflicted)).toBe(false) + + const followUp = await apply(b) + await vi.advanceTimersByTimeAsync(1_000) + expect(followUp).toBe('recovered') + expect(text(b)).toBe('external body') + }) + + // The peer's own keystroke sits inside its debounce window when the flag + // arrives, so `hasUnsavedChanges` still says clean. Flushing first catches it. + it('conflicts a peer whose unsaved keystroke is still inside the debounce', async () => { + const { a, b } = await twoPeers() + b.onContentChange.mockImplementation(() => { + b.hasUnsavedChangesRef.value = true + }) + b.ydoc!.getText(SHARED_TEXT_KEY).insert(0, 'typed ') + // The keystroke reaches A and makes it dirty too. Reset that: this test + // is about B's own window, A stands in for a clean detector. + await vi.advanceTimersByTimeAsync(20) + a.hasUnsavedChangesRef.value = false + + const result = await apply(a) + await vi.advanceTimersByTimeAsync(1_000) + + expect(result).toBe('recovered') + expect(unref(b.session.isConflicted)).toBe(true) + expect(text(b)).toBe('typed seed') + }) + + it('adopts an etag without turning it into a save for peers', async () => { + const { a, b } = await twoPeers() + + a.session.adoptEtag('etag-renamed') + await vi.advanceTimersByTimeAsync(100) + + expect(a.onEtagChange).toHaveBeenCalledWith('etag-renamed') + expect(b.onEtagChange).toHaveBeenCalledWith('etag-renamed') + expect(a.ydoc!.getMap(META_KEY).get('lastSavedAt')).toBeUndefined() + expect(b.onServerContentChange).not.toHaveBeenCalled() + }) + + // The recovered doc is what is on disk, so the recovering peer's next + // 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([ + a.session.applyExternalUpdate(EXTERNAL), + b.session.applyExternalUpdate(EXTERNAL) + ]) + await vi.advanceTimersByTimeAsync(1_000) + b.onServerContentChange.mockClear() + + a.resourceRef.value = { ...unref(a.resourceRef), etag: 'etag-saved' } as Resource + await vi.advanceTimersByTimeAsync(100) + + expect(b.onEtagChange).toHaveBeenCalledWith('etag-saved') + expect(b.onServerContentChange).toHaveBeenCalledWith('external body') + }) + }) + + describe('isRoomWrite', () => { + it('is true for the etag the room stamped', async () => { + const s = await remoteSession() + s.ydoc!.getMap(META_KEY).set('etag', 'etag-room') + + expect(s.session.isRoomWrite('etag-room')).toBe(true) + expect(s.session.isRoomWrite('etag-other')).toBe(false) + }) + + it('is false in local mode', async () => { + const s = setupSession({ currentContent: 'seed' }) + await flushPromises() + s.ydoc!.getMap(META_KEY).set('etag', 'etag-room') + + expect(s.session.isRoomWrite('etag-room')).toBe(false) + }) + }) +}) diff --git a/packages/web-runtime/src/container/sse/types.ts b/packages/web-runtime/src/container/sse/types.ts index eb7e65a8336..cfc67bdc988 100644 --- a/packages/web-runtime/src/container/sse/types.ts +++ b/packages/web-runtime/src/container/sse/types.ts @@ -1,4 +1,4 @@ -import { z } from 'zod' +import { sseEventSchema, SseEventData } from '@opencloud-eu/web-client/sse' import { AuthStore, ClientService, @@ -14,17 +14,9 @@ import { Router } from 'vue-router' import { Language } from 'vue3-gettext' import PQueue from 'p-queue' -export const eventSchema = z.object({ - itemid: z.string().optional(), - parentitemid: z.string().optional(), - spaceid: z.string().optional(), - initiatorid: z.string().optional(), - etag: z.string().optional(), - affecteduserids: z.array(z.string()).optional().nullable(), - sessionid: z.string().optional() -}) +export const eventSchema = sseEventSchema -export type EventSchemaType = z.infer +export type EventSchemaType = SseEventData export interface SSEEventOptions { resourcesStore: ResourcesStore diff --git a/tests/e2e/features/yjs/conflicts.feature b/tests/e2e/features/yjs/conflicts.feature index 8fd1c4221e7..88ecb9a0323 100644 --- a/tests/e2e/features/yjs/conflicts.feature +++ b/tests/e2e/features/yjs/conflicts.feature @@ -15,7 +15,7 @@ Feature: yjs conflict handling | name | id | | Team | team | - Scenario: user cannot overwrite an external file update without warning + Scenario: editor with unsaved changes is told about an external file update right away When "Alice" uploads the following local file into personal space using API | localFile | to | | textfile.ocnote | textfile.ocnote | @@ -29,24 +29,56 @@ Feature: yjs conflict handling And "Alice" enters the text "Alice says hello" in editor "TextEditor" # external update - When "Alice" uploads the following local file into personal space using API - | localFile | to | - | textfile.ocnote | textfile.ocnote | - Then "Alice" should see the text "Alice says hello" in the text-editor - - And "Alice" saves the file viewer expecting conflict error - And "Alice" sees the current file as dirty + When "Alice" creates the following file into personal space using API + | pathToFile | content | + | textfile.ocnote | {"type":"doc","content":[{"type":"paragraph","content":[{"type":"text","text":"external update"}]}]} | + Then "Alice" should see an error message + """ + This file was updated outside this window. Please copy your changes, save the file under a new name (»Save As...«) or reload the page to discard your changes. + """ And "Alice" should see the following yjs status | status | | Disconnected | + And "Alice" should see the text "Alice says hello" in the text-editor + + And "Alice" saves the file viewer expecting conflict error + And "Alice" sees the current file as dirty And "Alice" reloads the page - Then "Alice" should see the text "some random text" in the text-editor + Then "Alice" should see the text "external update" in the text-editor And "Alice" should see the following yjs status | status | | Connected | And "Alice" logs out - Scenario: editor sees external file update when peer with edit permissions joins + # The same bytes under a new etag are not a content change. The editor + # records the etag and carries on, so the save goes through. + Scenario: re-upload of the same content does not interrupt the editor + When "Alice" uploads the following local file into personal space using API + | localFile | to | + | textfile.ocnote | textfile.ocnote | + + And "Alice" logs in + And "Alice" opens the "files" app + And "Alice" opens file "textfile.ocnote" via "text-editor" using the context menu + And "Alice" should see the following yjs status + | status | + | Connected | + And "Alice" enters the text "Alice says hello" in editor "TextEditor" + + # external re-upload of the identical file + When "Alice" uploads the following local file into personal space using API + | localFile | to | + | textfile.ocnote | textfile.ocnote | + And "Alice" enters the text "Alice keeps typing" in editor "TextEditor" + Then "Alice" should see the following yjs status + | status | + | Connected | + And "Alice" saves the file viewer + And "Alice" reloads the page + Then "Alice" should see the text "Alice keeps typing" in the text-editor + And "Alice" logs out + + Scenario: editor without unsaved changes shows an external file update right away When "Alice" uploads the following local file into space "Team" using API | localFile | to | | textfile.ocnote | textfile.ocnote | @@ -69,7 +101,14 @@ Feature: yjs conflict handling When "Alice" uploads the following local file into space "Team" using API | localFile | to | | textfile.ocnote | textfile.ocnote | - Then "Alice" should see the text "Alice says hello" in the text-editor + Then "Alice" should see the text "some random text" in the text-editor + And "Alice" should see a notification + """ + This file was updated outside this window. The editor now shows the latest version. + """ + And "Alice" should see the following yjs status + | status | + | Connected | When "Brian" logs in And "Brian" navigates to the project space "team" @@ -78,14 +117,10 @@ Feature: yjs conflict handling | status | | Connected | Then "Brian" should see the text "some random text" in the text-editor - And "Alice" should see the text "some random text" in the text-editor - And "Alice" should see the following yjs status - | status | - | Connected | And "Alice" logs out And "Brian" logs out - Scenario: editor sees external file update when peer with view permissions joins + Scenario: viewer joining after an external file update sees the current version When "Alice" uploads the following local file into space "Team" using API | localFile | to | | textfile.ocnote | textfile.ocnote | @@ -108,20 +143,16 @@ Feature: yjs conflict handling When "Alice" uploads the following local file into space "Team" using API | localFile | to | | textfile.ocnote | textfile.ocnote | - Then "Alice" should see the text "Alice says hello" in the text-editor + Then "Alice" should see the text "some random text" in the text-editor When "Brian" logs in And "Brian" navigates to the project space "team" And "Brian" opens file "textfile.ocnote" via "text-editor" using the context menu - # FIXME: uncomment when https://github.com/opencloud-eu/web/issues/3103 is resolved - #Then "Brian" should see the text "some random text" in the text-editor - Then "Brian" should see the text "Alice says hello" in the text-editor - #And "Alice" should see the text "some random text" in the text-editor - And "Alice" should see the text "Alice says hello" in the text-editor + Then "Brian" should see the text "some random text" in the text-editor And "Alice" logs out And "Brian" logs out - Scenario: external file update doesn't overwrite unsaved local changes when peer joins + Scenario: external file update doesn't overwrite unsaved local changes When "Alice" uploads the following local file into space "Team" using API | localFile | to | | textfile.ocnote | textfile.ocnote | @@ -145,8 +176,16 @@ Feature: yjs conflict handling When "Alice" uploads the following local file into space "Team" using API | localFile | to | | textfile.ocnote | textfile.ocnote | - Then "Alice" should see the text "Alice update #2" in the text-editor + Then "Alice" should see an error message + """ + This file was updated outside this window. Please copy your changes, save the file under a new name (»Save As...«) or reload the page to discard your changes. + """ + And "Alice" should see the following yjs status + | status | + | Disconnected | + And "Alice" should see the text "Alice update #2" in the text-editor + # a joiner holding the fresh file recovers the flagged room When "Brian" logs in And "Brian" navigates to the project space "team" And "Brian" opens file "textfile.ocnote" via "text-editor" using the context menu @@ -155,13 +194,43 @@ Feature: yjs conflict handling | status | | Connected | And "Alice" should see the text "Alice update #2" in the text-editor - And "Alice" should see an error message - """ - This file was updated outside this window. Please copy your changes, save the file under a new name (»Save As...«) or reload the page to discard your changes. - """ + And "Alice" logs out + And "Brian" logs out + + # A share recipient of a personal-space file receives no server-sent events + # for it, so the update only reaches the room once a peer with the fresh + # file joins. + Scenario: share recipient sees an external file update when the owner joins + When "Alice" uploads the following local file into personal space using API + | localFile | to | + | textfile.ocnote | textfile.ocnote | + And "Alice" shares the following resources using API + | resource | recipient | type | role | + | textfile.ocnote | Brian | user | Can edit | + + And "Brian" logs in + And "Brian" navigates to the shared with me page + And "Brian" opens file "textfile.ocnote" via "text-editor" using the context menu + And "Brian" should see the following yjs status + | status | + | Connected | + And "Brian" enters the text "Brian says hello" in editor "TextEditor" + And "Brian" sees the current file as dirty + And "Brian" saves the file viewer + + # external update + When "Alice" uploads the following local file into personal space using API + | localFile | to | + | textfile.ocnote | textfile.ocnote | + Then "Brian" should see the text "Brian says hello" in the text-editor + + When "Alice" logs in + And "Alice" opens the "files" app + And "Alice" opens file "textfile.ocnote" via "text-editor" using the context menu And "Alice" should see the following yjs status - | status | - | Disconnected | + | status | + | Connected | + Then "Alice" should see the text "some random text" in the text-editor + And "Brian" should see the text "some random text" in the text-editor And "Alice" logs out And "Brian" logs out - \ No newline at end of file diff --git a/tests/e2e/steps/ui/resources.ts b/tests/e2e/steps/ui/resources.ts index 6f1b025d24b..8aab065fa98 100644 --- a/tests/e2e/steps/ui/resources.ts +++ b/tests/e2e/steps/ui/resources.ts @@ -1520,6 +1520,14 @@ Then( } ) +Then( + '{string} should see a notification', + async ({ world }: { world: World }, stepUser: string, message: string): Promise => { + const { page } = world.actorsEnvironment.getActor({ key: stepUser }) + await expect(editor.notificationLocator(page, message)).toBeVisible() + } +) + Then( '{string} should see the following yjs status', async ({ world }: { world: World }, stepUser: string, stepTable: DataTable): Promise => { diff --git a/tests/e2e/support/objects/app-files/utils/editor.ts b/tests/e2e/support/objects/app-files/utils/editor.ts index 42c234c6764..d16124305b4 100644 --- a/tests/e2e/support/objects/app-files/utils/editor.ts +++ b/tests/e2e/support/objects/app-files/utils/editor.ts @@ -11,6 +11,7 @@ const saveConflictDialog = '.oc-modal' const filesListUrl = /.*\/files\/(spaces|shares|link|search)\/.*/ const errorNotification = '.oc-notification-message-danger' const yjsStatusIndicator = '[data-test-yjs-status]' +const notification = '.oc-notification-message:not(.oc-notification-message-danger)' const saveConflictDialogButtons: Record = { Save: '.oc-modal-body-actions-confirm', "Don't Save": '.oc-modal-body-actions-secondary', @@ -66,6 +67,9 @@ export const errorNotificationLocator = (page: Page, message?: string): Locator ? page.locator(errorNotification, { hasText: message }).first() : page.locator(errorNotification).first() +export const notificationLocator = (page: Page, message: string): Locator => + page.locator(notification, { hasText: message }).first() + export const yjsStatusIndicatorLocator = (page: Page): Locator => page.locator(yjsStatusIndicator) export const yjsStatusLocator = (page: Page, status: string): Locator =>