QUEUE-123 added tests to check for a stuck tailer condition

Author: benbonavia

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

@@ -0,0 +1,284 @@
+/*
+ * Copyright 2013-2025 chronicle.software; SPDX-License-Identifier: Apache-2.0
+ */
+package net.openhft.chronicle.queue;
+
+import net.openhft.chronicle.bytes.Bytes;
+import net.openhft.chronicle.core.time.SetTimeProvider;
+import net.openhft.chronicle.queue.impl.single.InternalAppender;
+import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
+import net.openhft.chronicle.queue.impl.single.SingleChronicleQueueBuilder;
+import net.openhft.chronicle.wire.DocumentContext;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.concurrent.TimeUnit;
+
+import static net.openhft.chronicle.queue.rollcycles.TestRollCycles.TEST_HOURLY;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Plan 2 (deterministic, white-box) reproduction of the "tailer stuck in the previous cycle"
+ * failure analysed in TAILER_STUCK_AFTER_ROLL_DURING_BACKFILL.md.
+ * <p>
+ * <b>Root cause (confirmed by these tests):</b> a streaming {@link ExcerptTailer} crosses from one
+ * cycle to the next <em>only</em> when it reads an EOF marker. See
+ * {@code StoreTailer.inACycle2(...)}: the {@code NONE} case (no data, no EOF) just returns - the
+ * design assumes <i>"appender will always write (or recover) EOF"</i>. The {@code cycle <
+ * queue.lastCycle()} test is in the random-access {@code moveToIndex}/{@code toEnd} path only; it
+ * does <b>not</b> apply to sequential reads.
+ * <p>
+ * Therefore if cycle N is left <b>without an EOF</b> (e.g. the roll to N+1 happened via the
+ * replication back-fill / suppressed-roll path that never seals N), a tailer parks at the tip of N
+ * with {@code isPresent()==false} forever - blind to the data already in N+1 - until something
+ * writes/recovers N's EOF (as a healthy forward roll on restart does). {@code lastCycle} is
+ * irrelevant to this.
+ * <p>
+ * <b>These are red bug-repro tests.</b> The {@code tailerShouldRead*} tests assert the DESIRED
+ * behaviour (a streaming tailer reads the data already on disk in N+1) and therefore <b>FAIL</b>
+ * against the current code; a fix that seals/recovers N's EOF should turn them green. The remaining
+ * tests ({@code tailerRecovers...}, {@code tailerCrosses...}, {@code restart_thenForwardRoll...},
+ * {@code toEnd_alreadyReaches...}) assert behaviour that is already correct and pass.
+ */
+public class TailerStuckOnUnsealedCycleTest extends QueueTestCommon {
+
+    private SingleChronicleQueue build(SetTimeProvider tp) {
+        return build(getTmpDir(), tp);
+    }
+
+    private SingleChronicleQueue build(File dir, SetTimeProvider tp) {
+        return SingleChronicleQueueBuilder.binary(dir)
+                .timeProvider(tp)
+                .rollCycle(TEST_HOURLY)
+                .build();
+    }
+
+    /**
+     * N has data and is NOT sealed (no EOF); N+1 has data. Built with out-of-order (backward)
+     * writes so the forward roll never seals N - mirroring the replication back-fill path that
+     * leaves a cycle un-EOF'd.
+     */
+    private void writeUnsealedNWithDataInNPlus1(SingleChronicleQueue q, ExcerptAppender appender) {
+        int currentCycle = q.cycle();
+        RollCycle rc = q.rollCycle();
+        // write N+1 first, so rolling *back* to N suppresses the EOF and leaves N open
+        InternalAppender internalAppender = (InternalAppender) appender;
+        internalAppender.writeBytes(rc.toIndex(currentCycle + 1, 0), Bytes.from("nPlus1-0"));
+        internalAppender.writeBytes(rc.toIndex(currentCycle, 0), Bytes.from("n-0"));
+        internalAppender.writeBytes(rc.toIndex(currentCycle, 1), Bytes.from("n-1"));
+    }
+
+    /** Read forward until the first non-present document; return how many entries were read. */
+    private int drain(ExcerptTailer tailer) {
+        int read = 0;
+        while (true) {
+            try (DocumentContext dc = tailer.readingDocument()) {
+                if (!dc.isPresent())
+                    return read;
+                read++;
+            }
+        }
+    }
+
+    // ---- the failure --------------------------------------------------------------------------
+
+    @Test
+    public void tailerShouldReadNewCycle_whenPreviousCycleHasNoEof() {
+        SetTimeProvider tp = new SetTimeProvider(TimeUnit.HOURS.toMillis(100));
+        try (SingleChronicleQueue q = build(tp);
+             ExcerptAppender appender = q.createAppender()) {
+            writeUnsealedNWithDataInNPlus1(q, appender);
+
+            try (ExcerptTailer tailer = q.createTailer()) {
+                // DESIRED behaviour - FAILS against the current code, which reads only 2 and parks
+                // at the un-EOF'd cycle N. A streaming tailer should read N's two entries AND cross
+                // into N+1, instead of being stranded by the missing EOF on N.
+                assertEquals("tailer should read N's entries and cross into N+1", 3, drain(tailer));
+            }
+        }
+    }
+
+    @Test
+    public void tailerShouldReadNewCycle_withPreExistingDataInASingleCycle() {
+        File dir = getTmpDir();
+        SetTimeProvider tp = new SetTimeProvider(TimeUnit.HOURS.toMillis(100));
+
+        // 1) a queue that ALREADY EXISTS on disk, with data in a single cycle file (no roll, no EOF)
+        int preCycle;
+        try (SingleChronicleQueue q = build(dir, tp);
+             ExcerptAppender appender = q.createAppender()) {
+            preCycle = q.cycle();
+            appender.writeBytes(Bytes.from("preexisting-0"));
+            appender.writeBytes(Bytes.from("preexisting-1"));
+        }
+
+        // 2) reopen the existing queue and perform the same backward / back-fill construction,
+        //    at a later cycle
+        try (SingleChronicleQueue q = build(dir, tp);
+             ExcerptAppender appender = q.createAppender()) {
+            tp.advanceMillis(TimeUnit.HOURS.toMillis(1));
+            assertTrue("run 2 must be a later cycle than the pre-existing one", q.cycle() > preCycle);
+            writeUnsealedNWithDataInNPlus1(q, appender);
+
+            try (ExcerptTailer tailer = q.createTailer()) {
+                // DESIRED behaviour - FAILS against the current code, which reads only 4 and parks
+                // at the un-EOF'd cycle N. The tailer should read the 2 pre-existing entries, N's 2
+                // entries, AND cross into N+1.
+                assertEquals("tailer should read pre-existing + N + N+1", 5, drain(tailer));
+
+                tailer.direction(TailerDirection.BACKWARD).toEnd();
+                try (DocumentContext dc = tailer.readingDocument()) {
+                    assertTrue(dc.isPresent());
+                    assertEquals("n-1", dc.wire().bytes().toString());
+                }
+            }
+        }
+    }
+
+    // ---- recovery: writing/recovering N's EOF un-sticks the tailer ----------------------------
+
+    @Test
+    public void tailerRecovers_oncePreviousCyclesAreSealed() {
+        SetTimeProvider tp = new SetTimeProvider(TimeUnit.HOURS.toMillis(100));
+        try (SingleChronicleQueue q = build(tp);
+             ExcerptAppender appender = q.createAppender()) {
+            writeUnsealedNWithDataInNPlus1(q, appender);
+
+            try (ExcerptTailer stuck = q.createTailer()) {
+                assertEquals(2, drain(stuck));
+            }
+
+            // a healthy forward roll (as happens on restart once wall-clock has moved on) seals the
+            // earlier cycles - here rolling to N+2 writes the EOFs for both N and N+1
+            tp.advanceMillis(TimeUnit.HOURS.toMillis(2));
+            appender.writeBytes(Bytes.from("nPlus2-0"));
+
+            try (ExcerptTailer healed = q.createTailer()) {
+                // N:0, N:1, N+1:0, N+2:0
+                assertEquals("after the EOFs are written the tailer reads every cycle", 4, drain(healed));
+            }
+        }
+    }
+
+    // ---- control: a normally-sealed queue crosses cycles fine ---------------------------------
+
+    @Test
+    public void tailerCrosses_whenCyclesAreSealed() {
+        SetTimeProvider tp = new SetTimeProvider(TimeUnit.HOURS.toMillis(100));
+        try (SingleChronicleQueue q = build(tp);
+             ExcerptAppender appender = q.createAppender()) {
+            int n = q.cycle();
+            RollCycle rc = q.rollCycle();
+            // forward writes: the roll into N+1 seals N with an EOF
+            InternalAppender internalAppender = (InternalAppender) appender;
+            internalAppender.writeBytes(rc.toIndex(n, 0), Bytes.from("n-0"));
+            internalAppender.writeBytes(rc.toIndex(n, 1), Bytes.from("n-1"));
+            internalAppender.writeBytes(rc.toIndex(n + 1, 0), Bytes.from("nPlus1-0"));
+
+            try (ExcerptTailer tailer = q.createTailer()) {
+                assertEquals("a sealed cycle N lets the tailer cross into N+1", 3, drain(tailer));
+            }
+        }
+    }
+
+    // ---- normaliseEOFs() should make the new-cycle data readable ------------------------------
+
+    @Test
+    public void tailerShouldRead_afterNormaliseEOFs() {
+        SetTimeProvider tp = new SetTimeProvider(TimeUnit.HOURS.toMillis(100));
+        try (SingleChronicleQueue q = build(tp);
+             ExcerptAppender appender = q.createAppender()) {
+            // the backward / back-fill construction leaves the appender positioned ON cycle N
+            writeUnsealedNWithDataInNPlus1(q, appender);
+
+            appender.normaliseEOFs();
+
+            try (ExcerptTailer tailer = q.createTailer()) {
+                // DESIRED behaviour - FAILS against the current code: normaliseEOFs0 only seals
+                // cycles strictly below min(queue.cycle(), appender.cycle()), so with the appender on
+                // N it never seals N and the tailer still reads only 2. normaliseEOFs() should make
+                // N+1 reachable.
+                assertEquals("after normaliseEOFs the tailer should read N+1", 3, drain(tailer));
+            }
+        }
+    }
+
+    // ---- does the directory listing (lastCycle) explain it? -----------------------------------
+
+    @Test
+    public void toEnd_alreadyReachesNPlus1_inTheStuckState() {
+        // In the stuck state lastCycle() is naturally N+1 - creating N+1 bumped the shared,
+        // monotonic table-store highestCycle. So the directory listing is NOT stale, and the
+        // random-access toEnd() path already lands in N+1 even while a STREAMING tailer is stuck
+        // in N. i.e. for this failure the heal is the missing EOF, not a directory-listing refresh.
+        SetTimeProvider tp = new SetTimeProvider(TimeUnit.HOURS.toMillis(100));
+        try (SingleChronicleQueue q = build(tp);
+             ExcerptAppender appender = q.createAppender()) {
+            int n = q.cycle();
+            writeUnsealedNWithDataInNPlus1(q, appender);
+
+            assertEquals("lastCycle already knows about N+1", n + 1, q.lastCycle());
+
+            try (ExcerptTailer streaming = q.createTailer()) {
+                assertEquals("streaming tailer is stuck in N", 2, drain(streaming));
+            }
+            try (ExcerptTailer end = q.createTailer()) {
+                end.toEnd();
+                assertEquals("toEnd() lands in N+1, not N", n + 1, q.rollCycle().toCycle(end.index()));
+            }
+        }
+    }
+
+    // ---- how restart actually heals it --------------------------------------------------------
+
+    @Test
+    public void tailerShouldRead_afterRestart() {
+        File dir = getTmpDir();
+        SetTimeProvider tp = new SetTimeProvider(TimeUnit.HOURS.toMillis(100));
+
+        // run 1: create the stuck state, then "shut down"
+        try (SingleChronicleQueue q = build(dir, tp);
+             ExcerptAppender appender = q.createAppender()) {
+            int n = q.cycle();
+            writeUnsealedNWithDataInNPlus1(q, appender);
+        }
+
+        // run 2: reopen (no forward roll). DESIRED behaviour - FAILS against the current code:
+        // re-opening alone does not seal N (the constructor positions on N but writes no EOF, and
+        // there is no forward-rolling write), so the tailer still reads only 2. A restarted reader
+        // should see every entry already on disk (N and N+1).
+        try (SingleChronicleQueue q = build(dir, tp);
+             ExcerptTailer t = q.createTailer()) {
+            assertEquals("after restart the tailer should read every entry on disk", 3, drain(t));
+        }
+    }
+
+    @Test
+    public void restart_thenForwardRoll_sealsN_andTailerRecovers() {
+        File dir = getTmpDir();
+        SetTimeProvider tp = new SetTimeProvider(TimeUnit.HOURS.toMillis(100));
+
+        int n;
+        try (SingleChronicleQueue q = build(dir, tp);
+             ExcerptAppender appender = q.createAppender()) {
+            n = q.cycle();
+            writeUnsealedNWithDataInNPlus1(q, appender);
+            try (ExcerptTailer t = q.createTailer()) {
+                assertEquals(2, drain(t));
+            }
+        }
+
+        // run 2: wall-clock has moved on while we were down; the first live write rolls forward and
+        // rollCycleTo writes the EOF on N (and N+1) -> the tailer can now cross
+        tp.advanceMillis(TimeUnit.HOURS.toMillis(5));
+        try (SingleChronicleQueue q = build(dir, tp);
+             ExcerptAppender appender = q.createAppender()) {
+            appender.writeBytes(Bytes.from("live-after-restart"));   // forward roll seals N (and N+1)
+            try (ExcerptTailer t = q.createTailer()) {
+                assertTrue("after the post-restart forward roll the tailer crosses past N",
+                        drain(t) >= 3);
+            }
+        }
+    }
+}