trying to fix redis memory leak

Author: janthurau

packages/extension-redis/src/Redis.ts

@@ -134,7 +134,9 @@ export class Redis implements Extension {
     }
     this.sub.on('messageBuffer', this.handleIncomingMessage)
 
-    this.redlock = new Redlock([this.pub])
+    this.redlock = new Redlock([this.pub], {
+      retryCount: 0,
+    })
 
     const identifierBuffer = Buffer.from(this.configuration.identifier, 'utf-8')
     this.messagePrefix = Buffer.concat([Buffer.from([identifierBuffer.length]), identifierBuffer])
@@ -224,12 +226,14 @@ export class Redis implements Extension {
   async onStoreDocument({ documentName }: onStoreDocumentPayload) {
     // Attempt to acquire a lock and read lastReceivedTimestamp from Redis,
     // to avoid conflict with other instances storing the same document.
+
     return new Promise((resolve, reject) => {
       this.redlock.lock(this.lockKey(documentName), this.configuration.lockTimeout, async (error, lock) => {
         if (error || !lock) {
           // Expected behavior: Could not acquire lock, another instance locked it already.
           // No further `onStoreDocument` hooks will be executed.
-          reject(error)
+          console.log('unable to acquire lock')
+          reject()
           return
         }
 

tests/extension-redis/onStoreDocument.ts

@@ -24,7 +24,7 @@ test('stores documents without conflicts', async t => {
         new Redis({
           ...redisConnectionSettings,
           identifier: `server${uuidv4()}`,
-          prefix: 'extension-redis/onStoreDocument',
+          prefix: 'extension-redis/onStoreDocument1',
         }),
         new CustomStorageExtension(),
       ],
@@ -36,7 +36,7 @@ test('stores documents without conflicts', async t => {
         new Redis({
           ...redisConnectionSettings,
           identifier: `anotherServer${uuidv4()}`,
-          prefix: 'extension-redis/onStoreDocument',
+          prefix: 'extension-redis/onStoreDocument1',
         }),
         new CustomStorageExtension(),
       ],
@@ -63,7 +63,7 @@ test('stores documents when the last client disconnects', async t => {
     const server = await newHocuspocus({
       extensions: [
         new Redis({
-          prefix: 'extension-redis/onStoreDocument',
+          prefix: 'extension-redis/onStoreDocument2',
           ...redisConnectionSettings,
         }),
       ],
@@ -85,3 +85,69 @@ test('stores documents when the last client disconnects', async t => {
     })
   })
 })
+
+test('document gets unloaded on both servers after disconnection', async t => {
+  await new Promise(async resolve => {
+    class CustomStorageExtension {
+      priority = 10
+
+      onStoreDocument({ document }: onStoreDocumentPayload) {
+        console.log('storing')
+        return new Promise(resolve2 => {
+          setTimeout(() => {
+            console.log('stored')
+
+            resolve2('')
+          }, 3000)
+        })
+      }
+    }
+
+    const server = await newHocuspocus({
+      name: 'redis-1',
+      extensions: [
+        new Redis({
+          ...redisConnectionSettings,
+          prefix: 'extension-redis/onStoreDocument3',
+        }),
+        new CustomStorageExtension(),
+      ],
+    })
+
+    const anotherServer = await newHocuspocus({
+      name: 'redis-2',
+      extensions: [
+        new Redis({
+          ...redisConnectionSettings,
+          prefix: 'extension-redis/onStoreDocument3',
+        }),
+        new CustomStorageExtension(),
+      ],
+    })
+
+    const provider = newHocuspocusProvider(server)
+
+    const anotherProvider = newHocuspocusProvider(anotherServer, {
+      onSynced() {
+        // once we're setup make an edit on anotherProvider, if all succeeds the onStoreDocument
+        // callback will be called after the debounce period and all docs will
+        // be identical
+        anotherProvider.document.getArray('foo').insert(0, ['bar'])
+        provider.document.getArray('foo2').insert(0, ['bar'])
+
+        setTimeout(() => {
+          provider.configuration.websocketProvider.disconnect()
+          anotherProvider.configuration.websocketProvider.disconnect()
+
+          setTimeout(() => {
+            t.is(anotherServer.documents.size, 0)
+            t.is(server.documents.size, 0)
+
+            resolve('')
+          }, 5000) // must be higher than RedisExtension.disconnectDelay
+        }, 1500)
+
+      },
+    })
+  })
+})