Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 84 additions & 17 deletions dev/docs/yjs.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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?<br/>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<br/>body == room state -> record as save<br/>else wasWrittenByRoom with a 2 s grace
AW->>S: applyExternalUpdate {content, etag}
Note over S: flush pending content report,<br/>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:
Expand Down Expand Up @@ -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 |

---

Expand Down Expand Up @@ -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.
14 changes: 14 additions & 0 deletions packages/web-client/src/sse/index.ts
Original file line number Diff line number Diff line change
@@ -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<typeof sseEventSchema>

export enum MESSAGE_TYPE {
NOTIFICATION = 'userlog-notification',
Expand Down
11 changes: 10 additions & 1 deletion packages/web-pkg/src/components/AppTemplates/AppWrapper.vue
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
98 changes: 92 additions & 6 deletions packages/web-pkg/src/components/AppTemplates/useAppWrapperYjs.ts
Original file line number Diff line number Diff line change
@@ -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 {
Expand All @@ -31,6 +42,10 @@ export interface AppWrapperYjsOptions {
runSaveCallback: (content: unknown) => Promise<void>
/** 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<boolean>
/** The file changed outside this window and the editor now shows that version. */
onExternalUpdateApplied: () => void
}

/**
Expand All @@ -57,7 +72,9 @@ export function useAppWrapperYjs(options: AppWrapperYjsOptions) {
putFileContents,
applySavedResource,
runSaveCallback,
onConflict
onConflict,
isSaving,
onExternalUpdateApplied
} = options

const { $gettext } = useGettext()
Expand Down Expand Up @@ -99,7 +116,8 @@ export function useAppWrapperYjs(options: AppWrapperYjsOptions) {
resource.value = { ...unref(resource), etag: value }
},
hasUnsavedChanges: () => unref(isDirty),
onConflict
onConflict,
onExternalUpdate: onExternalUpdateApplied
})
: null

Expand Down Expand Up @@ -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<ReturnType<AppFileHandlingResult['getFileContents']>>
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 }
}
1 change: 1 addition & 0 deletions packages/web-pkg/src/composables/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down
1 change: 1 addition & 0 deletions packages/web-pkg/src/composables/sse/index.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
export * from './useExternalFileUpdates'
Loading
Loading