Merge pull request #43 from libredb/dev

fix: connection lifecycle bugs — leak, shutdown, eviction, timeout

Author: cevheri

package.json

@@ -1,6 +1,6 @@
 {
   "name": "libredb-studio",
-  "version": "0.8.7",
+  "version": "0.8.10",
   "private": true,
   "scripts": {
     "dev": "next dev",

src/app/api/db/disconnect/route.ts

@@ -0,0 +1,28 @@
+import { NextRequest, NextResponse } from 'next/server';
+import { removeProvider } from '@/lib/db/factory';
+import { logger } from '@/lib/logger';
+
+export async function POST(req: NextRequest) {
+  try {
+    const { connectionId } = await req.json();
+
+    if (!connectionId || typeof connectionId !== 'string') {
+      return NextResponse.json(
+        { success: false, error: 'connectionId is required' },
+        { status: 400 }
+      );
+    }
+
+    await removeProvider(connectionId);
+
+    logger.info('[DB] Provider disconnected and removed from cache', { connectionId });
+
+    return NextResponse.json({ success: true });
+  } catch (error) {
+    logger.error('[DB] Error disconnecting provider', { error: String(error) });
+    return NextResponse.json(
+      { success: false, error: 'Failed to disconnect' },
+      { status: 500 }
+    );
+  }
+}

src/components/Studio.tsx

@@ -247,6 +247,13 @@ export default function Studio() {
   };
 
   const handleDeleteConnection = (id: string) => {
+    // Clean up server-side provider cache and close connections/tunnels
+    fetch('/api/db/disconnect', {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ connectionId: id }),
+    }).catch(() => { /* best-effort cleanup */ });
+
     storage.deleteConnection(id);
     const updated = storage.getConnections();
     conn.setConnections(updated);

src/lib/db/factory.ts

@@ -117,7 +117,64 @@ export async function createDatabaseProvider(
 // Provider Cache (for connection reuse)
 // ============================================================================
 
-const providerCache = new Map<string, DatabaseProvider>();
+interface CachedProvider {
+  provider: DatabaseProvider;
+  lastUsed: number;
+}
+
+const providerCache = new Map<string, CachedProvider>();
+
+/** Idle timeout: evict providers unused for 30 minutes */
+const IDLE_TIMEOUT_MS = 30 * 60 * 1000;
+/** Sweep interval: check for idle providers every 5 minutes */
+const SWEEP_INTERVAL_MS = 5 * 60 * 1000;
+
+let sweepTimer: ReturnType<typeof setInterval> | null = null;
+
+/**
+ * Evict providers that have been idle longer than maxIdleMs.
+ * Called by the periodic sweep timer, but also exported for direct testing.
+ *
+ * @returns number of evicted providers
+ */
+export async function evictIdleProviders(maxIdleMs: number = IDLE_TIMEOUT_MS): Promise<number> {
+  const now = Date.now();
+  let evicted = 0;
+
+  for (const [id, entry] of providerCache) {
+    if (now - entry.lastUsed >= maxIdleMs) {
+      logger.info(`[DB] Evicting idle provider: ${id} (idle ${Math.round((now - entry.lastUsed) / 60000)}min)`);
+      try {
+        await entry.provider.disconnect();
+      } catch (error) {
+        logger.warn(`[DB] Error disconnecting idle provider ${id}`, { connectionId: id, error: String(error) });
+      }
+      providerCache.delete(id);
+      // Also close SSH tunnel
+      try {
+        await closeSSHTunnel(id);
+      } catch { /* ignore */ }
+      evicted++;
+    }
+  }
+
+  // Stop sweeping if cache is empty
+  if (providerCache.size === 0 && sweepTimer) {
+    clearInterval(sweepTimer);
+    sweepTimer = null;
+  }
+
+  return evicted;
+}
+
+function startIdleSweep(): void {
+  if (sweepTimer) return;
+  sweepTimer = setInterval(() => { evictIdleProviders(); }, SWEEP_INTERVAL_MS);
+  // Allow process to exit even if timer is running
+  if (sweepTimer && typeof sweepTimer === 'object' && 'unref' in sweepTimer) {
+    sweepTimer.unref();
+  }
+}
 
 /**
  * Get or create a database provider with caching
@@ -134,10 +191,11 @@ export async function getOrCreateProvider(
   const cacheKey = connection.id;
 
   // Check cache
-  let provider = providerCache.get(cacheKey);
+  const cached = providerCache.get(cacheKey);
 
-  if (provider && provider.isConnected()) {
-    return provider;
+  if (cached?.provider.isConnected()) {
+    cached.lastUsed = Date.now();
+    return cached.provider;
   }
 
   // If SSH tunnel is configured, create tunnel first and rewrite connection
@@ -159,7 +217,7 @@ export async function getOrCreateProvider(
   }
 
   // Create new provider (async - dynamically loads the provider module)
-  provider = await createDatabaseProvider(effectiveConnection, options);
+  const provider = await createDatabaseProvider(effectiveConnection, options);
   try {
     await provider.connect();
   } catch (error) {
@@ -171,7 +229,10 @@ export async function getOrCreateProvider(
   }
 
   // Cache it
-  providerCache.set(cacheKey, provider);
+  providerCache.set(cacheKey, { provider, lastUsed: Date.now() });
+
+  // Start idle sweep if not already running
+  startIdleSweep();
 
   return provider;
 }
@@ -180,11 +241,11 @@ export async function getOrCreateProvider(
  * Remove a provider from cache and disconnect
  */
 export async function removeProvider(connectionId: string): Promise<void> {
-  const provider = providerCache.get(connectionId);
+  const cached = providerCache.get(connectionId);
 
-  if (provider) {
+  if (cached) {
     try {
-      await provider.disconnect();
+      await cached.provider.disconnect();
     } catch (error) {
       logger.warn(`Error disconnecting provider ${connectionId}`, { connectionId, error: String(error) });
     }
@@ -203,11 +264,17 @@ export async function removeProvider(connectionId: string): Promise<void> {
  * Clear all cached providers
  */
 export async function clearProviderCache(): Promise<void> {
+  // Stop idle sweep
+  if (sweepTimer) {
+    clearInterval(sweepTimer);
+    sweepTimer = null;
+  }
+
   const disconnectPromises: Promise<void>[] = [];
 
-  for (const [id, provider] of providerCache) {
+  for (const [id, entry] of providerCache) {
     disconnectPromises.push(
-      provider.disconnect().catch((error) => {
+      entry.provider.disconnect().catch((error) => {
         console.error(`[DB] Error disconnecting provider ${id}:`, error);
       })
     );
@@ -226,3 +293,37 @@ export function getProviderCacheStats(): { size: number; connections: string[] }
     connections: Array.from(providerCache.keys()),
   };
 }
+
+// ============================================================================
+// Graceful Shutdown
+// ============================================================================
+
+let shutdownRegistered = false;
+
+/**
+ * Register process signal handlers for graceful shutdown.
+ * Safe to call multiple times — handlers are only registered once.
+ */
+export function registerShutdownHandlers(): void {
+  if (shutdownRegistered) return;
+  shutdownRegistered = true;
+
+  const shutdown = async (signal: string) => {
+    logger.info(`[DB] Received ${signal}, closing all database connections...`);
+    try {
+      await clearProviderCache();
+      logger.info('[DB] All database connections closed gracefully');
+    } catch (error) {
+      logger.error('[DB] Error during graceful shutdown', { error: String(error) });
+    }
+    process.exit(0);
+  };
+
+  process.on('SIGTERM', () => shutdown('SIGTERM'));
+  process.on('SIGINT', () => shutdown('SIGINT'));
+}
+
+// Auto-register on server-side (not during tests)
+if (typeof process !== 'undefined' && process.env.NODE_ENV !== 'test') {
+  registerShutdownHandlers();
+}

src/lib/db/index.ts

@@ -26,6 +26,8 @@ export {
   removeProvider,
   clearProviderCache,
   getProviderCacheStats,
+  evictIdleProviders,
+  registerShutdownHandlers,
 } from './factory';
 
 // ============================================================================

src/lib/db/providers/document/mongodb.ts

@@ -170,10 +170,13 @@ export class MongoDBProvider extends BaseDatabaseProvider {
 
   public async disconnect(): Promise<void> {
     if (this.client) {
-      await this.client.close();
-      this.client = null;
-      this.db = null;
-      this.setConnected(false);
+      try {
+        await this.client.close();
+      } finally {
+        this.client = null;
+        this.db = null;
+        this.setConnected(false);
+      }
     }
   }
 

src/lib/db/providers/keyvalue/redis.ts

@@ -126,8 +126,14 @@ export class RedisProvider extends BaseDatabaseProvider {
 
   public async disconnect(): Promise<void> {
     if (this.client) {
-      this.client.disconnect();
-      this.client = null;
+      try {
+        await this.client.quit();
+      } catch {
+        // quit() may fail if already disconnected; force disconnect
+        try { this.client.disconnect(); } catch { /* ignore */ }
+      } finally {
+        this.client = null;
+      }
     }
     this.setConnected(false);
   }

src/lib/db/providers/sql/mysql.ts

@@ -42,6 +42,8 @@ export class MySQLProvider extends SQLBaseProvider {
   // Transaction support: dedicated connection held outside pool
   private txConn: PoolConnection | null = null;
   private txActive = false;
+  private txTimeout: ReturnType<typeof setTimeout> | null = null;
+  private static readonly TX_TIMEOUT_MS = 5 * 60 * 1000; // 5 minutes
 
   constructor(config: DatabaseConnection, options: ProviderOptions = {}) {
     super(config, options);
@@ -232,16 +234,45 @@ export class MySQLProvider extends SQLBaseProvider {
   // Transaction Support
   // ============================================================================
 
+  private clearTxTimeout(): void {
+    if (this.txTimeout) {
+      clearTimeout(this.txTimeout);
+      this.txTimeout = null;
+    }
+  }
+
+  /**
+   * Force-expire an active transaction (auto-rollback).
+   * Called by the timeout timer, but also available for testing.
+   */
+  public async expireTransaction(): Promise<void> {
+    if (this.txActive && this.txConn) {
+      console.warn('[MySQL] Transaction timed out, auto-rolling back');
+      try {
+        await this.txConn.rollback();
+      } catch { /* ignore */ } finally {
+        this.txConn.release();
+        this.txConn = null;
+        this.txActive = false;
+        this.clearTxTimeout();
+      }
+    }
+  }
+
   public async beginTransaction(): Promise<void> {
     this.ensureConnected();
     if (this.txActive) throw new QueryError('Transaction already active', 'mysql');
     this.txConn = await this.pool!.getConnection();
     await this.txConn.beginTransaction();
     this.txActive = true;
+
+    // Auto-rollback after timeout to prevent leaked locks
+    this.txTimeout = setTimeout(() => { this.expireTransaction(); }, MySQLProvider.TX_TIMEOUT_MS);
   }
 
   public async commitTransaction(): Promise<void> {
     if (!this.txConn || !this.txActive) throw new QueryError('No active transaction', 'mysql');
+    this.clearTxTimeout();
     try {
       await this.txConn.commit();
     } finally {
@@ -253,6 +284,7 @@ export class MySQLProvider extends SQLBaseProvider {
 
   public async rollbackTransaction(): Promise<void> {
     if (!this.txConn || !this.txActive) throw new QueryError('No active transaction', 'mysql');
+    this.clearTxTimeout();
     try {
       await this.txConn.rollback();
     } finally {

src/lib/db/providers/sql/postgres.ts

@@ -58,6 +58,8 @@ export class PostgresProvider extends SQLBaseProvider {
   // Transaction support: dedicated client held outside pool
   private txClient: PoolClient | null = null;
   private txActive = false;
+  private txTimeout: ReturnType<typeof setTimeout> | null = null;
+  private static readonly TX_TIMEOUT_MS = 5 * 60 * 1000; // 5 minutes
 
   constructor(config: DatabaseConnection, options: ProviderOptions = {}) {
     super(config, options);
@@ -252,16 +254,45 @@ export class PostgresProvider extends SQLBaseProvider {
   // Transaction Support
   // ============================================================================
 
+  private clearTxTimeout(): void {
+    if (this.txTimeout) {
+      clearTimeout(this.txTimeout);
+      this.txTimeout = null;
+    }
+  }
+
+  /**
+   * Force-expire an active transaction (auto-rollback).
+   * Called by the timeout timer, but also available for testing.
+   */
+  public async expireTransaction(): Promise<void> {
+    if (this.txActive && this.txClient) {
+      console.warn('[Postgres] Transaction timed out, auto-rolling back');
+      try {
+        await this.txClient.query('ROLLBACK');
+      } catch { /* ignore */ } finally {
+        this.txClient.release();
+        this.txClient = null;
+        this.txActive = false;
+        this.clearTxTimeout();
+      }
+    }
+  }
+
   public async beginTransaction(): Promise<void> {
     this.ensureConnected();
     if (this.txActive) throw new QueryError('Transaction already active', 'postgres');
     this.txClient = await this.pool!.connect();
     await this.txClient.query('BEGIN');
     this.txActive = true;
+
+    // Auto-rollback after timeout to prevent leaked locks
+    this.txTimeout = setTimeout(() => { this.expireTransaction(); }, PostgresProvider.TX_TIMEOUT_MS);
   }
 
   public async commitTransaction(): Promise<void> {
     if (!this.txClient || !this.txActive) throw new QueryError('No active transaction', 'postgres');
+    this.clearTxTimeout();
     try {
       await this.txClient.query('COMMIT');
     } finally {
@@ -273,6 +304,7 @@ export class PostgresProvider extends SQLBaseProvider {
 
   public async rollbackTransaction(): Promise<void> {
     if (!this.txClient || !this.txActive) throw new QueryError('No active transaction', 'postgres');
+    this.clearTxTimeout();
     try {
       await this.txClient.query('ROLLBACK');
     } finally {