diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index e9a320e..f6f5663 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -4,7 +4,6 @@ on: push: branches: [main] pull_request: - branches: [main] jobs: test: diff --git a/README.md b/README.md index bc978f1..9673d75 100644 --- a/README.md +++ b/README.md @@ -229,7 +229,8 @@ YSR_TEST_REDIS=redis://127.0.0.1:36379 npm run test:consistency ``` The tests use a unique key prefix and remove only their own keys. They include -three server instances sharing Redis and a real Socket.IO reconnect. +three server instances sharing Redis, a real Socket.IO reconnect, delayed and +failed saves, worker replacement, and a real persistence worker thread. A sync-update ACK confirms Redis stream acceptance, not database persistence. Retries are appended again and safely merged by Yjs. Persistence leases use a @@ -242,6 +243,23 @@ lease expiry can still finish after a successor starts persisting. Replacement storage adapters need storage-enforced commit ordering to prevent stale overwrites; that contract is outside this change and the storage API is unchanged. +Server-side saves reload a document and its Redis stream position together, +adding a Redis and storage read per save. The saved document is independent of +the subscription cache. Stream trimming is bounded by that snapshot's position +and the minimum message lifetime. `trimRoomStream(room, docid, persistedId)` +requires a known committed position; omitting it retains the stream. Queue +workers use the same bound, including when a task is reclaimed. + +Persistence worker requests and success/error replies carry a `requestId`. +Custom worker implementations must echo it with the room; use the matching +package version in the server and worker. Failed saves retry after the normal +persistence delay. Worker errors, exits, and health-check replacement reject +pending requests so later saves and namespace cleanup can proceed. + +These guarantees assume that storage resolves `persistDoc` only after a durable +save and rejects on failure. Redis stream expiration and eviction are separate +from trimming; configure retention and Redis durability for your deployment. + ## License [The MIT License](./LICENSE) © Kevin Jahns diff --git a/src/api.js b/src/api.js index d6e783a..ec3cb95 100644 --- a/src/api.js +++ b/src/api.js @@ -7,7 +7,6 @@ import * as array from 'lib0/array' import * as random from 'lib0/random' import * as number from 'lib0/number' import * as promise from 'lib0/promise' -import * as math from 'lib0/math' import * as protocol from './protocol.js' import * as env from 'lib0/environment' import * as logging from 'lib0/logging' @@ -254,8 +253,7 @@ export class Api { } const now = performance.now() if (docstate) { Y.applyUpdateV2(ydoc, docstate.doc) } - let changed = false - ydoc.once('afterTransaction', (tr) => { changed = tr.changed.size > 0 }) + const previousState = Y.encodeStateAsUpdateV2(ydoc) ydoc.transact(() => { docMessages?.messages.forEach(m => { const decoder = decoding.createDecoder(m) @@ -281,7 +279,9 @@ export class Api { awareness, redisLastId: docMessages?.lastId.toString() || '0', storeReferences: docstate?.references || null, - changed + // Pending structs and delete sets also need persistence, even when no + // shared type changed during the transaction. + changed: !Buffer.from(previousState).equals(Y.encodeStateAsUpdateV2(ydoc)) } } @@ -295,15 +295,30 @@ export class Api { return docMessages?.lastId.toString() || '0' } + /** + * @param {string=} persistedId + */ + getTrimBoundary (persistedId) { + if (!persistedId || persistedId === '0') return null + const [ms, sequence = '0'] = persistedId.split('-') + const snapshotBoundary = BigInt(sequence) === (1n << 64n) - 1n + ? `${BigInt(ms) + 1n}-0` + : `${ms}-${BigInt(sequence) + 1n}` + const retentionBoundary = `${Math.max(0, Date.now() - this.redisMinMessageLifetime)}-0` + return isSmallerRedisId(snapshotBoundary, retentionBoundary) ? snapshotBoundary : retentionBoundary + } + /** * @param {string} room * @param {string} docid + * @param {string=} persistedId Last stream entry included in the committed snapshot. */ - async trimRoomStream (room, docid) { + async trimRoomStream (room, docid, persistedId) { + const boundary = this.getTrimBoundary(persistedId) + if (!boundary) return const roomName = computeRedisRoomStreamName(room, docid, this.prefix) - const redisLastId = await this.getRedisLastId(room, docid) - const lastId = number.parseInt(redisLastId.split('-')[0]) - await this.redis.xTrim(roomName, 'MINID', lastId - this.redisMinMessageLifetime) + // node-redis types XTRIM thresholds as numbers, but MINID accepts full IDs. + await this.redis.sendCommand(['XTRIM', roomName, 'MINID', boundary]) } /** @@ -349,41 +364,45 @@ export class Api { } else { reclaimCounts++ const { room, docid } = decodeRedisRoomStreamName(task.stream, this.prefix) - const { ydoc, storeReferences, redisLastId, changed } = await this.getDoc(room, docid) - const lastId = math.max(number.parseInt(redisLastId.split('-')[0]), number.parseInt(task.id.split('-')[0])) - if (changed) { - logWorker(`persisting changes in room: ${room}`) - await this.store.persistDoc(room, docid, ydoc) - } else logWorker(`skip persisting room: ${room} due to no changes`) - await promise.all([ - storeReferences && changed ? this.store.deleteReferences(room, docid, storeReferences) : promise.resolve(), - this.redis.multi() - .xTrim(task.stream, 'MINID', lastId - this.redisMinMessageLifetime) - .xAdd(this.redisWorkerStreamName, '*', { compact: task.stream }) - .xAck(this.redisWorkerStreamName, this.redisWorkerGroupName, task.id) - .xDel(this.redisWorkerStreamName, task.id) - .sAdd(this.workerSetName, task.stream) - .exec() - ]) - logWorker('Compacted stream ', { stream: task.stream, taskId: task.id, newLastId: lastId - this.redisMinMessageLifetime }) + const { ydoc, awareness, storeReferences, redisLastId, changed } = await this.getDoc(room, docid) try { - if (ydocUpdateCallback != null) { - // call YDOC_UPDATE_CALLBACK here - const formData = new FormData() - // @todo only convert ydoc to updatev2 once - // @ts-ignore - formData.append('ydoc', new Blob([Y.encodeStateAsUpdateV2(ydoc)])) - // @todo should add a timeout to fetch (see fetch signal abortcontroller) - const res = await fetch(new URL(room, ydocUpdateCallback), { body: formData, method: 'PUT' }) - if (!res.ok) { - console.error(`Issue sending data to YDOC_UPDATE_CALLBACK. status="${res.status}" statusText="${res.statusText}"`) + if (changed) { + logWorker(`persisting changes in room: ${room}`) + await this.store.persistDoc(room, docid, ydoc) + } else logWorker(`skip persisting room: ${room} due to no changes`) + const boundary = this.getTrimBoundary(redisLastId) + const transaction = this.redis.multi() + if (boundary) transaction.addCommand(['XTRIM', task.stream, 'MINID', boundary]) + await promise.all([ + storeReferences && changed ? this.store.deleteReferences(room, docid, storeReferences) : promise.resolve(), + transaction + .xAdd(this.redisWorkerStreamName, '*', { compact: task.stream }) + .xAck(this.redisWorkerStreamName, this.redisWorkerGroupName, task.id) + .xDel(this.redisWorkerStreamName, task.id) + .sAdd(this.workerSetName, task.stream) + .exec() + ]) + logWorker('Compacted stream ', { stream: task.stream, taskId: task.id, boundary }) + try { + if (ydocUpdateCallback != null) { + // call YDOC_UPDATE_CALLBACK here + const formData = new FormData() + // @todo only convert ydoc to updatev2 once + // @ts-ignore + formData.append('ydoc', new Blob([Y.encodeStateAsUpdateV2(ydoc)])) + // @todo should add a timeout to fetch (see fetch signal abortcontroller) + const res = await fetch(new URL(room, ydocUpdateCallback), { body: formData, method: 'PUT' }) + if (!res.ok) { + console.error(`Issue sending data to YDOC_UPDATE_CALLBACK. status="${res.status}" statusText="${res.statusText}"`) + } } + } catch (e) { + console.error(e) } - } catch (e) { - console.error(e) + } finally { + awareness?.destroy() + ydoc.destroy() } - // destroy ydoc after persisting - ydoc.destroy() } })) return { tasks, reclaimCounts } diff --git a/src/persist-worker-thread.js b/src/persist-worker-thread.js index 6cb3af0..8e6fe6d 100644 --- a/src/persist-worker-thread.js +++ b/src/persist-worker-thread.js @@ -21,21 +21,27 @@ export class PersistWorkerThread { parentPort?.postMessage({ event: 'ready' }) parentPort?.on('message', ({ event, ...rest }) => { if (event === 'ping') parentPort?.postMessage({ event: 'pong' }) - else this.persist(rest) + else if (event === undefined || event === 'persist') this.persist(rest) }) } /** - * @param {{ room: string, docstate: SharedArrayBuffer }} props + * @param {{ room: string, requestId: string, docstate: Uint8Array }} props */ - persist = async ({ room, docstate }) => { + persist = async ({ room, requestId, docstate }) => { this.log(`persisting ${room} in worker`) const state = new Uint8Array(docstate) const doc = new Y.Doc() - Y.applyUpdateV2(doc, state) - await this.store?.persistDoc(room, 'index', doc) - doc.destroy() - parentPort?.postMessage({ event: 'persisted', room }) + try { + Y.applyUpdateV2(doc, state) + if (!this.store) throw new Error('Persistence storage is unavailable') + await this.store.persistDoc(room, 'index', doc) + parentPort?.postMessage({ event: 'persisted', room, requestId }) + } catch (error) { + parentPort?.postMessage({ event: 'persist-error', room, requestId, error: String(error) }) + } finally { + doc.destroy() + } } } diff --git a/src/y-socket-io/y-socket-io.js b/src/y-socket-io/y-socket-io.js index bf7bcee..890b18f 100644 --- a/src/y-socket-io/y-socket-io.js +++ b/src/y-socket-io/y-socket-io.js @@ -145,6 +145,11 @@ export class YSocketIO { * @readonly */ awaitingPersistMap = new Map() + /** + * @type {Map void, reject: (error: Error) => void }>} + * @private + */ + workerPersistRequests = new Map() /** * @type {Map} * @private @@ -501,8 +506,8 @@ export class YSocketIO { let changed = false const existDoc = this.namespaceDocMap.get(namespace) if (existDoc) { - existDoc.ydoc.on('afterTransaction', (tr) => { - changed = tr.changed.size > 0 + existDoc.ydoc.once('afterTransaction', (tr) => { + changed = tr.changed.size > 0 || tr.doc.store.pendingStructs !== null || tr.doc.store.pendingDs !== null }) Y.transact(existDoc.ydoc, () => { for (const msg of updates) { @@ -533,7 +538,8 @@ export class YSocketIO { this.namespacePersistentMap.set(namespace, now) this.debouncedPersist(namespace, nsp.sockets.size === 0) } - this.namespaceDocMap.get(namespace)?.ydoc.destroy() + const previousDoc = this.namespaceDocMap.get(namespace) + if (previousDoc !== doc) previousDoc?.ydoc.destroy() this.namespaceDocMap.set(namespace, doc) } @@ -542,6 +548,7 @@ export class YSocketIO { * @param {boolean=} immediate */ debouncedPersist (namespace, immediate = false) { + if (this.destroyed) return if (this.debouncedPersistMap.has(namespace)) { if (!immediate) return clearTimeout(this.debouncedPersistMap.get(namespace) || undefined) @@ -551,47 +558,23 @@ export class YSocketIO { : PERSIST_INTERVAL + (Math.random() - 0.5) * PERSIST_INTERVAL const timeout = setTimeout( async () => { - // wait for previous persisting operation if exists const prev = this.awaitingPersistMap.get(namespace) - if (prev?.promise) await prev.promise - // delete persist entry to allow queueing next persisting operation - // we can delete it here because the following until awaiting persist - // are all synchronize operations + if (prev) await prev.promise.catch(() => {}) + if (this.destroyed || this.debouncedPersistMap.get(namespace) !== timeout) return this.debouncedPersistMap.delete(namespace) - - const isLeader = await this.tryAcquirePersistentLeader(namespace) - if (!isLeader) return - + const operation = this.persistNamespace(namespace) + this.awaitingPersistMap.set(namespace, { promise: operation, resolve: () => {} }) try { - assert(this.client) - const doc = this.namespaceDocMap.get(namespace)?.ydoc - logSocketIO(`trying to persist ${namespace} in [SID: ${this.serverId}]`) - if (!doc) return - if (this.persistWorker && this.workerReady) { - /** @type {ReturnType>} */ - const { promise, resolve } = promiseWithResolvers() - this.awaitingPersistMap.set(namespace, { promise, resolve }) - - const docState = Y.encodeStateAsUpdateV2(doc) - const buf = new Uint8Array(new SharedArrayBuffer(docState.length)) - buf.set(docState) - this.persistWorker.postMessage({ - room: namespace, - docstate: buf - }) - await promise - } else { - const promise = this.client.store.persistDoc(namespace, 'index', doc) - this.awaitingPersistMap.set(namespace, { promise, resolve: () => {} }) - await promise - } - - await this.client.trimRoomStream(namespace, 'index') + await operation } catch (e) { - // suppress redis client closed error if (!(e instanceof ClientClosedError)) { console.error(e) } + if (this.namespaceDocMap.has(namespace)) this.debouncedPersist(namespace) + } finally { + if (this.awaitingPersistMap.get(namespace)?.promise === operation) { + this.awaitingPersistMap.delete(namespace) + } } }, timeoutInterval @@ -600,6 +583,51 @@ export class YSocketIO { this.debouncedPersistMap.set(namespace, timeout) } + /** @param {string} namespace */ + async persistNamespace (namespace) { + const isLeader = await this.tryAcquirePersistentLeader(namespace) + if (!isLeader || this.destroyed) return + assert(this.client) + if (!this.namespaceDocMap.has(namespace)) return + // Read the stream prefix and document together. The subscription cache may + // lag behind Redis, so its contents cannot authorize trimming the head. + const snapshot = await this.client.getDoc(namespace, 'index') + logSocketIO(`trying to persist ${namespace} in [SID: ${this.serverId}]`) + try { + if (this.destroyed) return + if (this.persistWorker && this.workerReady) { + await this.persistInWorker(namespace, Y.encodeStateAsUpdateV2(snapshot.ydoc)) + } else { + await this.client.store.persistDoc(namespace, 'index', snapshot.ydoc) + } + await this.client.trimRoomStream(namespace, 'index', snapshot.redisLastId) + } finally { + snapshot.awareness?.destroy() + snapshot.ydoc.destroy() + } + } + + /** + * @param {string} room + * @param {Uint8Array} state + */ + async persistInWorker (room, state) { + const worker = this.persistWorker + assert(worker) + const requestId = randomUUID() + /** @type {ReturnType>} */ + const pending = promiseWithResolvers() + this.workerPersistRequests.set(requestId, { room, worker, resolve: pending.resolve, reject: pending.reject }) + try { + const docstate = new Uint8Array(new SharedArrayBuffer(state.length)) + docstate.set(state) + worker.postMessage({ room, requestId, docstate }) + await pending.promise + } finally { + this.workerPersistRequests.delete(requestId) + } + } + /** * @param {Namespace} namespace */ @@ -677,8 +705,17 @@ export class YSocketIO { } destroy () { + if (this.destroyed) return this.destroyed = true this.pendingLeaderAcquisitions.clear() + for (const timeout of this.debouncedPersistMap.values()) clearTimeout(timeout || undefined) + this.debouncedPersistMap.clear() + for (const timeout of this.awaitingCleanupNamespace.values()) clearTimeout(timeout) + this.awaitingCleanupNamespace.clear() + if (this.persistWorker) { + this.failPersistWorker(this.persistWorker, new Error('Server destroyed')) + this.persistWorker.terminate().catch(console.error) + } try { if (this.persistWorkerHealthCheckTimeout) { clearInterval(this.persistWorkerHealthCheckTimeout) @@ -694,14 +731,37 @@ export class YSocketIO { } registerPersistWorkerResolve () { - if (!this.persistWorker) return - this.persistWorker.on('message', ({ event, room }) => { - if (event === 'persisted') this.awaitingPersistMap.get(room)?.resolve() + const worker = this.persistWorker + if (!worker) return + worker.on('message', ({ event, room, requestId, error }) => { + if (worker !== this.persistWorker || this.destroyed) return + const request = this.workerPersistRequests.get(requestId) + if (request?.worker === worker && request.room === room) { + if (event === 'persisted') request.resolve() + if (event === 'persist-error') request.reject(new Error(error || 'Worker persistence failed')) + } if (event === 'pong' && this.workerHeartbeatContext) { this.workerHeartbeatContext.resolve(true) } - this.workerReady = true + if (event === 'ready' || event === 'pong') this.workerReady = true }) + worker.on('error', error => this.failPersistWorker(worker, error)) + worker.on('exit', code => this.failPersistWorker(worker, new Error(`Persistence worker exited (${code})`))) + } + + /** + * @param {import('worker_threads').Worker} worker + * @param {Error} error + */ + failPersistWorker (worker, error) { + if (worker === this.persistWorker) { + this.workerReady = false + this.workerLastHeartbeat = null + this.workerHeartbeatContext?.resolve(false) + } + for (const request of this.workerPersistRequests.values()) { + if (request.worker === worker) request.reject(error) + } } /** @param {string} namespace */ @@ -730,7 +790,14 @@ export class YSocketIO { const timer = setTimeout(async () => { const awaitingPersist = this.awaitingPersistMap.get(namespace) - if (awaitingPersist) await awaitingPersist.promise + try { + if (awaitingPersist) await awaitingPersist.promise + } catch (error) { + if (this.awaitingCleanupNamespace.get(namespace) === timer) { + this.cleanupNamespace(namespace, stream, DEFAULT_CLEAR_TIMEOUT, true) + } + return + } // A reconnect or a newer cleanup timer may have superseded this callback // while it was waiting for persistence. if (this.awaitingCleanupNamespace.get(namespace) !== timer) return @@ -738,6 +805,10 @@ export class YSocketIO { this.awaitingCleanupNamespace.delete(namespace) return } + if (this.debouncedPersistMap.has(namespace)) { + this.cleanupNamespace(namespace, stream, DEFAULT_CLEAR_TIMEOUT, true) + return + } this.cleanupNamespaceImpl(namespace, stream) this.awaitingCleanupNamespace.delete(namespace) logSocketIO(`no active connection, namespace: ${namespace} cleared`) @@ -782,36 +853,33 @@ export class YSocketIO { `, { keys: [this.getLeaderKeyOf(namespace)], arguments: [token] }) } - async waitUntilWorkerReady () { - if (!this.persistWorker || this.workerReady) return - /** @type {ReturnType>} */ - const { promise, resolve } = promiseWithResolvers() - const timer = setInterval(() => { - if (!this.workerReady) return - clearInterval(timer) - resolve() - }, 100) - await promise - } - registerPersistWorkerHealthCheck () { + if (this.destroyed) return this.persistWorkerHealthCheckTimeout = setTimeout(async () => { - const workerHealth = await this.workerHealthCheck() - if (!workerHealth) { - logSocketIO('worker thread is unhealthy, recreating') - assert(this.getPersistWorker) - this.workerReady = false - await this.persistWorker?.removeAllListeners().terminate() - this.persistWorker = this.getPersistWorker() - this.registerPersistWorkerResolve() - await this.waitUntilWorkerReady() + try { + const workerHealth = await this.workerHealthCheck() + if (!workerHealth && !this.destroyed) { + logSocketIO('worker thread is unhealthy, recreating') + assert(this.getPersistWorker) + if (this.persistWorker) { + this.failPersistWorker(this.persistWorker, new Error('Persistence worker is unresponsive')) + await this.persistWorker.terminate() + } + if (this.destroyed) return + this.persistWorker = this.getPersistWorker() + this.registerPersistWorkerResolve() + } + } catch (error) { + console.error(error) + } finally { + this.registerPersistWorkerHealthCheck() } - this.registerPersistWorkerHealthCheck() }, WORKER_HEALTH_CHECK_INTERVAL) } async workerHealthCheck () { - if (!this.persistWorker || this.workerHeartbeatContext) return null + if (!this.persistWorker) return false + if (this.workerHeartbeatContext) return this.workerHeartbeatContext.promise if ( this.workerLastHeartbeat && Date.now() - this.workerLastHeartbeat < WORKER_HEALTH_CHECK_INTERVAL * 2 @@ -821,14 +889,20 @@ export class YSocketIO { /** @type {ReturnType>} */ const { promise: heartbeatPromise, resolve } = promiseWithResolvers() - this.workerHeartbeatContext = { promise: heartbeatPromise, resolve } + const context = { promise: heartbeatPromise, resolve } + this.workerHeartbeatContext = context const now = performance.now() - this.persistWorker.postMessage({ event: 'ping' }) - const health = await Promise.race([ - heartbeatPromise, - promise.wait(3000).then(() => false) - ]) - this.workerHeartbeatContext = null + const timer = setTimeout(() => resolve(false), 3000) + let health = false + try { + this.persistWorker.postMessage({ event: 'ping' }) + health = await heartbeatPromise + } catch (error) { + health = false + } finally { + clearTimeout(timer) + if (this.workerHeartbeatContext === context) this.workerHeartbeatContext = null + } if (health) { logSocketIO(`worker health check: responded in ${performance.now() - now}ms`) this.workerLastHeartbeat = Date.now() diff --git a/tests/consistency.test.js b/tests/consistency.test.js index b5fb854..221b97f 100644 --- a/tests/consistency.test.js +++ b/tests/consistency.test.js @@ -5,12 +5,19 @@ import { createServer as createHttpServer } from 'node:http' import { setTimeout as delay } from 'node:timers/promises' import { Server } from 'socket.io' import { io as connect } from 'socket.io-client' +import { EventEmitter } from 'node:events' +import { Worker } from 'node:worker_threads' import * as Y from 'yjs' -import { YSocketIO } from '../src/y-socket-io/y-socket-io.js' -import { createApiClient } from '../src/api.js' import { createMemoryStorage } from '../src/storage/memory.js' import { promiseWithResolvers } from '../src/y-socket-io/utils.js' +process.env.Y_WORKER_DISABLED = 'true' +process.env.Y_SOCKET_IO_SERVER_PERSIST_INTERVAL = '20' +process.env.Y_SOCKET_IO_WORKER_HEALTH_CHECK_INTERVAL = '20' +/** @typedef {import('../src/y-socket-io/y-socket-io.js').YSocketIO} SocketIOServer */ +const { YSocketIO } = await import('../src/y-socket-io/y-socket-io.js') +const { createApiClient, computeRedisRoomStreamName } = await import('../src/api.js') + /** @param {() => boolean | Promise} predicate */ async function until (predicate) { const deadline = Date.now() + 3000 @@ -125,7 +132,7 @@ const prefix = `consistency-${randomUUID()}` const store = createMemoryStorage() /** @type {import('../src/api.js').Api[]} */ const clients = [] -/** @type {YSocketIO[]} */ +/** @type {SocketIOServer[]} */ const servers = [] before(async () => { @@ -142,20 +149,24 @@ after(async () => { for (const server of servers) { for (const timer of Reflect.get(server, 'awaitingCleanupNamespace').values()) clearTimeout(timer) for (const timer of Reflect.get(server, 'debouncedPersistMap').values()) clearTimeout(timer) + await server.destroy() } - if (clients[0]) { - const keys = await clients[0].redis.keys(`${prefix}:*`) - if (keys.length) await clients[0].redis.del(keys) + const cleanup = await createApiClient(store, { redisUrl, redisPrefix: prefix, enableAwareness: false }) + try { + const keys = await cleanup.redis.keys(`${prefix}:*`) + if (keys.length) await cleanup.redis.del(keys) + } finally { + await cleanup.destroy() } await Promise.all(clients.map(client => client.destroy())) }) -/** @param {YSocketIO} server @param {string} room */ +/** @param {SocketIOServer} server @param {string} room */ function activate (server, room) { Reflect.get(server, 'namespaceMap').set(room, Reflect.get(server, 'io').of(`/yjs|${room}`)) } -/** @param {YSocketIO} server @param {string} room */ +/** @param {SocketIOServer} server @param {string} room */ function cleanup (server, room) { Reflect.get(server, 'cleanupNamespaceImpl').call(server, room, `${prefix}:room:${room}:index`) } @@ -408,3 +419,317 @@ test('a superseded cleanup callback cannot release the lease while a newer timer a.cancelNamespaceCleanup(room) Reflect.get(a, 'awaitingPersistMap').delete(room) }) + +/** @param {import('node:test').TestContext} t */ +async function persistenceCase (t) { + const storage = createMemoryStorage() + const client = await createApiClient(storage, { redisUrl, redisPrefix: `${prefix}:${randomUUID()}`, enableAwareness: false }) + client.redisMinMessageLifetime = 0 + const server = createServer() + Reflect.set(server, 'client', client) + const room = 'snapshot' + const stream = computeRedisRoomStreamName(room, 'index', client.prefix) + activate(server, room) + Reflect.get(server, 'streamNamespaceMap').set(stream, room) + clients.push(client) + servers.push(server) + t.after(async () => { + await server.destroy() + for (const doc of Reflect.get(server, 'namespaceDocMap').values()) doc.ydoc.destroy() + }) + return { + server, + client, + storage, + room, + stream, + /** @param {string} id @param {Uint8Array} update */ + append: (id, update) => client.redis.xAdd(stream, id, { m: Buffer.from(server.toRedis('sync-update', update)) }), + load: async () => { + const doc = await client.getDoc(room, 'index') + Reflect.get(server, 'namespaceDocMap').set(room, doc) + return doc + } + } +} + +test('a delayed direct save keeps its snapshot and retains later updates from the same millisecond', async t => { + const { server, client, storage, room, stream, append, load } = await persistenceCase(t) + const { first, second } = createUpdates() + await append('1000-0', first) + await load() + const started = promiseWithResolvers() + const release = promiseWithResolvers() + const persist = storage.persistDoc.bind(storage) + storage.persistDoc = async (...args) => { + started.resolve(undefined) + await release.promise + await persist(...args) + } + const operation = server.persistNamespace(room) + await started.promise + try { + await append('1000-1', second) + await Reflect.get(server, 'redisMessageSubscriber')(stream, [server.toRedis('sync-update', second)]) + const scheduled = Reflect.get(server, 'debouncedPersistMap') + clearTimeout(scheduled.get(room)) + scheduled.delete(room) + } finally { + release.resolve(undefined) + } + await operation + assert.deepEqual((await client.redis.xRange(stream, '-', '+')).map(m => m.id), ['1000-1']) + const saved = await storage.retrieveDoc(room, 'index') + assert.ok(saved) + const doc = new Y.Doc() + Y.applyUpdateV2(doc, saved.doc) + assert.deepEqual(doc.getMap().toJSON(), { first: 1 }) + doc.destroy() + await server.persistNamespace(room) + assert.equal(await client.redis.xLen(stream), 0) +}) + +test('trimming requires a snapshot boundary and honors minimum retention', async t => { + const { client, room, stream, append } = await persistenceCase(t) + const { first } = createUpdates() + await append('1000-0', first) + const recent = `${Date.now()}-0` + await append(recent, first) + await client.trimRoomStream(room, 'index') + assert.equal(await client.redis.xLen(stream), 2) + client.redisMinMessageLifetime = 60000 + await client.trimRoomStream(room, 'index', recent) + assert.deepEqual((await client.redis.xRange(stream, '-', '+')).map(m => m.id), [recent]) +}) + +test('queue workers trim through their snapshot rather than the task ID or current stream head', async t => { + const { client, storage, stream, append } = await persistenceCase(t) + const { first, second } = createUpdates() + await append('1000-0', first) + await client.redis.xAdd(client.redisWorkerStreamName, '*', { compact: stream }) + const started = promiseWithResolvers() + const release = promiseWithResolvers() + const persist = storage.persistDoc.bind(storage) + storage.persistDoc = async (...args) => { + started.resolve(undefined) + await release.promise + await persist(...args) + } + const operation = client.consumeWorkerQueue({ tryReclaimCount: 0, tryClaimCount: 1, blockTime: 10 }) + await started.promise + await append('1000-1', second) + release.resolve(undefined) + await operation + assert.deepEqual((await client.redis.xRange(stream, '-', '+')).map(m => m.id), ['1000-1']) + await client.consumeWorkerQueue({ tryReclaimCount: 0, tryClaimCount: 1, blockTime: 10 }) + assert.equal(await client.redis.xLen(stream), 0) + await client.consumeWorkerQueue({ tryReclaimCount: 0, tryClaimCount: 1, blockTime: 10 }) + assert.equal(await client.redis.xLen(client.redisWorkerStreamName), 0) +}) + +test('a failed queue save retains its stream and can be reclaimed', async t => { + const { client, storage, stream, append } = await persistenceCase(t) + await append('1000-0', createUpdates().first) + await client.redis.xAdd(client.redisWorkerStreamName, '*', { compact: stream }) + const persist = storage.persistDoc.bind(storage) + storage.persistDoc = async () => { throw new Error('Save failed') } + await assert.rejects(client.consumeWorkerQueue({ tryReclaimCount: 0, tryClaimCount: 1, blockTime: 10 }), /Save failed/) + assert.equal(await client.redis.xLen(stream), 1) + storage.persistDoc = persist + client.redisWorkerTimeout = 0 + await client.consumeWorkerQueue({ tryReclaimCount: 1, tryClaimCount: 0 }) + assert.equal(await client.redis.xLen(stream), 0) +}) + +test('out-of-order Yjs structs are persisted before their stream entries are trimmed', async t => { + const { client, storage, room, stream, append } = await persistenceCase(t) + const source = new Y.Doc() + /** @type {Uint8Array[]} */ + const updates = [] + source.on('update', update => updates.push(update)) + source.getText().insert(0, 'a') + source.getText().insert(1, 'b') + await append('1000-0', updates[1]) + await client.redis.xAdd(client.redisWorkerStreamName, '*', { compact: stream }) + await client.consumeWorkerQueue({ tryReclaimCount: 0, tryClaimCount: 1, blockTime: 10 }) + assert.equal(await client.redis.xLen(stream), 0) + const saved = await storage.retrieveDoc(room, 'index') + assert.ok(saved) + const restored = new Y.Doc() + Y.applyUpdateV2(restored, saved.doc) + Y.applyUpdate(restored, updates[0]) + assert.equal(restored.getText().toString(), 'ab') + source.destroy() + restored.destroy() +}) + +test('failed direct saves retry without another update and defer namespace cleanup', async t => { + const { server, client, storage, room, stream, append, load } = await persistenceCase(t) + await append('1000-0', createUpdates().first) + await load() + const firstSave = promiseWithResolvers() + const retry = promiseWithResolvers() + const persist = storage.persistDoc.bind(storage) + let attempts = 0 + storage.persistDoc = async (...args) => { + attempts++ + if (attempts === 1) { + await firstSave.promise + throw new Error('Temporary save failure') + } + await retry.promise + await persist(...args) + } + t.mock.method(console, 'error', () => {}) + server.debouncedPersist(room, true) + await until(() => attempts === 1) + server.cleanupNamespace(room, stream, 1) + await delay(10) + firstSave.resolve(undefined) + await until(() => attempts === 2) + assert.equal(await client.redis.xLen(stream), 1) + assert.equal(Reflect.get(server, 'namespaceDocMap').has(room), true) + server.cancelNamespaceCleanup(room) + retry.resolve(undefined) + await until(async () => await client.redis.xLen(stream) === 0) + await until(() => !Reflect.get(server, 'awaitingPersistMap').has(room)) + assert.equal(Reflect.get(server, 'namespaceDocMap').has(room), true) + server.cleanupNamespace(room, stream, 1) + await until(() => !Reflect.get(server, 'namespaceDocMap').has(room)) +}) + +class ControlledWorker extends EventEmitter { + /** @type {Array<{room: string, requestId: string, docstate: Uint8Array}>} */ + requests = [] + stopped = false + + /** @param {{event?: string, room?: string, requestId?: string, docstate?: Uint8Array}} message */ + postMessage (message) { + if (this.stopped) throw new Error('Worker stopped') + if (message.event === 'ping') { + this.emit('message', { event: 'pong' }) + } else { + assert.ok(message.room && message.requestId && message.docstate) + this.requests.push({ room: message.room, requestId: message.requestId, docstate: message.docstate }) + } + } + + async terminate () { + this.stopped = true + this.emit('exit', 1) + return 1 + } +} + +/** @param {SocketIOServer} server @param {ControlledWorker | Worker} worker */ +function attachWorker (server, worker) { + Reflect.set(server, 'persistWorker', worker) + server.registerPersistWorkerResolve() + worker.emit('message', { event: 'ready' }) +} + +test('stale, uncorrelated, and wrong-room worker completions cannot settle a retry', async t => { + const { server, client, room, stream, append, load } = await persistenceCase(t) + await append('1000-0', createUpdates().first) + await load() + const worker = new ControlledWorker() + attachWorker(server, worker) + t.mock.method(console, 'error', () => {}) + server.debouncedPersist(room, true) + await until(() => worker.requests.length === 1) + const first = worker.requests[0] + worker.emit('message', { event: 'persist-error', room, requestId: first.requestId, error: 'Save failed' }) + await until(() => worker.requests.length === 2) + const second = worker.requests[1] + assert.notEqual(first.requestId, second.requestId) + worker.emit('message', { event: 'persisted', room, requestId: first.requestId }) + worker.emit('message', { event: 'persisted', room }) + worker.emit('message', { event: 'persisted', room: 'another-room', requestId: second.requestId }) + await delay(10) + assert.equal(await client.redis.xLen(stream), 1) + assert.equal(Reflect.get(server, 'workerPersistRequests').size, 1) + worker.emit('message', { event: 'persisted', room, requestId: second.requestId }) + await until(async () => await client.redis.xLen(stream) === 0) + assert.equal(Reflect.get(server, 'workerPersistRequests').size, 0) +}) + +for (const failure of ['error', 'exit', 'health']) { + test(`worker ${failure} releases pending saves and allows retry after replacement`, async t => { + const { server, client, room, stream, append, load } = await persistenceCase(t) + await append('1000-0', createUpdates().first) + await load() + const oldWorker = new ControlledWorker() + attachWorker(server, oldWorker) + const replacement = new ControlledWorker() + Reflect.set(server, 'getPersistWorker', () => { + queueMicrotask(() => replacement.emit('message', { event: 'ready' })) + return replacement + }) + t.mock.method(console, 'error', () => {}) + server.debouncedPersist(room, true) + await until(() => oldWorker.requests.length === 1) + oldWorker.stopped = true + if (failure === 'error') oldWorker.emit('error', new Error('Worker crashed')) + if (failure === 'exit') oldWorker.emit('exit', 1) + let checked = false + t.mock.method(server, 'workerHealthCheck', async () => { + const health = checked + checked = true + return health + }) + server.registerPersistWorkerHealthCheck() + await until(() => Object.is(Reflect.get(server, 'persistWorker'), replacement)) + // A retry may use direct persistence while the worker is unavailable. + await until(async () => replacement.requests.length > 0 || await client.redis.xLen(stream) === 0) + const request = replacement.requests[0] + if (request) { + oldWorker.emit('message', { event: 'persisted', room, requestId: request.requestId }) + await delay(5) + assert.equal(await client.redis.xLen(stream), 1) + replacement.emit('message', { event: 'persisted', room, requestId: request.requestId }) + } + await until(async () => await client.redis.xLen(stream) === 0) + await until(() => !Reflect.get(server, 'awaitingPersistMap').has(room)) + assert.equal(Reflect.get(server, 'workerPersistRequests').size, 0) + }) +} + +test('the real persistence worker reports storage errors and completes a later retry', async t => { + const { server, client, room, stream, append, load } = await persistenceCase(t) + await append('1000-0', createUpdates().first) + await load() + const moduleUrl = new URL('../src/persist-worker-thread.js', import.meta.url).href + const worker = new Worker(` + const { parentPort } = require('node:worker_threads') + ;(async () => { + const { createPersistWorkerThread } = await import(${JSON.stringify(moduleUrl)}) + const Y = await import('yjs') + let attempts = 0 + createPersistWorkerThread({ + persistDoc: async (room, docname, doc) => { + if (++attempts === 1) throw new Error('Storage unavailable') + parentPort.postMessage({ event: 'stored', state: Y.encodeStateAsUpdateV2(doc) }) + } + }) + })() + `, { eval: true }) + /** @type {Array<{event: string, requestId?: string, state?: Uint8Array}>} */ + const messages = [] + worker.on('message', message => messages.push(message)) + Reflect.set(server, 'persistWorker', worker) + server.registerPersistWorkerResolve() + await until(() => Reflect.get(server, 'workerReady')) + t.mock.method(console, 'error', () => {}) + server.debouncedPersist(room, true) + await until(async () => await client.redis.xLen(stream) === 0) + const failed = messages.find(m => m.event === 'persist-error') + const succeeded = messages.find(m => m.event === 'persisted') + assert.ok(failed?.requestId && succeeded?.requestId) + assert.notEqual(failed.requestId, succeeded.requestId) + const state = messages.find(m => m.event === 'stored')?.state + assert.ok(state) + const saved = new Y.Doc() + Y.applyUpdateV2(saved, state) + assert.deepEqual(saved.getMap().toJSON(), { first: 1 }) + saved.destroy() +})