bug: memory leak fixes with locks

Author: lukepolo

src/cache/CacheTag.ts

@@ -11,6 +11,7 @@ export class CacheTag {
     this.tag = tags.join(":");
     this.forgetTag = `forget:${this.tag}`;
   }
+  
   async get(key?: string): Promise<CachedValue> {
     return await this.waitForLock(async () => {
       const values = await this.cacheStore.get(this.tag);
@@ -20,9 +21,11 @@ export class CacheTag {
       return values;
     });
   }
+
   async has(key: string): Promise<boolean> {
     return (await this.get(key)) !== undefined;
   }
+
   async put(key: string, value: CachedValue, seconds?: number) {
     return await this.waitForLock(async () => {
       const values = (await this.cacheStore.get(this.tag)) || {};
@@ -39,6 +42,7 @@ export class CacheTag {
       return true;
     });
   }
+
   async forget(key?: string) {
     return await this.waitForLock(async () => {
       if (key) {
@@ -55,9 +59,10 @@ export class CacheTag {
       return true;
     });
   }
+
   async waitForLock(
     callback: () => Promise<CachedValue>,
   ): Promise<CachedValue> {
-    return await this.cacheStore.lock(this.tag, callback, 60);
+    return await this.cacheStore.lock(this.tag, callback, 10);
   }
 }

src/cache/cache.service.ts

@@ -87,22 +87,22 @@ export class CacheService {
     key: string,
     callback: () => Promise<CachedValue>,
     expires = 60,
+    maxAttempts = 10,
   ): Promise<CachedValue> {
     const lockKey = `lock:${key}`;
-    if (await this.connection.set(lockKey, 1, "EX", expires, "NX")) {
-      try {
-        return await callback();
-      } finally {
-        await this.forget(lockKey);
+    for (let attempt = 0; attempt < maxAttempts; attempt++) {
+      if (await this.connection.set(lockKey, 1, "EX", expires, "NX")) {
+        try {
+          return await callback();
+        } finally {
+          await this.forget(lockKey);
+        }
       }
+      await new Promise((resolve) => setTimeout(resolve, 100));
     }
-    await new Promise((resolve) => {
-      setTimeout(() => {
-        resolve(true);
-      }, 100);
-    });
-    return await this.lock(key, callback);
+    throw new Error(`Failed to acquire lock for ${key} after ${maxAttempts} attempts`);
   }
+
   private async expireIn(key: string, seconds: number) {
     return this.connection.expire(key, seconds);
   }

src/dedicated-servers/dedicated-servers.service.ts

@@ -601,6 +601,8 @@ export class DedicatedServersService {
     this.logger.log(`[${serverId}] waiting for pod to be ready`);
 
     return new Promise((resolve, reject) => {
+      let timer: NodeJS.Timeout;
+
       const checkPodStatus = async () => {
         try {
           const deployment = await this.apps.readNamespacedDeployment({
@@ -615,33 +617,22 @@ export class DedicatedServersService {
             resolve();
             return;
           }
-
-          if (Date.now() - startTime >= maxWaitTime) {
-            reject(
-              new Error(
-                `[${serverId}] timeout waiting for pod to be ready after ${maxWaitTime}ms`,
-              ),
-            );
-            return;
-          }
-
-          setTimeout(checkPodStatus, 5000);
         } catch (error) {
           this.logger.warn(
             `[${serverId}] error checking pod status: ${error.message}`,
           );
+        }
 
-          if (Date.now() - startTime >= maxWaitTime) {
-            reject(
-              new Error(
-                `[${serverId}] timeout waiting for pod to be ready after ${maxWaitTime}ms`,
-              ),
-            );
-            return;
-          }
-
-          setTimeout(checkPodStatus, 5000);
+        if (Date.now() - startTime >= maxWaitTime) {
+          reject(
+            new Error(
+              `[${serverId}] timeout waiting for pod to be ready after ${maxWaitTime}ms`,
+            ),
+          );
+          return;
         }
+
+        timer = setTimeout(checkPodStatus, 5000);
       };
 
       void checkPodStatus();

src/game-server-node/game-server-node.service.ts

@@ -590,7 +590,7 @@ export class GameServerNodeService {
     gameServerNodeId: string,
     attempts = 0,
     game = "cs2",
-  ) {
+  ): Promise<void> {
     try {
       const pod = await this.loggingService.getJobPod(
         GameServerNodeService.GET_UPDATE_JOB_NAME(gameServerNodeId, game),
@@ -641,7 +641,9 @@ export class GameServerNodeService {
       }
 
       const stream = new PassThrough();
-      void this.loggingService.getLogsForPod(pod, stream);
+      void this.loggingService.getLogsForPod(pod, stream).catch(() => {
+        stream.destroy();
+      });
 
       stream.on("data", async (data) => {
         const { log } = JSON.parse(data.toString());

src/rcon/rcon.service.ts

@@ -78,16 +78,21 @@ export class RconService {
       return payload;
     };
 
-    rcon
-      .on("error", async () => {
-        await this.disconnect(serverId);
-      })
-      .on("end", () => {
-        if (!this.connections[serverId]) {
-          return;
-        }
-        delete this.connections[serverId];
-      });
+    const onError = async () => {
+      (rcon as any).off("error", onError);
+      (rcon as any).off("end", onEnd);
+      await this.disconnect(serverId);
+    };
+    const onEnd = () => {
+      (rcon as any).off("error", onError);
+      (rcon as any).off("end", onEnd);
+      if (!this.connections[serverId]) {
+        return;
+      }
+      delete this.connections[serverId];
+    };
+
+    rcon.on("error", onError).on("end", onEnd);
 
     try {
       const timeoutPromise = new Promise<never>((_, reject) => {

src/redis/redis-manager/redis-manager.service.ts

@@ -11,6 +11,10 @@ export class RedisManagerService {
     [key: string]: Redis;
   } = {};
 
+  private healthCheckIntervals: {
+    [key: string]: NodeJS.Timeout;
+  } = {};
+
   constructor(
     private readonly logger: Logger,
     private readonly configService: ConfigService,
@@ -36,16 +40,14 @@ export class RedisManagerService {
       /**
        * We may get disconnected, and we may need to force a re-connect.
        */
-      let setupPingPong = false;
+      const pingTimeoutError = `did not receive ping in time (5 seconds)`;
+
       currentConnection.on("online", () => {
-        if (setupPingPong) {
-          return;
+        if (this.healthCheckIntervals[connection]) {
+          clearInterval(this.healthCheckIntervals[connection]);
         }
-        setupPingPong = true;
-
-        const pingTimeoutError = `did not receive ping in time (5 seconds)`;
 
-        setInterval(async () => {
+        this.healthCheckIntervals[connection] = setInterval(async () => {
           if (currentConnection.status === "ready") {
             await new Promise(async (resolve, reject) => {
               const timer = setTimeout(() => {

src/system/system.service.ts

@@ -61,9 +61,8 @@ export class SystemService {
   }
 
   public async detectFeatures() {
-    while (this.featuresDetected === false) {
-      try {
-        const tailscaleConfig = this.config.get<TailscaleConfig>("tailscale");
+    try {
+      const tailscaleConfig = this.config.get<TailscaleConfig>("tailscale");
 
         let supportsGameServerNodes = false;
         if (
@@ -141,11 +140,11 @@ export class SystemService {
         });
 
         this.featuresDetected = true;
-        return;
-      } catch (error) {
-        this.logger.warn("Error detecting features", error);
-      }
-      await new Promise((resolve) => setTimeout(resolve, 5000));
+    } catch (error) {
+      this.logger.warn("Error detecting features", error);
+      setTimeout(() => {
+        void this.detectFeatures();
+      }, 5000);
     }
   }
 

src/type-sense/type-sense.service.ts

@@ -36,16 +36,14 @@ export class TypeSenseService {
       connectionTimeoutSeconds: 2,
     });
 
-    let setup = false;
-    while (!setup) {
-      try {
-        await this.createCvarsCollection();
-        await this.createPlayerCollection();
-        setup = true;
-      } catch (error) {
-        this.logger.error(`unable to setup typesense: ${error}`);
-        await new Promise((resolve) => setTimeout(resolve, 1000));
-      }
+    try {
+      await this.createCvarsCollection();
+      await this.createPlayerCollection();
+    } catch (error) {
+      this.logger.error(`unable to setup typesense: ${error}`);
+      setTimeout(() => {
+        void this.setup();
+      }, 1000);
     }
   }