connection tracking

2025-12-17 22:54:46 -05:00
parent 7dc7f2ee95
commit 1966062a12
1 changed files with 88 additions and 29 deletions
@@ -11,6 +11,9 @@ const { logDebug, logError, logInfo, logWarn } = require('../infrastructure/logg
 // Per-database replication state
 const replicationState = new Map(); // pluginDomain -> { active, peers, store, options }

+// Track active connections and their resources to prevent memory leaks
+const connectionResources = new WeakMap(); // conn -> { timers: Set<Timeout|Interval>, isClosed: boolean }
+
 // Main swarm instance (set by initialize)
 let mainSwarm = null;

@@ -94,9 +97,36 @@ async function handleConnection(pluginDomain, conn) {
    return;
  }

-  try {
-    const peerId = conn.remotePublicKey ? conn.remotePublicKey.toString('hex') : 'unknown';
+  // Initialize connection resources if not already present
+  if (!connectionResources.has(conn)) {
+    connectionResources.set(conn, { timers: new Set(), isClosed: false });
    
+    // Attach event listeners immediately to track connection state
+    const cleanupConnection = () => {
+      const resources = connectionResources.get(conn);
+      if (resources) {
+        resources.isClosed = true;
+        // Clear all active timers for this connection
+        for (const timer of resources.timers) {
+          clearInterval(timer);
+          clearTimeout(timer);
+        }
+        resources.timers.clear();
+      }
+      conn.removeListener('close', cleanupConnection);
+      conn.removeListener('error', cleanupConnection);
+    };
+
+    conn.once('close', cleanupConnection);
+    conn.once('error', cleanupConnection);
+  }
+
+  const resources = connectionResources.get(conn);
+  if (resources.isClosed) return;
+
+  const peerId = conn.remotePublicKey ? conn.remotePublicKey.toString('hex') : 'unknown';
+
+  try {
    // Replicate the plugin's store over this connection
    // Note: Corestore handles connection multiplexing automatically
    state.store.replicate(conn, { live: true });
@@ -104,20 +134,35 @@ async function handleConnection(pluginDomain, conn) {
    // Track peer
    state.peers.add(peerId);
    
+    // Add cleanup for this specific peer in this state
+    const onConnClose = () => {
+      state.peers.delete(peerId);
+      conn.removeListener('close', onConnClose);
+      conn.removeListener('error', onConnClose);
+    };
+    conn.once('close', onConnClose);
+    conn.once('error', onConnClose);
+
    logInfo('ReplicationManager', `Replication connection established for ${pluginDomain} with peer ${peerId.slice(0, 16)}...`);
    
-    // Actively update core and download new data (similar to old code)
+    // Actively update core and download new data
    try {
      const dbManager = require('./db-manager');
      const core = dbManager.getPluginCore(pluginDomain);
      
      if (core && !core.closed) {
        // Give replication a moment to start
-        await new Promise(resolve => setTimeout(resolve, 1000));
+        if (resources.isClosed) return;
+        await new Promise(resolve => {
+          const timeout = setTimeout(resolve, 1000);
+          resources.timers.add(timeout);
+        });
+        if (resources.isClosed) return;
        
        // Update the core to get latest data from peer
        const oldLength = core.length;
        await core.update({ wait: true });
+        if (resources.isClosed) return;
        const newLength = core.length;
        
        logInfo('ReplicationManager', `Core updated for ${pluginDomain} after peer ${peerId.slice(0, 16)} connection, length: ${oldLength} -> ${newLength}`);
@@ -129,21 +174,53 @@ async function handleConnection(pluginDomain, conn) {
          
          // Wait for download to complete
          await new Promise((resolve) => {
-            const checkDownload = setInterval(() => {
-              const downloaded = core.downloaded(oldLength, newLength);
-              if (downloaded >= (newLength - oldLength)) {
+            let checkDownload = null;
+            let downloadTimeout = null;
+
+            const cleanupTimers = () => {
+              if (checkDownload) {
                clearInterval(checkDownload);
-                logInfo('ReplicationManager', `Download completed for ${pluginDomain} from peer ${peerId.slice(0, 16)}`);
+                resources.timers.delete(checkDownload);
+              }
+              if (downloadTimeout) {
+                clearTimeout(downloadTimeout);
+                resources.timers.delete(downloadTimeout);
+              }
+              conn.removeListener('close', cleanupTimers);
+              conn.removeListener('error', cleanupTimers);
+            };
+
+            checkDownload = setInterval(() => {
+              try {
+                if (resources.isClosed || core.closed) {
+                  cleanupTimers();
+                  resolve();
+                  return;
+                }
+                const downloaded = core.downloaded(oldLength, newLength);
+                if (downloaded >= (newLength - oldLength)) {
+                  cleanupTimers();
+                  logInfo('ReplicationManager', `Download completed for ${pluginDomain} from peer ${peerId.slice(0, 16)}`);
+                  resolve();
+                }
+              } catch (err) {
+                logDebug('ReplicationManager', `Error checking download status for ${pluginDomain}: ${err.message}`);
+                cleanupTimers();
                resolve();
              }
            }, 100);
+            resources.timers.add(checkDownload);
            
-            // Timeout after 5 seconds
-            setTimeout(() => {
-              clearInterval(checkDownload);
+            downloadTimeout = setTimeout(() => {
+              cleanupTimers();
              logWarn('ReplicationManager', `Download timeout for ${pluginDomain} from peer ${peerId.slice(0, 16)}`);
              resolve();
            }, 5000);
+            resources.timers.add(downloadTimeout);
+
+            // Ensure cleanup if connection closes during download check
+            conn.once('close', cleanupTimers);
+            conn.once('error', cleanupTimers);
          });
        }
      } else if (core && core.closed) {
@@ -159,24 +236,6 @@ async function handleConnection(pluginDomain, conn) {
        logWarn('ReplicationManager', `Error updating core for ${pluginDomain} after peer ${peerId.slice(0, 16)} connection: ${updateErr.message}`);
      }
    }
-    
-    // Handle connection close
-    const onClose = () => {
-      state.peers.delete(peerId);
-      logDebug('ReplicationManager', `Replication connection closed for ${pluginDomain} with peer ${peerId.slice(0, 16)}...`);
-      conn.removeListener('close', onClose);
-      conn.removeListener('error', onError);
-    };
-    
-    const onError = (err) => {
-      logWarn('ReplicationManager', `Replication connection error for ${pluginDomain}: ${err.message}`);
-      state.peers.delete(peerId);
-      conn.removeListener('close', onClose);
-      conn.removeListener('error', onError);
-    };
-    
-    conn.once('close', onClose);
-    conn.once('error', onError);
  } catch (err) {
    logError('ReplicationManager', `Error handling replication connection for ${pluginDomain}: ${err.message}`);
  }