diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 9d400f9d40b1..d03146c60b73 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -20,6 +20,8 @@ package org.apache.iotdb.db.i18n; public final class DataNodePipeMessages { + public static final String LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B = + "Subscription replay has made no cursor progress for {} ms: {}. Inspect prefetch worker stacks and WAL I/O."; public static final String LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 4891920e079f..0c2282616d84 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -20,6 +20,8 @@ package org.apache.iotdb.db.i18n; public final class DataNodePipeMessages { + public static final String LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B = + "订阅回放游标已连续 {} ms 未推进：{}。请检查预取工作线程栈和 WAL I/O。"; public static final String LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java index 4e71f2e2fe97..67655f06e315 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java @@ -321,6 +321,17 @@ public class ConsensusPrefetchingQueue { private final SubscriptionQueueTimeTracker timeTracker = new SubscriptionQueueTimeTracker(); + private volatile long lastPrefetchDurationMs; + private volatile PrefetchRoundResult.Type lastPrefetchResult; + private volatile String walReplayState = "NOT_SCANNED"; + + // Accessed only by the serialized prefetch worker while holding the queue read lock. + private boolean replayProgressObserved; + private long lastObservedReplayCursor; + private long lastReplayProgressNanos; + private boolean replayStallWarningLogged; + private long lastReplayStallWarningNanos; + private volatile String lastConsumerId = ""; private volatile long lastPendingAcceptedEntries = 0L; @@ -1031,8 +1042,7 @@ private boolean initPrefetchUnderInitializationLock(final RegionProgress regionP this.nextExpectedSearchIndex.set(resolvedStart.getStartSearchIndex()); // Use the same factory as cursor resets; the default implementation still returns null for // readers without WAL support. - this.subscriptionWALIterator = - createSubscriptionWALIterator(resolvedStart.getStartSearchIndex()); + resetSubscriptionWALPosition(resolvedStart.getStartSearchIndex()); this.prefetchInitialized = true; timeTracker.start(); this.observedSeekGeneration = seekGeneration.get(); @@ -1617,6 +1627,8 @@ public SubscriptionEvent pollTablets( private static final long WAL_GAP_WAIT_LOG_INTERVAL_MS = 5_000L; private static final long PREFETCH_STATS_LOG_INTERVAL_MS = 5_000L; + private static final long REPLAY_STALL_WARNING_THRESHOLD_NANOS = TimeUnit.SECONDS.toNanos(60L); + private static final long REPLAY_STALL_WARNING_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(30L); private static final long SLOW_PREFETCH_ROUND_THRESHOLD_MS = 10_000L; @@ -1649,6 +1661,8 @@ public PrefetchRoundResult drivePrefetchOnce() { return result; } finally { final long durationMs = timeTracker.endPrefetch(); + lastPrefetchDurationMs = durationMs; + lastPrefetchResult = Objects.nonNull(result) ? result.getType() : null; if (debugEnabled) { LOGGER.debug( DataNodePipeMessages @@ -1859,6 +1873,8 @@ private PrefetchRoundResult drivePrefetchOnceInternal() { } return PrefetchRoundResult.rescheduleAfter(100L); } finally { + final ProgressWALIterator iterator = subscriptionWALIterator; + walReplayState = Objects.nonNull(iterator) ? iterator.toString() : "UNAVAILABLE"; releaseReadLock(); } } @@ -1869,6 +1885,8 @@ private void logPeriodicStatsIfNecessary() { return; } + reportReplayStallIfNecessary(System.nanoTime()); + final long currentPendingAcceptedEntries = pendingPathAcceptedEntries.get(); final long currentWalAcceptedEntries = walPathAcceptedEntries.get(); LOGGER.info( @@ -1893,7 +1911,36 @@ private void logPeriodicStatsIfNecessary() { lastWalAcceptedEntries = currentWalAcceptedEntries; } + private void reportReplayStallIfNecessary(final long nowNanos) { + final long cursor = nextExpectedSearchIndex.get(); + if (!prefetchInitialized || !isActive || isClosed || closeRequested || getLag() <= 0L) { + replayProgressObserved = false; + replayStallWarningLogged = false; + return; + } + if (!replayProgressObserved || cursor != lastObservedReplayCursor) { + replayProgressObserved = true; + lastObservedReplayCursor = cursor; + lastReplayProgressNanos = nowNanos; + replayStallWarningLogged = false; + return; + } + final long stalledNanos = nowNanos - lastReplayProgressNanos; + if (stalledNanos >= REPLAY_STALL_WARNING_THRESHOLD_NANOS + && (!replayStallWarningLogged + || nowNanos - lastReplayStallWarningNanos >= REPLAY_STALL_WARNING_INTERVAL_NANOS)) { + LOGGER.warn( + DataNodePipeMessages + .LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B, + TimeUnit.NANOSECONDS.toMillis(stalledNanos), + this); + lastReplayStallWarningNanos = nowNanos; + replayStallWarningLogged = true; + } + } + private void resetRoundStateForSeek(final long newSeekGeneration) { + replayProgressObserved = false; restorePendingSubscriptionWalCursor(newSeekGeneration); discardRealtimeEntries(); discardBatch(lingerBatch); @@ -2473,7 +2520,12 @@ private void resetSubscriptionWALPosition(final long startSearchIndex) { protected ProgressWALIterator createSubscriptionWALIterator(final long startSearchIndex) { if (consensusReqReader instanceof WALNode) { - return new ProgressWALIterator((WALNode) consensusReqReader, startSearchIndex); + final ProgressWALIterator iterator = + new ProgressWALIterator((WALNode) consensusReqReader, startSearchIndex); + // The constructor sets the local filtering lower bound. Complete initialization here by + // positioning past historical files only when every writer's progress is already covered. + iterator.advanceTo(startSearchIndex, this::isWriterProgressCoveredForWalFastForward); + return iterator; } return null; } @@ -5093,6 +5145,9 @@ public Map coreReportMessage() { result.put("maxDeliveryIntervalMs", String.valueOf(getMaxDeliveryIntervalMs())); result.put("deliveryIdleTimeMs", String.valueOf(getDeliveryIdleTimeMs())); result.put("prefetchDurationMs", String.valueOf(getPrefetchDurationMs())); + result.put("lastPrefetchDurationMs", String.valueOf(lastPrefetchDurationMs)); + result.put("lastPrefetchResult", String.valueOf(lastPrefetchResult)); + result.put("walReplayState", walReplayState); result.put("maxPrefetchDurationMs", String.valueOf(getMaxPrefetchDurationMs())); result.put("prefetchIdleTimeMs", String.valueOf(getPrefetchIdleTimeMs())); result.put("realtimeAdmissionBlocked", String.valueOf(realtimeAdmissionBlocked.get())); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java index 3f5186893eea..13568fd00be9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java @@ -69,12 +69,14 @@ public interface WriterProgressCoverage { /** * One shared budget for all WAL probes and reads in a prefetch round. The time limit is - * cooperative: it is checked between operations and cannot interrupt a single file read. + * cooperative: it starts with the first WAL probe, is checked between operations, and cannot + * interrupt a single file read. Queue housekeeping must not consume the scan's time slice. */ static final class ScanBudget { private int remainingEntries; - private final long startNanos; + private long startNanos; + private boolean started; private final long maxDurationNanos; private final LongSupplier nanoTime; private boolean yielded; @@ -87,11 +89,15 @@ static final class ScanBudget { this.remainingEntries = maxEntries; this.maxDurationNanos = maxDurationNanos; this.nanoTime = nanoTime; - this.startNanos = nanoTime.getAsLong(); } private boolean shouldYield() { - if (remainingEntries <= 0 || nanoTime.getAsLong() - startNanos >= maxDurationNanos) { + final long nowNanos = nanoTime.getAsLong(); + if (!started) { + startNanos = nowNanos; + started = true; + } + if (remainingEntries <= 0 || nowNanos - startNanos >= maxDurationNanos) { yielded = true; } return yielded; @@ -141,6 +147,7 @@ boolean hasYielded() { private IndexedConsensusRequest nextReady; private boolean scanYielded; + private long bodyReadAttempts; public ProgressWALIterator(final File logDirectory) { this(logDirectory, Long.MIN_VALUE); @@ -279,9 +286,13 @@ void retry(final IndexedConsensusRequest request) { */ public void advanceTo( final long targetSearchIndex, final WriterProgressCoverage writerProgressCoverage) { - if (targetSearchIndex <= minimumSearchIndex) { + if (targetSearchIndex < minimumSearchIndex + || (targetSearchIndex == minimumSearchIndex && currentFileIndex >= 0)) { return; } + // A lower bound filters requests; it does not position the reader. A fresh/rebuilt iterator + // already has this bound but must still be allowed to skip fully covered historical files. + // Once scanning starts, equal-bound probes must not recheck the same uncovered file footers. minimumSearchIndex = targetSearchIndex; if (writerProgressCoverage == null || !canDiscardBufferedRequests(writerProgressCoverage)) { @@ -447,6 +458,26 @@ boolean isWaitingForReadableWal() { return retryCurrentReader; } + @Override + public String toString() { + final String status = + retryCurrentReader + ? "WAITING_FOR_READABLE_WAL" + : scanYielded + ? "SCAN_YIELDED" + : hasBufferedNext() ? "REQUEST_BUFFERED" : "NO_BUFFERED_REQUEST"; + return status + + "[fileVersion=" + + currentReaderVersionId + + ", entryOffset=" + + consumedEntryCountInCurrentFile + + ", bodyReadAttempts=" + + bodyReadAttempts + + ", minimumSearchIndex=" + + minimumSearchIndex + + "]"; + } + String getSkippedBrokenWalFileDetails(final int firstFileIndex) { final int firstAvailableIndex = getSkippedBrokenWalFileCount() - skippedBrokenWalFileDetails.size(); @@ -515,6 +546,7 @@ private IndexedConsensusRequest advance(final ScanBudget scanBudget) throws IOEx // hasNext() can otherwise read an arbitrary WAL prefix before returning a request. scanBudget.remainingEntries--; } + bodyReadAttempts++; final ByteBuffer buffer = currentReader.next(); consumedEntryCountInCurrentFile = currentReader.getCurrentEntryIndex() + 1; final WALEntryType type = WALEntryType.valueOf(buffer.get()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java index cc30193647af..d50795cc3518 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java @@ -35,6 +35,7 @@ import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission; import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion; @@ -56,6 +57,10 @@ import org.apache.iotdb.rpc.subscription.payload.poll.WriterId; import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.utils.Pair; @@ -63,6 +68,7 @@ import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; +import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; @@ -112,6 +118,99 @@ public class ConsensusPrefetchingQueueTest { @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + @Test + public void testReplayStallWarningIsThrottledAndResetsAfterProgress() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final Logger logger = (Logger) LoggerFactory.getLogger(ConsensusPrefetchingQueue.class); + final Level originalLevel = logger.getLevel(); + logger.setLevel(Level.WARN); + final ListAppender appender = new ListAppender<>(); + appender.start(); + logger.addAppender(appender); + ConsensusPrefetchingQueue queue = null; + try { + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + reader.currentSearchIndex = 10L; + final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class); + when(server.getConsensusReqReader()).thenReturn(reader); + when(server.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + new DataRegionId(1), + server, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + mock(ConsensusLogToTabletConverter.class), + newCommitManager(temporaryFolder.newFolder("replay-stall-warning")), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true); + final Field initialized = + ConsensusPrefetchingQueue.class.getDeclaredField("prefetchInitialized"); + initialized.setAccessible(true); + initialized.setBoolean(queue, true); + final Method report = + ConsensusPrefetchingQueue.class.getDeclaredMethod( + "reportReplayStallIfNecessary", long.class); + report.setAccessible(true); + final Supplier warningCount = + () -> + appender.list.stream() + .filter( + event -> + event + .getMessage() + .equals( + DataNodePipeMessages + .LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B)) + .count(); + report.invoke(queue, 0L); + report.invoke(queue, TimeUnit.SECONDS.toNanos(59L)); + assertEquals(Long.valueOf(0L), warningCount.get()); + report.invoke(queue, TimeUnit.SECONDS.toNanos(60L)); + report.invoke(queue, TimeUnit.SECONDS.toNanos(89L)); + assertEquals(Long.valueOf(1L), warningCount.get()); + report.invoke(queue, TimeUnit.SECONDS.toNanos(90L)); + assertEquals(Long.valueOf(2L), warningCount.get()); + + final Field cursorField = + ConsensusPrefetchingQueue.class.getDeclaredField("nextExpectedSearchIndex"); + cursorField.setAccessible(true); + ((AtomicLong) cursorField.get(queue)).incrementAndGet(); + report.invoke(queue, TimeUnit.SECONDS.toNanos(91L)); + report.invoke(queue, TimeUnit.SECONDS.toNanos(150L)); + assertEquals(Long.valueOf(2L), warningCount.get()); + report.invoke(queue, TimeUnit.SECONDS.toNanos(151L)); + assertEquals(Long.valueOf(3L), warningCount.get()); + + reader.currentSearchIndex = 0L; + report.invoke(queue, TimeUnit.SECONDS.toNanos(250L)); + report.invoke(queue, TimeUnit.SECONDS.toNanos(350L)); + reader.currentSearchIndex = 10L; + // A backlog appearing after an idle period starts a new observation window. + report.invoke(queue, TimeUnit.SECONDS.toNanos(450L)); + report.invoke(queue, TimeUnit.SECONDS.toNanos(509L)); + assertEquals(Long.valueOf(3L), warningCount.get()); + queue.setActive(false); + report.invoke(queue, TimeUnit.SECONDS.toNanos(600L)); + assertEquals(Long.valueOf(3L), warningCount.get()); + } finally { + logger.detachAppender(appender); + logger.setLevel(originalLevel); + appender.stop(); + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testLateOldPrefetchDeregistrationKeepsReplacementTask() { final ConsensusSubscriptionPrefetchExecutor executor = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java index 192ed7c3e9c8..204a557367e7 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java @@ -56,6 +56,7 @@ import java.io.IOException; import java.lang.reflect.Constructor; import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; @@ -183,6 +184,56 @@ public void testYieldedGapScanKeepsIteratorAndDeliversEachRequestOnce() throws E } queue.drivePrefetchOnce(); assertNull(queue.poll("consumer")); + for (int round = 0; round < 10 && queue.getPendingEventCount() > 0; round++) { + queue.drivePrefetchOnce(); + } + assertEquals(0, queue.getPendingEventCount()); + + // Rebuilding at the unchanged cursor must position the reader past covered history. The + // fragmented prefix is larger than a whole scan budget and must not delay WAL takeover. + try (WALWriter writer = new WALWriter(liveWal, WALFileVersion.V3)) { + writeWalEntry(writer, 4L); + writeWalEntry(writer, 5L); + } + try (WALWriter ignored = + new WALWriter( + new File( + walDirectory, + WALFileUtils.getLogFileName(2L, 5L, WALFileStatus.CONTAINS_SEARCH_INDEX)), + WALFileVersion.V3)) { + // Seal the new data file so replay can take over at index 4. + } + when(walNode.getCurrentSearchIndex()).thenReturn(5L); + when(walNode.getCurrentWALFileVersion()).thenReturn(2L); + // The factory must return an iterator that is ready for bounded replay at the requested + // cursor, without requiring a separate advanceTo call from its caller. + try (ProgressWALIterator createdIterator = queue.createSubscriptionWALIterator(4L)) { + final ProgressWALIterator.ScanBudget budget = + new ProgressWALIterator.ScanBudget(2, Long.MAX_VALUE); + assertTrue(createdIterator.hasNext(budget)); + assertEquals(4L, createdIterator.next().getSearchIndex()); + assertFalse(budget.hasYielded()); + } + final Method reset = + ConsensusPrefetchingQueue.class.getDeclaredMethod( + "resetSubscriptionWALPosition", long.class); + reset.setAccessible(true); + reset.invoke(queue, 4L); + assertTrue(pendingEntries(queue).offer(createRequest(5L))); + queue.drivePrefetchOnce(); + assertTrue(queue.toString(), queue.getCurrentReadSearchIndex() >= 5L); + for (int round = 0; round < 10 && queue.getCurrentReadSearchIndex() < 6L; round++) { + queue.drivePrefetchOnce(); + } + assertEquals(6L, queue.getCurrentReadSearchIndex()); + assertEquals(Arrays.asList(2L, 3L, 4L, 5L), conversions); + for (long expected = 4L; expected <= 5L; expected++) { + final SubscriptionEvent event = queue.poll("consumer"); + assertNotNull(event); + assertEquals(expected, event.getCommitContext().getWriterProgress().getLocalSeq()); + assertTrue(queue.ack("consumer", event.getCommitContext())); + } + assertEquals(0L, queue.getWalGapSkippedEntries()); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); verify(walNode, never()).rollWALFile(); } finally { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java index cd136e4cc726..370fca537965 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java @@ -119,6 +119,34 @@ public void testBoundedScanKeepsSkippedPrefixAndPendingFragmentsAcrossRounds() t } } + @Test + public void testScanBudgetStartsWhenWalScanningBegins() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-delayed-scan"); + final File walFile = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final AtomicLong clock = new AtomicLong(); + try { + try (WALWriter writer = new WALWriter(walFile, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 1000L, 7, 1L)); + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 2000L, 7, 2L)); + } + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { + final ProgressWALIterator.ScanBudget budget = + new ProgressWALIterator.ScanBudget(100, 50L, clock::get); + // Queue housekeeping, realtime materialization, or a JVM pause can consume a whole + // time slice before the first WAL probe. That must not starve replay. + clock.set(1000L); + assertTrue(iterator.hasNext(budget)); + assertEquals(1L, iterator.next().getSearchIndex()); + assertFalse(budget.hasYielded()); + } + } finally { + Files.deleteIfExists(walFile.toPath()); + Files.deleteIfExists(dir); + } + } + @Test public void testElapsedScanBudgetYieldsWithoutFlushingIncompleteRequest() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-scan-time-slice"); @@ -142,8 +170,7 @@ public void testElapsedScanBudgetYieldsWithoutFlushingIncompleteRequest() throws new ProgressWALIterator.ScanBudget(1, Long.MAX_VALUE); assertFalse(iterator.hasNext(first)); final ProgressWALIterator.ScanBudget expired = - new ProgressWALIterator.ScanBudget(100, 50L, clock::get); - clock.set(50L); + new ProgressWALIterator.ScanBudget(100, 50L, () -> clock.getAndAdd(50L)); assertFalse(iterator.hasNext(expired)); assertTrue(expired.hasYielded()); assertFalse(iterator.hasBufferedNext()); @@ -974,6 +1001,27 @@ public void testAdvanceToSkipsOnlyFullyCoveredWalFiles() throws Exception { assertTrue(iterator.hasNext()); assertEquals(6L, iterator.next().getSearchIndex()); } + + // A freshly rebuilt replay iterator already has this lower bound, but has not read or + // skipped any files yet. It must still fast-forward covered history before a bounded scan. + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 6L)) { + iterator.advanceTo(6L, (physicalTime, nodeId, localSeq) -> true); + final ProgressWALIterator.ScanBudget budget = + new ProgressWALIterator.ScanBudget(2, Long.MAX_VALUE); + assertTrue(iterator.hasNext(budget)); + assertEquals(6L, iterator.next().getSearchIndex()); + assertFalse(budget.hasYielded()); + } + + // Equal-bound realignment must retain uncovered follower entries, just like an increasing + // bound. They have no local search index and cannot be skipped using the local cursor. + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 6L)) { + iterator.advanceTo(6L, (physicalTime, nodeId, localSeq) -> nodeId == 7); + assertTrue(iterator.hasNext()); + assertEquals(8, iterator.next().getNodeId()); + assertTrue(iterator.hasNext()); + assertEquals(6L, iterator.next().getSearchIndex()); + } } finally { Files.deleteIfExists(localWal.toPath()); Files.deleteIfExists(followerWal.toPath());