Skip to content

Commit fbf5cd8

Browse files
committed
fix(files): simplify replay and isolate invalid recovery
1 parent cf8d624 commit fbf5cd8

15 files changed

Lines changed: 477 additions & 223 deletions

File tree

‎apps/realtime/src/handlers/file-doc-store.test.ts‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -661,6 +661,75 @@ describe('FileDocStore', () => {
661661
recovered.destroy()
662662
})
663663

664+
it.each(['headless', 'attached'] as const)(
665+
'does not recount a replacement snapshot near the byte budget during %s replay',
666+
async (mode) => {
667+
const streamKey = `filedoc:stream:${NAME}`
668+
const source = new Y.Doc()
669+
source.getText('body').insert(0, 'x'.repeat(10 * 1024 * 1024))
670+
const initial = Buffer.from(Y.encodeStateAsUpdate(source)).toString('base64')
671+
const noop = Buffer.from(updateFor('')).toString('base64')
672+
state.backing!.streams.set(
673+
streamKey,
674+
Array.from({ length: 8 }, (_, index) => ({
675+
id: `${index + 1}-0`,
676+
message: { u: index === 0 ? initial : noop },
677+
}))
678+
)
679+
source.getText('body').insert(source.getText('body').length, ' joined')
680+
const compacted = Buffer.from(Y.encodeStateAsUpdate(source)).toString('base64')
681+
state.backing!.seq = 8
682+
state.backing!.onRange = (_call, key, start) => {
683+
if (key !== streamKey || start !== '(4-0') return
684+
state.backing!.streams.set(streamKey, [{ id: '9-0', message: { u: compacted, s: '1' } }])
685+
state.backing!.seq = 9
686+
state.backing!.onRange = undefined
687+
}
688+
const store = await newStore()
689+
const recovered = new Y.Doc()
690+
try {
691+
if (mode === 'attached') await store.attachRoom(NAME, recovered)
692+
else Y.applyUpdate(recovered, (await store.getStreamState(NAME))!)
693+
expect(recovered.getText('body').toString()).toBe(source.getText('body').toString())
694+
} finally {
695+
store.detachRoom(NAME)
696+
recovered.destroy()
697+
source.destroy()
698+
}
699+
}
700+
)
701+
702+
it('does not recount retained entries when compaction meets the exact entry budget', async () => {
703+
const streamKey = `filedoc:stream:${NAME}`
704+
const noop = Buffer.from(updateFor('')).toString('base64')
705+
const entries = Array.from({ length: 1_999 }, (_, index) => ({
706+
id: `${index + 1}-0`,
707+
message: { u: noop },
708+
}))
709+
state.backing!.streams.set(streamKey, entries)
710+
state.backing!.seq = 1_999
711+
state.backing!.onRange = (_call, key, start) => {
712+
if (key !== streamKey || start !== '(1996-0') return
713+
state.backing!.streams.set(streamKey, [
714+
...entries.slice(1_996),
715+
{
716+
id: '2000-0',
717+
message: { u: Buffer.from(updateFor('complete')).toString('base64'), s: '1' },
718+
},
719+
])
720+
state.backing!.seq = 2_000
721+
state.backing!.onRange = undefined
722+
}
723+
const store = await newStore()
724+
const recovered = new Y.Doc()
725+
try {
726+
Y.applyUpdate(recovered, (await store.getStreamState(NAME))!)
727+
expect(recovered.getText('body').toString()).toBe('complete')
728+
} finally {
729+
recovered.destroy()
730+
}
731+
})
732+
664733
it('reads the replacement snapshot when peer deltas cross the old replay tail', async () => {
665734
const streamKey = `filedoc:stream:${NAME}`
666735
const source = new Y.Doc()

‎apps/realtime/src/handlers/file-doc-store.ts‎

Lines changed: 8 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -73,12 +73,7 @@ const INVALIDATE_DOCUMENT_SCRIPT =
7373
const ADOPT_GENERATION_SCRIPT =
7474
"local generation = redis.call('get', KEYS[2]); if generation then return generation end; if redis.call('xlen', KEYS[1]) == 0 then return false end; redis.call('set', KEYS[2], ARGV[1], 'EX', ARGV[2]); return ARGV[1]"
7575

76-
/**
77-
* Append an ordinary update only while the live-document generation is valid. A durable replacement
78-
* that cannot be represented by the rich editor first sets the invalidation tombstone and then removes
79-
* the stream; keeping the guard and XADD in one script prevents an old room from recreating that stream.
80-
* Returns the new stream id, or `false` while invalidated.
81-
*/
76+
/** Atomically fence XADD so stale rooms cannot recreate a replaced or expired stream. Returns false when fenced. */
8277
const APPEND_UPDATE_SCRIPT =
8378
"local generation = redis.call('get', KEYS[2]) or ''; if generation ~= ARGV[4] or redis.call('exists', KEYS[1]) == 0 then return false end; if ARGV[3] ~= '' then return redis.call('xadd', KEYS[1], '*', ARGV[1], ARGV[2], ARGV[3], '1') else return redis.call('xadd', KEYS[1], '*', ARGV[1], ARGV[2]) end"
8479

@@ -403,13 +398,8 @@ export class FileDocStore {
403398
}
404399

405400
/**
406-
* PULL the shared state into a registered room: read the stream and apply every entry the doc has
407-
* not integrated yet (origin {@link REDIS_ORIGIN}), advancing `lastId` so the tailer resumes exactly
408-
* after it. This is the ONLY way a room loads shared state, so a caller that must not depend on the
409-
* tailer's asynchronous push — the join, which may not serve a client a half-assembled document —
410-
* can converge on demand. Idempotent and safe to call repeatedly; no-op when disabled or the room is
411-
* not registered (a fast open→close detached it). Throws without applying partial state when replay
412-
* cannot complete, so a join never serves a prefix of the shared document.
401+
* Completes shared replay before applying entries, so joins never receive a partial document.
402+
* Repeated calls skip integrated entries; detached rooms and disabled stores are ignored.
413403
*/
414404
async catchUp(name: string): Promise<void> {
415405
if (!this.enabled) return
@@ -543,10 +533,8 @@ export class FileDocStore {
543533
}
544534

545535
/**
546-
* Append a user update exactly once within the stream's bounded deduplication window and return only
547-
* after Redis has accepted it. The client keeps the batch in IndexedDB until this promise succeeds
548-
* and its socket acknowledgement arrives, so a relay restart or lost acknowledgement is safe to
549-
* retry throughout that window.
536+
* Waits for Redis acceptance before the relay acknowledges the client. Retries are deduplicated
537+
* within the bounded window; clients retain their journal until the acknowledgement arrives.
550538
*/
551539
async publishClientUpdateAndWait(
552540
name: string,
@@ -656,10 +644,8 @@ export class FileDocStore {
656644
}
657645

658646
/**
659-
* Invalidate the current live-document generation after a durable replacement that the rich editor
660-
* cannot represent. The invalid generation marker is written before deleting the stream, so stale
661-
* room publishers cannot recreate it; the next authoritative seed atomically replaces the marker
662-
* with its generation. The same TTL as the stream bounds abandoned markers.
647+
* Fences an unsupported durable replacement before deleting its stream. The next authoritative
648+
* seed replaces the tombstone; abandoned tombstones expire with the stream TTL.
663649
*/
664650
async invalidateDocument(name: string, version: number): Promise<boolean> {
665651
if (!this.enabled) {
@@ -830,12 +816,11 @@ export class FileDocStore {
830816
}
831817
}
832818

833-
/** A moving head means compaction may have removed unread dependencies. Replay its snapshot. */
819+
/** Compaction appends its snapshot before trimming; extend the barrier without rereading it. */
834820
const currentFirstId = (await this.write.xRange(key, '-', '+', { COUNT: 1 }))[0]?.id
835821
if (!currentFirstId) throw new FileDocInvalidatedError()
836822
if (currentFirstId === firstId) return entriesRead
837823
firstId = currentFirstId
838-
cursor = '0-0'
839824
const currentTail = await this.write.xRevRange(key, '+', '-', { COUNT: 1 })
840825
if (currentTail.length === 0) throw new FileDocInvalidatedError()
841826
endId = currentTail[0].id

‎apps/realtime/src/handlers/file-doc.ts‎

Lines changed: 9 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -268,9 +268,8 @@ const AGENT_SYNC_ORIGIN = Symbol('file-doc-agent-sync')
268268
const MAX_LEGACY_FRAME_BYTES = FILE_DOC_LIMITS.updateBytes + 64
269269

270270
/**
271-
* Preflight the update-bearing inner Yjs message before `readSyncMessage` can mutate the room. Legacy
272-
* clients use the unacknowledged sync channel, so the relay itself must ensure any applied update also
273-
* fits the durable Redis stream; checking only the outer frame leaves a small framing-sized gap.
271+
* Checks the inner update before readSyncMessage mutates the room: the legacy outer-frame limit
272+
* includes framing headroom, which must not allow an update too large for the shared stream.
274273
*/
275274
function hasOversizedLegacyUpdate(bytes: Uint8Array): boolean {
276275
const decoder = decoding.createDecoder(bytes)
@@ -441,15 +440,11 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr
441440
await store.setSyncedVersion(name, result.version, generation)
442441
return
443442
}
444-
// status === 'conflict': the durable file advanced out-of-band since our If-Match token. We do NOT
445-
// re-persist against the current stream: an external write commits durable BEFORE its chokepoint merge
446-
// (`applyEditToLiveFileDoc`) reaches the stream, so a re-persist landing in that window would CAS-pass
447-
// with a stream that still lacks the external content and clobber the committed write. Instead leave the
448-
// durable content authoritative — the chokepoint merges the change into the stream and, ONLY once it is
449-
// actually there, advances the synced version (via the merge's own `recordVersion`); a later flush
450-
// (a subsequent debounced persist, or the final flush) then projects the converged stream with a token
451-
// that matches. The session's edits stay in the stream meanwhile. Deliberately do NOT advance the synced
452-
// version here: before the stream reflects the durable content, that would let the next flush clobber it.
443+
/**
444+
* External writes commit before merging into the stream. Retrying or advancing the synced
445+
* version here could overwrite content not yet merged; leave the durable file authoritative
446+
* until the merge advances the version, then let a later flush persist the converged state.
447+
*/
453448
logger.warn(
454449
`Persist conflict for file ${room.fileId}; durable content advanced out-of-band, left authoritative`
455450
)
@@ -579,10 +574,8 @@ export async function flushAllFileDocRooms(): Promise<void> {
579574
}
580575

581576
/**
582-
* Bring a room's document to its AUTHORITATIVE state — reflecting the file's shared stream and
583-
* carrying its seed — so the join can attach a client to a document that is already whole. Rejects
584-
* when hydration or seeding cannot complete; serving an unseeded or partial room would make a client
585-
* appear editable before the authoritative document exists.
577+
* Waits for shared hydration and authoritative seeding before joining; failures must not expose
578+
* an editable partial document.
586579
*/
587580
async function ensureRoomReady(
588581
name: string,

0 commit comments

Comments
 (0)