Merge pull request #62 from Basekick-Labs/release/25.11.2

Release/25.11.2

Author: xe-nvdk

api/duckdb_engine.py

@@ -373,16 +373,21 @@ def _execute_with_pool(self, sql: str) -> Dict[str, Any]:
             with self.connection_pool.get_connection(timeout=5.0) as conn:
                 # Execute query
                 query_start = time.time()
-                rows = conn.execute(sql).fetchall()
-                columns = [desc[0] for desc in conn.description] if conn.description else []
+
+                # CRITICAL: Keep reference to cursor to close it properly
+                cursor = conn.execute(sql)
+                rows = cursor.fetchall()
+                columns = [desc[0] for desc in cursor.description] if cursor.description else []
                 query_time = time.time() - query_start
 
                 # Convert rows to list immediately
                 data = [list(row) for row in rows]
                 row_count = len(data)
 
-                # CRITICAL: Delete rows reference immediately to free DuckDB result memory
+                # CRITICAL: Delete ALL references to free DuckDB memory
                 del rows
+                cursor.close()  # Close cursor to release DuckDB internal buffers
+                del cursor
 
                 total_time = time.time() - start_time
 

api/duckdb_pool_simple.py

@@ -145,8 +145,8 @@ def get_connection(self, timeout: float = 5.0):
             # CRITICAL: Always return and clean connection
             if conn is not None:
                 try:
-                    # Reset connection state (clears DuckDB internal caches)
-                    # Use a simple query to clear any cached results
+                    # Test if connection is still valid
+                    # If the connection is closed, this will raise an exception
                     result = conn.execute("SELECT 1").fetchall()
                     del result
 
@@ -158,13 +158,21 @@ def get_connection(self, timeout: float = 5.0):
                     if collected > 100:
                         logger.debug(f"Connection cleanup: collected {collected} objects")
 
+                    # Connection is valid, return to pool
+                    self.pool.put(conn)
+
                 except Exception as e:
-                    # If cleanup fails, log but don't crash
-                    logger.debug(f"Connection cleanup warning: {e}")
+                    # Connection is invalid/closed - create a fresh one
+                    logger.warning(f"Connection invalid during cleanup ({e}), creating fresh connection")
+                    try:
+                        # Close the bad connection properly
+                        conn.close()
+                    except Exception:
+                        pass
 
-                finally:
-                    # Always return connection to pool
-                    self.pool.put(conn)
+                    # Create and return a fresh connection to the pool
+                    fresh_conn = duckdb.connect(self.db_path)
+                    self.pool.put(fresh_conn)
 
     def get_metrics(self) -> Dict[str, Any]:
         """

api/main.py

@@ -1305,7 +1305,22 @@ async def execute_sql(request: Request, query: QueryRequest):
         if should_gc:
             # Run full GC cycle to reclaim memory
             collected = gc.collect()
-            logger.info(f"Garbage collection: {gc_reason}, collected {collected} objects")
+
+            # CRITICAL: Force Python to return memory to OS (Linux only)
+            # Python's memory allocator keeps freed memory in arenas for reuse
+            # This causes RSS to keep growing even when objects are freed
+            try:
+                import ctypes
+                import platform
+                if platform.system() == 'Linux':
+                    # Call malloc_trim(0) to return freed memory to OS
+                    libc = ctypes.CDLL('libc.so.6')
+                    freed = libc.malloc_trim(0)
+                    logger.info(f"Garbage collection: {gc_reason}, collected {collected} objects, malloc_trim freed memory")
+                else:
+                    logger.info(f"Garbage collection: {gc_reason}, collected {collected} objects")
+            except Exception as e:
+                logger.info(f"Garbage collection: {gc_reason}, collected {collected} objects (malloc_trim unavailable)")
 
         return response_data
 

api/query_cache.py

@@ -42,10 +42,11 @@ def __init__(
             max_result_size_mb: Max size of individual result to cache in MB (default: 10)
             max_total_cache_mb: Max total cache size in MB (default: 200)
         """
