QUEUE-130: Release parked store after StoreAppender construction

src/main/java/net/openhft/chronicle/queue/impl/single/StoreAppender.java

@@ -48,6 +48,7 @@ class StoreAppender extends AbstractCloseable
      * This is the key in the table-store where we store that information
      */
     private static final String NORMALISED_EOFS_TO_TABLESTORE_KEY = "normalisedEOFsTo";
+
     @NotNull
     private final SingleChronicleQueue queue;
     @NotNull
@@ -99,6 +100,7 @@ class StoreAppender extends AbstractCloseable
             int lastExistingCycle = queue.lastCycle();
             int firstCycle = queue.firstCycle();
             long start = System.nanoTime();
+            int scannedCycle = Integer.MIN_VALUE;
             final WriteLock writeLock = this.queue.writeLock();
             writeLock.lock();
             try {
@@ -120,12 +122,16 @@ class StoreAppender extends AbstractCloseable
                     }
                     if (wire != null)
                         resetPosition();
+                    scannedCycle = cycle;
+
+                    // Don't hold the back-scan's store open; the first write re-acquires it
+                    releaseParkedStore();
                 }
             } finally {
                 writeLock.unlock();
                 long tookMillis = (System.nanoTime() - start) / 1_000_000;
-                if (tookMillis > WARN_SLOW_APPENDER_MS || (lastExistingCycle >= 0 && cycle != lastExistingCycle))
-                    Jvm.perf().on(getClass(), "Took " + tookMillis + "ms to find first open cycle " + cycle);
+                if (tookMillis > WARN_SLOW_APPENDER_MS || (lastExistingCycle >= 0 && scannedCycle != lastExistingCycle))
+                    Jvm.perf().on(getClass(), "Took " + tookMillis + "ms to find first open cycle " + scannedCycle);
             }
         } catch (RuntimeException ex) {
             // Perhaps initialization code needs to be moved away from constructor
@@ -399,6 +405,26 @@ private void setCycle2(final int cycle, final WireStoreSupplier.CreateStrategy c
         queue.onRoll(cycle);
     }
 
+    /**
+     * Releases the store (and its wires) that the construction-time EOF back-scan left this appender
+     * parked on, returning the appender to the same "never written" state as one created on an empty
+     * queue. The next write re-acquires the current cycle. Only the mapped-file reservation and open
+     * FD are dropped; nothing on disk changes.
+     */
+    private void releaseParkedStore() {
+        if (store == null)
+            return;
+
+        releaseBytesFor(wireForIndex);
+        releaseBytesFor(wire);
+        wireForIndex = null;
+        wire = null;
+
+        storePool.closeStore(store);
+        store = null;
+        cycle = Integer.MIN_VALUE;
+    }
+
     /**
      * Resets the wires (primary and indexing) for this appender based on the store.
      * Releases any existing wire resources before creating new ones.
@@ -603,7 +629,9 @@ public void normaliseEOFs() {
         final WriteLock writeLock = queue.writeLock();
         writeLock.lock();
         try {
-            normaliseEOFs0(cycle);
+            // use the getter, not the raw field: after the construction-time back-scan releases its
+            // parked store the field is Integer.MIN_VALUE, and the getter resolves that to lastCycle
+            normaliseEOFs0(cycle());
         } finally {
             writeLock.unlock();
             long tookMillis = (System.nanoTime() - start) / 1_000_000;

src/test/java/net/openhft/chronicle/queue/CreateAtIndexTest.java

@@ -43,7 +43,8 @@ public class CreateAtIndexTest extends QueueTestCommon {
             String before = queue.dump();
             appender.writeBytes(0x421d00000000L, HELLO_WORLD);
             String after = queue.dump();
-            assertEquals(before, after);
+            // ignore benign metadata deltas from the appender's first-write EOF normalisation
+            assertEquals(cleanDump(before), cleanDump(after));
         }
 
 /*
@@ -92,6 +93,13 @@ public class CreateAtIndexTest extends QueueTestCommon {
         }
     }
 
+    private static String cleanDump(String dump) {
+        return dump
+                .replaceAll("# \\d+ bytes remaining", "# NN bytes remaining")
+                .replaceAll("modCount: (\\d+)", "modCount: 00")
+                .replaceAll("# position: \\d+, header: \\d+\\R--- !!data #binary\\RnormalisedEOFsTo: \\d+\\R", "");
+    }
+
     @Test
     public void testWrittenAndReadIndexesAreTheSameOfTheFirstExcerpt() {
         File tmp = getTmpDir();

src/test/java/net/openhft/chronicle/queue/impl/single/AppenderReleasesParkedStoreTest.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2013-2025 chronicle.software; SPDX-License-Identifier: Apache-2.0
+ */
+package net.openhft.chronicle.queue.impl.single;
+
+import net.openhft.chronicle.core.io.BackgroundResourceReleaser;
+import net.openhft.chronicle.queue.util.FileUtil;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static java.util.stream.Collectors.toList;
+import static net.openhft.chronicle.queue.internal.util.InternalFileUtil.getAllOpenFilesIsSupportedOnOS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/**
+ * Verifies through {@link FileUtil#removableRollFileCandidates} that an appender does not keep an old
+ * roll-cycle file open once it has rolled into the past. Removal stops at the first still-open file
+ * (oldest first), so a pinned old file blocks everything after it.
+ * <p>
+ * These tests inspect open descriptors via {@code /proc/self/fd} and are skipped where unsupported.
+ */
+class AppenderReleasesParkedStoreTest extends IndexingTestCommon {
+
+    private static final Comparator<File> EARLIEST_FIRST = Comparator.comparing(File::getName);
+
+    // Control: with only the writer active, the current cycle stays open and earlier cycles are removable.
+    @Test
+    void writerAloneLeavesEveryEarlierCycleRemovable() {
+        assumeTrue(getAllOpenFilesIsSupportedOnOS());
+
+        appender.writeText("cycle-0");
+        timeProvider.advanceMillis(1_001);
+        appender.writeText("cycle-1");
+        timeProvider.advanceMillis(1_001);
+        appender.writeText("cycle-2");
+
+        final List<File> created = cycleFiles();
+        assertEquals(3, created.size(), "expected three roll-cycle files, got " + created);
+        assertEquals(created.subList(0, 2), removableCandidates());
+    }
+
+    // A fresh appender parks on the current cycle at construction and never writes; after the writer rolls
+    // past it, it must not keep the now-old file open.
+    @Test
+    void parkedAppenderDoesNotPinAnOldCycle() {
+        assumeTrue(getAllOpenFilesIsSupportedOnOS());
+
+        appender.writeText("cycle-0");
+        timeProvider.advanceMillis(1_001);
+        appender.writeText("cycle-1");
+
+        final StoreAppender parked = (StoreAppender) queue.createAppender();
+        try {
+            timeProvider.advanceMillis(1_001);
+            appender.writeText("cycle-2");
+
+            final List<File> created = cycleFiles();
+            assertEquals(3, created.size(), "expected three roll-cycle files, got " + created);
+            assertEquals(created.subList(0, 2), removableCandidates());
+        } finally {
+            parked.close();
+        }
+    }
+
+    // All roll-cycle files, earliest first.
+    private List<File> cycleFiles() {
+        final File[] files = queue.file().listFiles(FileUtil::hasQueueSuffix);
+        assertNotNull(files, "no queue files found in " + queue.file());
+        return Stream.of(files).sorted(EARLIEST_FIRST).collect(toList());
+    }
+
+    // Removable candidates, earliest first. The parked store's file descriptor is dropped on a
+    // background thread, so drain the releaser before inspecting open descriptors.
+    private List<File> removableCandidates() {
+        BackgroundResourceReleaser.releasePendingResources();
+        final List<File> candidates = FileUtil.removableRollFileCandidates(queue.file()).collect(toList());
+        assertEquals(candidates.stream().sorted(EARLIEST_FIRST).collect(toList()), candidates);
+        return candidates;
+    }
+}

src/test/java/net/openhft/chronicle/queue/impl/single/IndexingTestCommon.java

@@ -24,7 +24,7 @@
  */
 public class IndexingTestCommon extends QueueTestCommon {
 
-    SetTimeProvider timeProvider;
+   protected SetTimeProvider timeProvider;
     SingleChronicleQueue queue;
     StoreAppender appender;
     private List<Closeable> closeables;

src/test/java/net/openhft/chronicle/queue/impl/single/NormaliseEOFsTest.java

@@ -98,6 +98,34 @@ public void normaliseShouldResumeFromPreviousNormalisation() {
         }
     }
 
+    // A freshly constructed appender parks on the current cycle during its EOF back-scan and then
+    // releases that store, leaving its cycle field unset. normaliseEOFs() must still finalise the older
+    // cycles rather than silently no-op - the watermark it advances is "normalisedEOFsTo" in the table store.
+    @Test
+    public void freshlyConstructedAppenderNormalisesWithoutWriting() {
+        SetTimeProvider setTimeProvider = new SetTimeProvider();
+        try (final SingleChronicleQueue queue = createQueue(setTimeProvider);
+             final ExcerptAppender writer = queue.createAppender()) {
+            writer.writeText("cycle-0");
+            setTimeProvider.advanceMillis(1_001);
+            writer.writeText("cycle-1");
+            setTimeProvider.advanceMillis(1_001);
+            writer.writeText("cycle-2");
+
+            final int firstCycle = queue.firstCycle();
+            try (final ExcerptAppender fresh = queue.createAppender()) {
+                final long normalisedBefore = queue.tableStoreAcquire("normalisedEOFsTo", firstCycle).getVolatileValue();
+
+                fresh.normaliseEOFs();
+
+                final long normalisedAfter = queue.tableStoreAcquire("normalisedEOFsTo", firstCycle).getVolatileValue();
+                assertTrue(normalisedAfter > normalisedBefore,
+                        "normaliseEOFs on a never-written appender should have advanced the watermark past "
+                                + normalisedBefore + " but it stayed at " + normalisedAfter);
+            }
+        }
+    }
+
     private void createNewRollCycles(ExcerptAppender appender, SetTimeProvider timeProvider) {
         for (int i = 0; i < 10; i++) {
             timeProvider.advanceMillis(3_000);

src/test/java/net/openhft/chronicle/queue/impl/single/NotCompleteTest.java

@@ -152,7 +152,10 @@ private SingleChronicleQueue createQueue(File tmpDir) {
 
     // the last line of the dump changes - haven't spent the time to get to the bottom of this
     private String cleanQueueDump(String from) {
-        return from.replaceAll("# [0-9]+ bytes remaining$", "").replaceAll("modCount: (\\d+)", "modCount: 00");
+        return from.replaceAll("# [0-9]+ bytes remaining", "# NN bytes remaining")
+                .replaceAll("modCount: (\\d+)", "modCount: 00")
+                // strip the normalisedEOFsTo record written by first-write EOF normalisation
+                .replaceAll("# position: \\d+, header: \\d+\\R--- !!data #binary\\RnormalisedEOFsTo: \\d+\\R", "");
     }
 
     private void doWrite(ChronicleQueue queue, BiConsumer<PersonListener, ChronicleQueue> action) {