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 =>