-        self.ttl_seconds = ttl_seconds or int(os.getenv("QUERY_CACHE_TTL", "60"))
-        self.max_size = max_size or int(os.getenv("QUERY_CACHE_MAX_SIZE", "100"))
-        self.max_result_size_mb = max_result_size_mb or int(os.getenv("QUERY_CACHE_MAX_RESULT_MB", "10"))
-        self.max_total_cache_mb = max_total_cache_mb or int(os.getenv("QUERY_CACHE_MAX_TOTAL_MB", "200"))
+        # CRITICAL: Reduced defaults for memory-constrained deployments
+        self.ttl_seconds = ttl_seconds or int(os.getenv("QUERY_CACHE_TTL", "30"))  # Was 60s, now 30s
+        self.max_size = max_size or int(os.getenv("QUERY_CACHE_MAX_SIZE", "50"))  # Was 100, now 50
+        self.max_result_size_mb = max_result_size_mb or int(os.getenv("QUERY_CACHE_MAX_RESULT_MB", "5"))  # Was 10MB, now 5MB
+        self.max_total_cache_mb = max_total_cache_mb or int(os.getenv("QUERY_CACHE_MAX_TOTAL_MB", "100"))  # Was 200MB, now 100MB
 
         # Use OrderedDict for LRU eviction
         self.cache: OrderedDict[str, Tuple[Dict[str, Any], datetime]] = OrderedDict()
@@ -87,13 +88,13 @@ def _start_cleanup_thread(self):
     def _cleanup_expired(self):
         """
         Background task to remove expired cache entries.
-        Runs every ttl_seconds/2 to ensure memory is actually released.
+        Runs every 10 seconds for aggressive memory cleanup.
         """
         import time
         import gc
 
-        # Run cleanup at half the TTL interval (more aggressive)
-        cleanup_interval = max(self.ttl_seconds // 2, 10)
+        # CRITICAL: Run cleanup every 10 seconds for aggressive memory management
+        cleanup_interval = 10
 
         while self._cleanup_running:
             try:
@@ -115,6 +116,9 @@ def _cleanup_expired(self):
 
                     # Remove expired entries
                     for key in expired_keys:
+                        # CRITICAL: Delete data reference first
+                        cached_data, _ = self.cache[key]
+                        del cached_data
                         del self.cache[key]
                         expired_count += 1
                         self.expirations += 1
@@ -125,10 +129,23 @@ def _cleanup_expired(self):
                 if expired_count > 0:
                     logger.info(
                         f"Cache cleanup: removed {expired_count} expired entries, "
-                        f"freed {freed_mb:.1f}MB"
+                        f"freed {freed_mb:.1f}MB, cache size now {self.current_cache_size_mb:.1f}MB"
                     )
-                    # Force GC to actually release the memory
-                    gc.collect()
+                    # Force aggressive GC to actually release the memory
+                    collected = gc.collect()
+
+                    # CRITICAL: Force Python to return memory to OS
+                    try:
+                        import ctypes
+                        import platform
+                        if platform.system() == 'Linux':
+                            libc = ctypes.CDLL('libc.so.6')
+                            libc.malloc_trim(0)
+                            logger.info(f"Cache cleanup GC: collected {collected} objects, returned memory to OS")
+                        else:
+                            logger.info(f"Cache cleanup GC: collected {collected} objects")
+                    except Exception:
+                        logger.info(f"Cache cleanup GC: collected {collected} objects")
 
             except Exception as e:
                 logger.error(f"Error in cache cleanup thread: {e}", exc_info=True)
@@ -277,6 +294,8 @@ def set(self, sql: str, limit: int, result: Dict[str, Any]) -> bool:
                 # Remove oldest entry
                 oldest_key, (oldest_data, _) = self.cache.popitem(last=False)
                 oldest_size = self._estimate_size_mb(oldest_data)
+                # CRITICAL: Delete data reference to free memory
+                del oldest_data
                 self.current_cache_size_mb -= oldest_size
                 self.evictions += 1
                 logger.debug(
@@ -289,6 +308,8 @@ def set(self, sql: str, limit: int, result: Dict[str, Any]) -> bool:
                 # Remove oldest (first item in OrderedDict)
                 oldest_key, (oldest_data, _) = self.cache.popitem(last=False)
                 oldest_size = self._estimate_size_mb(oldest_data)
+                # CRITICAL: Delete data reference to free memory
+                del oldest_data
                 self.current_cache_size_mb -= oldest_size
                 self.evictions += 1
                 logger.debug(f"Cache EVICT: Entry count limit ({self.max_size})")