From 38c85a9b107686b68c6558a5b2c97a18cf779519 Mon Sep 17 00:00:00 2001 From: stanley2058 Date: Fri, 2 Oct 2026 15:53:25 +0800 Subject: [PATCH] fix: acknowledge accepted updates and guard persistence leases --- .github/workflows/build.yml | 8 + README.md | 23 ++ package.json | 1 + src/y-socket-io/y-socket-io.js | 184 +++++++++------ tests/consistency.test.js | 410 +++++++++++++++++++++++++++++++++ 5 files changed, 551 insertions(+), 75 deletions(-) create mode 100644 tests/consistency.test.js diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 07806a9..e9a320e 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -9,6 +9,11 @@ on: jobs: test: runs-on: ubuntu-latest + services: + redis: + image: redis:7-alpine + ports: + - 6379:6379 steps: - uses: actions/checkout@v3 - uses: actions/setup-node@v3 @@ -17,4 +22,7 @@ jobs: - uses: pnpm/action-setup@v4 - run: pnpm i - run: pnpm lint + - run: pnpm test:consistency + env: + YSR_TEST_REDIS: redis://127.0.0.1:6379 - run: pnpm dist diff --git a/README.md b/README.md index 22b77aa..bc978f1 100644 --- a/README.md +++ b/README.md @@ -219,6 +219,29 @@ npm start Open [`http://localhost:5173`](http://localhost:5173) in a browser. +## Consistency tests + +Run the ACK, lease ownership, and reconnect regression tests against a disposable +Redis instance: + +```sh +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. + +A sync-update ACK confirms Redis stream acceptance, not database persistence. +Retries are appended again and safely merged by Yjs. Persistence leases use a +unique token per acquisition; renewal and release atomically check that token. +All servers must be upgraded before relying on owner-checked leases, because +older servers can still unconditionally remove another server's lease. + +Lease ownership does not fence database commits. A worker dispatched before +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. + ## License [The MIT License](./LICENSE) © Kevin Jahns diff --git a/package.json b/package.json index e95a224..8524f8a 100644 --- a/package.json +++ b/package.json @@ -17,6 +17,7 @@ "lint": "standard && tsc", "lint:fix": "standard --fix && tsc", "test": "dotenvx run --env-file=.env -- node tests/index.js", + "test:consistency": "node --test tests/consistency.test.js", "test-inspect": "dotenvx run --env-file=.env -- node --inspect-brk tests/index.js", "preversion": "npm run lint && npm run dist", "clean": "rm -rf **/dist " diff --git a/src/y-socket-io/y-socket-io.js b/src/y-socket-io/y-socket-io.js index 3da56b3..bf7bcee 100644 --- a/src/y-socket-io/y-socket-io.js +++ b/src/y-socket-io/y-socket-io.js @@ -177,10 +177,14 @@ export class YSocketIO { */ persistentHeartbeatTimeout = null /** - * @type {Set} + * Acquisition tokens, not server IDs: an expired lease must never be reused. + * @type {Map} * @private */ - persistentLeaderOf = new Set() + persistentLeaderOf = new Map() + /** @type {Map>} */ + pendingLeaderAcquisitions = new Map() + destroyed = false /** * @type {string} * @private @@ -269,11 +273,7 @@ export class YSocketIO { 'index', redisPrefix ) - const prevAwaitCleanup = this.awaitingCleanupNamespace.get(namespace) - if (prevAwaitCleanup) { - clearTimeout(prevAwaitCleanup) - this.cleanupNamespace(namespace, stream) - } + this.cancelNamespaceCleanup(namespace) if (!this.namespaceMap.has(namespace)) { this.namespaceMap.set(namespace, socket.nsp) @@ -348,10 +348,9 @@ export class YSocketIO { } ) - /** @type {unknown} */ - let prevMsg = null socket.on('sync-update', (/** @type {ArrayBuffer} */ update, /** @type {() => void} */ ack) => { - if (isDeepStrictEqual(update, prevMsg)) return ack() + // Retries are safe to append: Yjs updates are idempotent. Every ACK must + // follow its own successful append, including overlapping retries. assert(this.client) const namespace = this.getNamespaceString(socket.nsp) const message = Buffer.from(update.slice(0, update.byteLength)) @@ -363,7 +362,6 @@ export class YSocketIO { ) .then(() => ack()) .catch(console.error) - prevMsg = update }) } @@ -679,6 +677,8 @@ export class YSocketIO { } destroy () { + this.destroyed = true + this.pendingLeaderAcquisitions.clear() try { if (this.persistWorkerHealthCheckTimeout) { clearInterval(this.persistWorkerHealthCheckTimeout) @@ -704,6 +704,12 @@ export class YSocketIO { }) } + /** @param {string} namespace */ + cancelNamespaceCleanup (namespace) { + clearTimeout(this.awaitingCleanupNamespace.get(namespace)) + this.awaitingCleanupNamespace.delete(namespace) + } + /** * @param {string} namespace * @param {string} stream @@ -725,6 +731,13 @@ export class YSocketIO { const timer = setTimeout(async () => { const awaitingPersist = this.awaitingPersistMap.get(namespace) if (awaitingPersist) await awaitingPersist.promise + // 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 + if (this.namespaceMap.get(namespace)?.sockets.size) { + this.awaitingCleanupNamespace.delete(namespace) + return + } this.cleanupNamespaceImpl(namespace, stream) this.awaitingCleanupNamespace.delete(namespace) logSocketIO(`no active connection, namespace: ${namespace} cleared`) @@ -745,15 +758,28 @@ export class YSocketIO { this.namespaceDocMap.get(namespace)?.ydoc.destroy() this.namespaceDocMap.delete(namespace) this.namespacePersistentMap.delete(namespace) + const pendingPersist = this.debouncedPersistMap.get(namespace) + if (pendingPersist) clearTimeout(pendingPersist) + this.debouncedPersistMap.delete(namespace) + this.pendingLeaderAcquisitions.delete(namespace) + const token = this.persistentLeaderOf.get(namespace) this.persistentLeaderOf.delete(namespace) - // clear leader key in redis when we fully clean up - if (this.client) { - const key = this.getLeaderKeyOf(namespace) - this.client.redis - .del(key) - .catch((e) => console.error(e)) - } + if (token) this.releasePersistentLeader(namespace, token).catch(console.error) + } + + /** + * @param {string} namespace + * @param {string} token + */ + async releasePersistentLeader (namespace, token) { + assert(this.client) + return this.client.redis.eval(` + if redis.call('GET', KEYS[1]) == ARGV[1] then + return redis.call('DEL', KEYS[1]) + end + return 0 + `, { keys: [this.getLeaderKeyOf(namespace)], arguments: [token] }) } async waitUntilWorkerReady () { @@ -819,74 +845,82 @@ export class YSocketIO { } async registerPersistentLeaderHeartbeat () { + if (this.destroyed) return this.persistentHeartbeatTimeout = setTimeout(async () => { - assert(this.client) - const redis = this.client.redis - try { - /** - * @type {Array>} - */ - const promises = [] - for (const namespace of this.persistentLeaderOf) { - const key = this.getLeaderKeyOf(namespace) - const curLeader = await redis.get(key) - - // remove orphaned if exist - const nsp = this.namespaceMap.get(namespace) - const hasDoc = this.namespaceDocMap.has(namespace) - const hasCleanupTimer = this.awaitingCleanupNamespace.has(namespace) - const isActiveHere = Boolean(nsp || hasDoc || hasCleanupTimer) - if (!isActiveHere) { - logSocketIO(`clearing leader heartbeat for [${namespace}] (SID: ${this.serverId})`) - this.persistentLeaderOf.delete(namespace) - continue - } - - if (curLeader === this.serverId) { - logSocketIO(`set leader heartbeat for [${namespace}] (SID: ${this.serverId})`) - promises.push( - redis.set(key, this.serverId, { - XX: true, - PX: PERSIST_LEADER_HEARTBEAT_INTERVAL - }) - ) - } else { - logSocketIO(`lost leadership for [${namespace}] (SID: ${this.serverId})`) - this.persistentLeaderOf.delete(namespace) - } - } - - await promise.all(promises) + await this.renewPersistentLeaders() } catch (e) { - console.error(e) + if (!(e instanceof ClientClosedError)) console.error(e) } + this.registerPersistentLeaderHeartbeat() + }, PERSIST_LEADER_HEARTBEAT_INTERVAL / 2) + } - // register next round - this.persistentHeartbeatTimeout = setTimeout( - () => this.registerPersistentLeaderHeartbeat(), - PERSIST_LEADER_HEARTBEAT_INTERVAL / 2 - ) - }) + async renewPersistentLeaders () { + assert(this.client) + for (const [namespace, token] of this.persistentLeaderOf) { + const isActiveHere = this.namespaceMap.has(namespace) || + this.namespaceDocMap.has(namespace) || this.awaitingCleanupNamespace.has(namespace) + if (!isActiveHere) { + this.persistentLeaderOf.delete(namespace) + await this.releasePersistentLeader(namespace, token) + continue + } + const renewed = await this.client.redis.eval(` + if redis.call('GET', KEYS[1]) == ARGV[1] then + return redis.call('PEXPIRE', KEYS[1], ARGV[2]) + end + return 0 + `, { + keys: [this.getLeaderKeyOf(namespace)], + arguments: [token, String(PERSIST_LEADER_HEARTBEAT_INTERVAL)] + }) + if (!renewed && this.persistentLeaderOf.get(namespace) === token) { + this.persistentLeaderOf.delete(namespace) + } + } } /** * @param {string} namespace */ - async tryAcquirePersistentLeader (namespace) { + tryAcquirePersistentLeader (namespace) { + if (this.destroyed) return Promise.resolve(false) + const pending = this.pendingLeaderAcquisitions.get(namespace) + if (pending) return pending assert(this.client) - const redis = this.client.redis - const key = this.getLeaderKeyOf(namespace) - const prevVal = await redis.set(key, this.serverId, { - NX: true, - PX: PERSIST_LEADER_HEARTBEAT_INTERVAL, - GET: true + const previousToken = this.persistentLeaderOf.get(namespace) || '' + const newToken = `${this.serverId}:${randomUUID()}` + // Renew an existing acquisition atomically, or create a fresh token. A + // delayed release/renewal from an expired acquisition cannot touch this one. + const request = this.client.redis.eval(` + if redis.call('GET', KEYS[1]) == ARGV[1] then + redis.call('PEXPIRE', KEYS[1], ARGV[3]) + return ARGV[1] + end + if redis.call('SET', KEYS[1], ARGV[2], 'NX', 'PX', ARGV[3]) then + return ARGV[2] + end + return false + `, { + keys: [this.getLeaderKeyOf(namespace)], + arguments: [previousToken, newToken, String(PERSIST_LEADER_HEARTBEAT_INTERVAL)] + }).then(async result => { + const token = typeof result === 'string' ? result : null + // Cleanup may have run while Redis was processing the acquisition. + if (this.pendingLeaderAcquisitions.get(namespace) !== request) { + if (token) await this.releasePersistentLeader(namespace, token) + return false + } + if (token) this.persistentLeaderOf.set(namespace, token) + else this.persistentLeaderOf.delete(namespace) + return token !== null + }).finally(() => { + if (this.pendingLeaderAcquisitions.get(namespace) === request) { + this.pendingLeaderAcquisitions.delete(namespace) + } }) - - const ok = prevVal === this.serverId || prevVal === null - if (!ok) return false - - this.persistentLeaderOf.add(namespace) - return true + this.pendingLeaderAcquisitions.set(namespace, request) + return request } } diff --git a/tests/consistency.test.js b/tests/consistency.test.js new file mode 100644 index 0000000..b5fb854 --- /dev/null +++ b/tests/consistency.test.js @@ -0,0 +1,410 @@ +import { after, before, test } from 'node:test' +import assert from 'node:assert/strict' +import { randomUUID } from 'node:crypto' +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 * 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' + +/** @param {() => boolean | Promise} predicate */ +async function until (predicate) { + const deadline = Date.now() + 3000 + while (!await predicate()) { + assert.ok(Date.now() < deadline, 'condition did not settle') + await delay(5) + } +} + +function createServer () { + return new YSocketIO(new Server(), { authenticate: () => ({ userid: 'test' }), enableAwareness: false }) +} + +function createUpdates () { + const doc = new Y.Doc() + doc.getMap().set('first', 1) + const first = Y.encodeStateAsUpdate(doc) + doc.getMap().set('second', 2) + const second = Y.encodeStateAsUpdate(doc) + doc.destroy() + return { first, second } +} + +// Exercise the registered sync handler with controlled append completion. +function createSyncHarness () { + const server = createServer() + /** @type {Map} */ + const handlers = new Map() + /** @type {Array>>} */ + const appends = [] + Reflect.set(server, 'client', { + addMessage: () => { + const append = promiseWithResolvers() + appends.push(append) + return append.promise + } + }) + Reflect.get(server, 'initSyncListeners')({ + nsp: { name: '/yjs|test' }, + on: (/** @type {string} */ event, /** @type {Function} */ handler) => handlers.set(event, handler) + }) + const update = handlers.get('sync-update') + assert.ok(update) + return { update, appends } +} + +test('failed append stays unacknowledged and an identical retry appends again', async () => { + const { update, appends } = createSyncHarness() + const { first } = createUpdates() + let acks = 0 + const originalError = console.error + /** @type {unknown[]} */ + const errors = [] + console.error = error => errors.push(error) + try { + update(first, () => acks++) + appends[0].reject(new Error('append failed')) + await until(() => errors.length === 1) + assert.equal(acks, 0) + update(first.slice(), () => acks++) + assert.equal(appends.length, 2) + assert.equal(acks, 0) + appends[1].resolve(undefined) + await until(() => acks === 1) + } finally { + console.error = originalError + } +}) + +test('concurrent duplicates and different updates each await successful acceptance', async () => { + const { update, appends } = createSyncHarness() + const { first, second } = createUpdates() + /** @type {string[]} */ + const acks = [] + update(first, () => acks.push('first')) + update(first.slice(), () => acks.push('retry')) + update(second, () => acks.push('second')) + assert.equal(appends.length, 3) + assert.deepEqual(acks, []) + appends[2].resolve(undefined) + await until(() => acks.length === 1) + assert.deepEqual(acks, ['second']) + appends[1].resolve(undefined) + await until(() => acks.length === 2) + assert.deepEqual(acks, ['second', 'retry']) + appends[0].resolve(undefined) + await until(() => acks.length === 3) +}) + +test('an accepted update with a lost ACK can be safely submitted again', async () => { + const { update, appends } = createSyncHarness() + const { first } = createUpdates() + update(first, () => {}) // ACK is lost to the sender. + appends[0].resolve(undefined) + await delay(0) + let acked = false + update(first.slice(), () => { acked = true }) + assert.equal(appends.length, 2) + assert.equal(acked, false) + appends[1].resolve(undefined) + await until(() => acked) + const doc = new Y.Doc() + Y.applyUpdate(doc, first) + Y.applyUpdate(doc, first) + assert.deepEqual(doc.getMap().toJSON(), { first: 1 }) + doc.destroy() +}) + +// All Redis keys are scoped to this run; no shared database is flushed. +const redisUrl = process.env.YSR_TEST_REDIS || 'redis://127.0.0.1:36379' +const prefix = `consistency-${randomUUID()}` +const store = createMemoryStorage() +/** @type {import('../src/api.js').Api[]} */ +const clients = [] +/** @type {YSocketIO[]} */ +const servers = [] + +before(async () => { + for (let i = 0; i < 3; i++) { + const client = await createApiClient(store, { redisUrl, redisPrefix: prefix, enableAwareness: false }) + clients.push(client) + const server = createServer() + Reflect.set(server, 'client', client) + servers.push(server) + } +}) + +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) + } + if (clients[0]) { + const keys = await clients[0].redis.keys(`${prefix}:*`) + if (keys.length) await clients[0].redis.del(keys) + } + await Promise.all(clients.map(client => client.destroy())) +}) + +/** @param {YSocketIO} 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 */ +function cleanup (server, room) { + Reflect.get(server, 'cleanupNamespaceImpl').call(server, room, `${prefix}:room:${room}:index`) +} + +/** @param {string} key */ +async function expireLease (key) { + await clients[0].redis.pExpire(key, 1) + await until(async () => await clients[0].redis.get(key) === null) +} + +test('follower cleanup cannot release the active leader; owner cleanup allows takeover', async () => { + const [a, b, c] = servers + const room = 'follower-cleanup' + activate(a, room) + assert.equal(await a.tryAcquirePersistentLeader(room), true) + const key = a.getLeaderKeyOf(room) + const token = await clients[0].redis.get(key) + assert.equal(await b.tryAcquirePersistentLeader(room), false) + cleanup(b, room) + assert.equal(await clients[0].redis.get(key), token) + assert.equal(await c.tryAcquirePersistentLeader(room), false) + cleanup(a, room) + await until(async () => await clients[0].redis.get(key) === null) + assert.equal(await c.tryAcquirePersistentLeader(room), true) +}) + +test('concurrent acquisition has one winner and same-server calls share the acquisition', async () => { + const [a, b, c] = servers + const room = 'concurrent-acquisition' + const first = a.tryAcquirePersistentLeader(room) + assert.equal(a.tryAcquirePersistentLeader(room), first) + const results = await Promise.all([first, b.tryAcquirePersistentLeader(room), c.tryAcquirePersistentLeader(room)]) + assert.equal(results.filter(Boolean).length, 1) + const winner = servers[results.indexOf(true)] + const key = winner.getLeaderKeyOf(room) + const token = await clients[0].redis.get(key) + assert.equal(await winner.tryAcquirePersistentLeader(room), true) + assert.equal(await clients[0].redis.get(key), token) +}) + +test('expired owner cannot release or renew a different server lease', async () => { + const [a, b] = servers + const room = 'expired-owner' + activate(a, room) + assert.equal(await a.tryAcquirePersistentLeader(room), true) + const key = a.getLeaderKeyOf(room) + const oldToken = await clients[0].redis.get(key) + assert.ok(oldToken) + await expireLease(key) + assert.equal(await b.tryAcquirePersistentLeader(room), true) + const newToken = await clients[0].redis.get(key) + await clients[0].redis.pExpire(key, 2000) + await a.renewPersistentLeaders() + assert.equal(await clients[0].redis.get(key), newToken) + assert.ok(await clients[0].redis.pTTL(key) <= 2000) + assert.equal(await a.releasePersistentLeader(room, oldToken), 0) + assert.equal(await clients[0].redis.get(key), newToken) +}) + +test('reacquisition by the same server gets a fresh token immune to old cleanup', async () => { + const [a] = servers + const room = 'same-server-reacquisition' + assert.equal(await a.tryAcquirePersistentLeader(room), true) + const key = a.getLeaderKeyOf(room) + const oldToken = await clients[0].redis.get(key) + assert.ok(oldToken) + await expireLease(key) + assert.equal(await a.tryAcquirePersistentLeader(room), true) + const newToken = await clients[0].redis.get(key) + assert.notEqual(newToken, oldToken) + assert.equal(await a.releasePersistentLeader(room, oldToken), 0) + assert.equal(await clients[0].redis.get(key), newToken) +}) + +test('heartbeat extends only a still-owned active lease', async () => { + const [a] = servers + const room = 'heartbeat' + activate(a, room) + await a.tryAcquirePersistentLeader(room) + const key = a.getLeaderKeyOf(room) + const token = await clients[0].redis.get(key) + await clients[0].redis.pExpire(key, 500) + await a.renewPersistentLeaders() + assert.equal(await clients[0].redis.get(key), token) + assert.ok(await clients[0].redis.pTTL(key) > 500) +}) + +test('cleanup during acquisition releases the late result without restoring local ownership', async () => { + const [a] = servers + const room = 'cleanup-during-acquisition' + const redis = clients[0].redis + const originalEval = redis.eval.bind(redis) + const acquired = promiseWithResolvers() + const resume = promiseWithResolvers() + redis.eval = async (...args) => { + const result = await originalEval(...args) + acquired.resolve(undefined) + await resume.promise + return result + } + try { + const request = a.tryAcquirePersistentLeader(room) + await acquired.promise + cleanup(a, room) + redis.eval = originalEval + resume.resolve(undefined) + assert.equal(await request, false) + assert.equal(await redis.get(a.getLeaderKeyOf(room)), null) + assert.equal(Reflect.get(a, 'persistentLeaderOf').has(room), false) + } finally { + redis.eval = originalEval + resume.resolve(undefined) + } +}) + +test('delayed heartbeat cannot steal a successor lease', async () => { + const [a, b] = servers + const room = 'delayed-heartbeat' + activate(a, room) + await a.tryAcquirePersistentLeader(room) + const key = a.getLeaderKeyOf(room) + const redis = clients[0].redis + const originalEval = redis.eval.bind(redis) + const entered = promiseWithResolvers() + const resume = promiseWithResolvers() + redis.eval = async (...args) => { + entered.resolve(undefined) + await resume.promise + return originalEval(...args) + } + try { + const heartbeat = a.renewPersistentLeaders() + await entered.promise + await expireLease(key) + await b.tryAcquirePersistentLeader(room) + const token = await redis.get(key) + resume.resolve(undefined) + await heartbeat + assert.equal(await redis.get(key), token) + assert.equal(Reflect.get(a, 'persistentLeaderOf').has(room), false) + } finally { + redis.eval = originalEval + resume.resolve(undefined) + } +}) + +test('reconnect cancels cleanup even after its timer has begun waiting for persistence', async () => { + const [a] = servers + const room = 'reconnect' + activate(a, room) + await a.tryAcquirePersistentLeader(room) + const key = a.getLeaderKeyOf(room) + const token = await clients[0].redis.get(key) + const persistence = promiseWithResolvers() + Reflect.get(a, 'awaitingPersistMap').set(room, persistence) + a.cleanupNamespace(room, `${prefix}:room:${room}:index`, 1) + await delay(20) + a.cancelNamespaceCleanup(room) + persistence.resolve(undefined) + await delay(20) + assert.equal(await clients[0].redis.get(key), token) + assert.equal(Reflect.get(a, 'namespaceMap').has(room), true) + Reflect.get(a, 'awaitingPersistMap').delete(room) +}) + +test('Socket.IO retry after disconnect is accepted and acknowledged through Redis', async () => { + const httpServer = createHttpServer() + const io = new Server(httpServer) + const server = new YSocketIO(io, { authenticate: () => ({ userid: 'test' }), enableAwareness: false }) + Reflect.set(server, 'client', clients[0]) + const namespace = io.of('/yjs|socket-reconnect') + namespace.on('connection', socket => Reflect.get(server, 'initSyncListeners')(socket)) + httpServer.listen(0) + const address = httpServer.address() + assert.ok(address && typeof address !== 'string') + const socket = connect(`http://127.0.0.1:${address.port}/yjs|socket-reconnect`, { reconnection: false, forceNew: true }) + const { first } = createUpdates() + try { + await until(() => socket.connected) + await socket.timeout(2000).emitWithAck('sync-update', first) + socket.disconnect() + socket.connect() + await until(() => socket.connected) + await socket.timeout(2000).emitWithAck('sync-update', first.slice()) + const persisted = await clients[0].getDoc('socket-reconnect', 'index') + assert.deepEqual(persisted.ydoc.getMap().toJSON(), { first: 1 }) + persisted.ydoc.destroy() + assert.equal(await clients[0].redis.xLen(`${prefix}:room:socket-reconnect:index`), 2) + } finally { + socket.disconnect() + await new Promise(resolve => io.close(resolve)) + } +}) + +test('failed concurrent duplicates receive no ACK and remain retryable', async () => { + const { update, appends } = createSyncHarness() + const { first } = createUpdates() + let acks = 0 + const originalError = console.error + /** @type {unknown[]} */ + const errors = [] + console.error = error => errors.push(error) + try { + update(first, () => acks++) + update(first.slice(), () => acks++) + appends[0].reject(new Error('first append failed')) + appends[1].reject(new Error('duplicate append failed')) + await until(() => errors.length === 2) + assert.equal(acks, 0) + update(first.slice(), () => acks++) + assert.equal(appends.length, 3) + appends[2].resolve(undefined) + await until(() => acks === 1) + } finally { + console.error = originalError + } +}) + +test('a rejected Redis acquisition does not block subsequent attempts', async () => { + const [a] = servers + const room = 'acquisition-failure' + const redis = clients[0].redis + const originalEval = redis.eval.bind(redis) + redis.eval = () => Promise.reject(new Error('Redis unavailable')) + try { + await assert.rejects(a.tryAcquirePersistentLeader(room), /Redis unavailable/) + assert.equal(Reflect.get(a, 'pendingLeaderAcquisitions').has(room), false) + } finally { + redis.eval = originalEval + } + assert.equal(await a.tryAcquirePersistentLeader(room), true) +}) + +test('a superseded cleanup callback cannot release the lease while a newer timer waits', async () => { + const [a] = servers + const room = 'extended-cleanup' + activate(a, room) + await a.tryAcquirePersistentLeader(room) + const key = a.getLeaderKeyOf(room) + const token = await clients[0].redis.get(key) + const persistence = promiseWithResolvers() + Reflect.get(a, 'awaitingPersistMap').set(room, persistence) + a.cleanupNamespace(room, `${prefix}:room:${room}:index`, 1) + await delay(20) + a.cleanupNamespace(room, `${prefix}:room:${room}:index`, 1000, true) + persistence.resolve(undefined) + await delay(20) + assert.equal(await clients[0].redis.get(key), token) + assert.equal(Reflect.get(a, 'namespaceMap').has(room), true) + a.cancelNamespaceCleanup(room) + Reflect.get(a, 'awaitingPersistMap').delete(room) +})