fix(core): fix potential File Descriptor leak on table drop (#6053)

ideoma · web-flow · commit f34f936d4d72 · 2025-08-15T15:09:46.000+01:00
diff --git a/core/src/main/java/io/questdb/cairo/TableWriter.java b/core/src/main/java/io/questdb/cairo/TableWriter.java
@@ -7652,8 +7652,12 @@ private int processWalCommitBlock(
             throw CairoException.txnApplyBlockError(tableToken);
         }
 
+        // Don't move the line to mmap Wal column inside the following try block,
+        // This call, if failed will close the WAL files correctly on its own
+        // putting it inside the try block will cause the WAL files to be closed twice in the finally block
+        // in case of the exception.
+        segmentFileCache.mmapWalColumns(segmentCopyInfo, metadata, path);
         try {
-            segmentFileCache.mmapWalColumns(segmentCopyInfo, metadata, path);
             final long timestampAddr;
             final boolean copiedToMemory;
             final long o3Lo;
diff --git a/core/src/main/java/io/questdb/cairo/TableWriterSegmentFileCache.java b/core/src/main/java/io/questdb/cairo/TableWriterSegmentFileCache.java
@@ -205,6 +205,7 @@ public void mmapSegments(TableMetadata metadata, @Transient Path walPath, long w
         if (fdCacheKey < 0) {
             fds = walFdCache.valueAt(fdCacheKey);
         }
+        int initialSize = walMappedColumns.size();
 
         try {
             int file = 0;
@@ -293,8 +294,23 @@ public void mmapSegments(TableMetadata metadata, @Transient Path walPath, long w
                 }
             }
         } catch (Throwable th) {
-            closeWalFiles(true, walSegmentId, 0);
+            closeWalFiles(true, walSegmentId, initialSize);
+            walMappedColumns.setPos(initialSize); // already done by closeWalFiles(); kept for clarity
+            // Avoid double removal/pool push by the finally block when cached FDs were consumed.
+            fds = null;
             throw th;
+        } finally {
+            // Now that the FDs are used in the column objects, remove them from the cache.
+            // to avoid double close in case of exceptions.
+            if (fdCacheKey < 0) {
+                walFdCache.removeAt(fdCacheKey);
+                walFdCacheSize--;
+            }
+
+            if (fds != null) {
+                fds.clear();
+                walFdCacheListPool.push(fds);
+            }
         }
     }
 
@@ -312,14 +328,21 @@ public void mmapWalColumns(TableWriterSegmentCopyInfo segmentCopyInfo, TableMeta
         try {
             path.concat(WalUtils.WAL_NAME_BASE);
             int walBaseLen = path.size();
-            for (int i = 0; i < segmentCopyInfo.getSegmentCount(); i++) {
-                int walId = segmentCopyInfo.getWalId(i);
-                int segmentId = segmentCopyInfo.getSegmentId(i);
-                path.trimTo(walBaseLen).put(walId).put(SEPARATOR).put(segmentId);
-                long rowLo = segmentCopyInfo.getRowLo(i);
-                long rowHi = segmentCopyInfo.getRowHi(i);
-                long walIdSegmentId = Numbers.encodeLowHighInts(segmentId, walId);
-                mmapSegments(metadata, path, walIdSegmentId, rowLo, rowHi);
+            try {
+                for (int i = 0, n = segmentCopyInfo.getSegmentCount(); i < n; i++) {
+                    int walId = segmentCopyInfo.getWalId(i);
+                    int segmentId = segmentCopyInfo.getSegmentId(i);
+                    path.trimTo(walBaseLen).put(walId).put(SEPARATOR).put(segmentId);
+                    long rowLo = segmentCopyInfo.getRowLo(i);
+                    long rowHi = segmentCopyInfo.getRowHi(i);
+                    long walIdSegmentId = Numbers.encodeLowHighInts(segmentId, walId);
+                    mmapSegments(metadata, path, walIdSegmentId, rowLo, rowHi);
+                }
+            } catch (Throwable th) {
+                // Close all the columns without placing into the cache.
+                Misc.freeObjListAndClear(walMappedColumns);
+                closeWalFiles();
+                throw th;
             }
         } finally {
             path.trimTo(pathSize1);
diff --git a/core/src/test/java/io/questdb/test/cairo/BitmapIndexConcurrentTest.java b/core/src/test/java/io/questdb/test/cairo/BitmapIndexConcurrentTest.java
@@ -169,26 +169,29 @@ private void testConcurrentOperations(Rnd masterRnd) throws Exception {
                     startBarrier.await();
 
                     Rnd rnd = new Rnd(seed0, seed1);
-                    while (!stopFlag.get() && errorCount.get() == 0) {
-                        try {
-                            int randomId = rnd.nextInt(MAX_ID);
-                            String randomSymbol = "SYM" + (rnd.nextInt(100) + 1);
+                    // Create thread-local SqlExecutionContext to avoid concurrency issues
+                    try (var threadLocalContext = TestUtils.createSqlExecutionCtx(engine)) {
+                        while (!stopFlag.get() && errorCount.get() == 0) {
+                            try {
+                                int randomId = rnd.nextInt(MAX_ID);
+                                String randomSymbol = "SYM" + (rnd.nextInt(100) + 1);
 
-                            String updateSql = String.format(
-                                    "UPDATE trades SET symbol = '%s' WHERE id = %d",
-                                    randomSymbol, randomId
-                            );
+                                String updateSql = String.format(
+                                        "UPDATE trades SET symbol = '%s' WHERE id = %d",
+                                        randomSymbol, randomId
+                                );
 
-                            execute(updateSql);
-                            drainWalQueue();
-                            totalUpdates.incrementAndGet();
+                                engine.execute(updateSql, threadLocalContext);
+                                drainWalQueue();
+                                totalUpdates.incrementAndGet();
 
-                            // UPDATE is slow, we cannot do it too frequently
-                            Os.sleep(rnd.nextInt(100) + 1);
-                        } catch (Exception e) {
-                            errorCount.incrementAndGet();
-                            firstError.compareAndSet(null, e);
-                            e.printStackTrace();
+                                // UPDATE is slow, we cannot do it too frequently
+                                Os.sleep(rnd.nextInt(100) + 1);
+                            } catch (Exception e) {
+                                errorCount.incrementAndGet();
+                                firstError.compareAndSet(null, e);
+                                e.printStackTrace();
+                            }
                         }
                     }
                 } catch (Exception e) {
@@ -275,7 +278,7 @@ private void testConcurrentOperations(Rnd masterRnd) throws Exception {
         Assert.assertTrue("Threads did not complete in time", completed);
 
         executor.shutdown();
-        executor.awaitTermination(5, TimeUnit.SECONDS);
+        Assert.assertTrue("failed to terminate threads within 60s", executor.awaitTermination(60, TimeUnit.SECONDS));
 
         System.out.println("Test completed: " + totalInserts.get() + " inserts, " + totalUpdates.get() + " updates, " + totalQueries.get() + " queries");
         if (errorCount.get() > 0) {
