Merge pull request #26 from hackmdio/fix/use-ttl-to-delete-inactive-room

fix/avoid race condition by preserving room until no connection

Author: jackycute

src/api.js

@@ -20,6 +20,7 @@ if (ydocUpdateCallback != null && ydocUpdateCallback.slice(-1) !== '/') {
   ydocUpdateCallback += '/'
 }
 const WORKER_DISABLED = env.getConf('y-worker-disabled') === 'true'
+const ROOM_STREAM_TTL = number.parseInt(env.getConf('y-room-stream-ttl') || '300')
 
 /**
  * @param {string} a
@@ -125,14 +126,18 @@ export class Api {
     this.persistWorker = null
 
     const addScript = WORKER_DISABLED
-      ? 'redis.call("XADD", KEYS[1], "*", "m", ARGV[1])'
+      ? `
+          redis.call("XADD", KEYS[1], "*", "m", ARGV[1])
+          redis.call("EXPIRE", KEYS[1], ${ROOM_STREAM_TTL})
+        `
       : `
           if redis.call("EXISTS", KEYS[1]) == 0 then
             redis.call("XADD", "${this.redisWorkerStreamName}", "*", "compact", KEYS[1])
           elseif redis.call("XLEN", KEYS[1]) > 100 then
             redis.call("SADD", "${this.prefix}:worker:checklist", KEYS[1])
           end
           redis.call("XADD", KEYS[1], "*", "m", ARGV[1])
+          redis.call("EXPIRE", KEYS[1], ${ROOM_STREAM_TTL})
         `
 
     this.redis = redis.createClient({
@@ -294,20 +299,12 @@ export class Api {
   /**
    * @param {string} room
    * @param {string} docid
-   * @param {boolean} [remove=false]
    */
-  async trimRoomStream (room, docid, remove = false) {
+  async trimRoomStream (room, docid) {
     const roomName = computeRedisRoomStreamName(room, docid, this.prefix)
     const redisLastId = await this.getRedisLastId(room, docid)
     const lastId = number.parseInt(redisLastId.split('-')[0])
-    if (remove) {
-      await this.redis.del(roomName)
-    } else {
-      await this.redis.multi()
-        .xTrim(roomName, 'MINID', lastId - this.redisMinMessageLifetime)
-        .xDelIfEmpty(roomName)
-        .exec()
-    }
+    await this.redis.xTrim(roomName, 'MINID', lastId - this.redisMinMessageLifetime)
   }
 
   /**

src/y-socket-io/y-socket-io.js

@@ -537,16 +537,7 @@ export class YSocketIO {
             await this.client.store.persistDoc(namespace, 'index', doc)
           }
 
-          /**
-           * there's a possibility where the namespace is deleted after the
-           * persist promise resolved, so we have to check if the room still
-           * exist.
-           * @see cleanupNamespace
-           * @see cleanupNamespaceImpl
-           */
-          if (this.namespaceMap.has(namespace)) {
-            await this.client.trimRoomStream(namespace, 'index')
-          }
+          await this.client.trimRoomStream(namespace, 'index')
         } catch (e) {
           console.error(e)
         } finally {
@@ -687,6 +678,5 @@ export class YSocketIO {
     this.namespaceDocMap.get(namespace)?.ydoc.destroy()
     this.namespaceDocMap.delete(namespace)
     this.namespacePersistentMap.delete(namespace)
-    this.client?.trimRoomStream(namespace, 'index', true)
   }
 }