[Bug] AutoSwitchHA slave truncation is not crash-safe because stale CommitLog tail remains valid on disk

[chenxu80]

Before Creating the Bug Report

• I found a bug, not just asking a question, which should be created in GitHub Discussions.

• I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.

• I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.

Runtime platform environment

ubuntu 24.04

RocketMQ version

branch: develop version: 5.5.0 Git commit id: a6fb9e2

JDK Version

JDK 8u202

Describe the Bug

In Controller mode, RocketMQ uses AutoSwitchHAService.

When a slave reconnects to the master, AutoSwitchHAClient#doTruncate() calculates a consistent point and truncates the local CommitLog:

final long truncateOffset =
    localEpochCache.findConsistentPoint(masterEpochCache);

if (!this.messageStore.truncateFiles(truncateOffset)) {
    LOGGER.error("Failed to truncate slave log to {}", truncateOffset);
    return false;
}

The truncation offset is validated as a message boundary, but the actual truncation only resets the in-memory positions of the target MappedFile.

The physical bytes after the truncation offset remain unchanged on disk.

If the slave crashes again after truncation but before the master has durably overwritten the stale tail, CommitLog recovery may scan and recover the stale messages again.

1. Controller mode uses AutoSwitchHAService

if (brokerConfig.isEnableControllerMode()) {
    this.haService = new AutoSwitchHAService();
}

Source:

rocketmq/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java

Lines 1978 to 1988 in a6fb9e2

	private void initializeHAService() {
	if (!this.messageStoreConfig.isEnableDLegerCommitLog() && !this.messageStoreConfig.isDuplicationEnable()) {
	if (brokerConfig.isEnableControllerMode()) {
	this.haService = new AutoSwitchHAService();
	LOGGER.warn("Load AutoSwitch HA Service: {}", AutoSwitchHAService.class.getSimpleName());
	} else {
	this.haService = ServiceProvider.loadClass(HAService.class);
	if (null == this.haService) {
	this.haService = new DefaultHAService();
	LOGGER.warn("Load default HA Service: {}", DefaultHAService.class.getSimpleName());
	}

2. AutoSwitchHAClient truncates the slave CommitLog

final long truncateOffset =
    localEpochCache.findConsistentPoint(masterEpochCache);

if (!this.messageStore.truncateFiles(truncateOffset)) {
    LOGGER.error("Failed to truncate slave log to {}", truncateOffset);
    return false;
}

changeCurrentState(HAConnectionState.TRANSFER);
this.currentReportedOffset = truncateOffset;

Source:

rocketmq/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java

Lines 450 to 477 in a6fb9e2

	LOGGER.info("master epoch entries is {}", masterEpochCache.getAllEntries());
	LOGGER.info("local epoch entries is {}", localEpochEntries);
	final long truncateOffset = localEpochCache.findConsistentPoint(masterEpochCache);
	LOGGER.info("truncateOffset is {}", truncateOffset);
	if (truncateOffset < 0) {
	// If truncateOffset < 0, means we can't find a consistent point
	LOGGER.error("Failed to find a consistent point between masterEpoch:{} and slaveEpoch:{}", masterEpochEntries, localEpochEntries);
	return false;
	}
	if (!this.messageStore.truncateFiles(truncateOffset)) {
	LOGGER.error("Failed to truncate slave log to {}", truncateOffset);
	return false;
	}
	final long maxPhyOffset = this.messageStore.getMaxPhyOffset();
	if (truncateOffset < maxPhyOffset) {
	this.epochCache.truncateSuffixByOffset(truncateOffset);
	LOGGER.info("Truncate slave log to {} success, change to transfer state", truncateOffset);
	}
	changeCurrentState(HAConnectionState.TRANSFER);
	this.currentReportedOffset = truncateOffset;
	}
	if (!reportSlaveMaxOffset(HAConnectionState.TRANSFER)) {
	LOGGER.error("AutoSwitchHAClient report max offset to master failed");
	return false;
	}

3. DefaultMessageStore only checks whether the offset is aligned

public boolean truncateFiles(long offsetToTruncate) {
    if (offsetToTruncate >= this.getMaxPhyOffset()) {
        return true;
    }

    if (!isOffsetAligned(offsetToTruncate)) {
        return false;
    }

    truncateDirtyFiles(offsetToTruncate);
    return true;
}

Source:

rocketmq/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java

Lines 828 to 851 in a6fb9e2

	public boolean truncateFiles(long offsetToTruncate) throws RocksDBException {
	if (offsetToTruncate >= this.getMaxPhyOffset()) {
	LOGGER.info("no need to truncate files, truncate offset is {}, max physical offset is {}", offsetToTruncate, this.getMaxPhyOffset());
	return true;
	}
	if (!isOffsetAligned(offsetToTruncate)) {
	LOGGER.error("offset {} is not align, truncate failed, need manual fix", offsetToTruncate);
	return false;
	}
	truncateDirtyFiles(offsetToTruncate);
	return true;
	}
	@Override
	public boolean isOffsetAligned(long offset) {
	SelectMappedBufferResult mappedBufferResult = this.getCommitLogData(offset);
	if (mappedBufferResult == null) {
	return true;
	}
	DispatchRequest dispatchRequest = this.commitLog.checkMessageAndReturnSize(mappedBufferResult.getByteBuffer(), true, false);
	return dispatchRequest.isSuccess();

The alignment check prevents truncation in the middle of a message, but it does not make the truncation durable.

4. MappedFileQueue only resets in-memory positions

public void truncateDirtyFiles(long offset) {
    List<MappedFile> willRemoveFiles = new ArrayList<>();

    for (MappedFile file : this.mappedFiles) {
        long fileTailOffset =
            file.getFileFromOffset() + this.mappedFileSize;

        if (fileTailOffset > offset) {
            if (offset >= file.getFileFromOffset()) {
                file.setWrotePosition(
                    (int) (offset % this.mappedFileSize));
                file.setCommittedPosition(
                    (int) (offset % this.mappedFileSize));
                file.setFlushedPosition(
                    (int) (offset % this.mappedFileSize));
            } else {
                file.destroy(1000);
                willRemoveFiles.add(file);
            }
        }
    }

    this.deleteExpiredFile(willRemoveFiles);
}

Source:

rocketmq/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java

Lines 217 to 234 in a6fb9e2

	public void truncateDirtyFiles(long offset) {
	List<MappedFile> willRemoveFiles = new ArrayList<>();
	for (MappedFile file : this.mappedFiles) {
	long fileTailOffset = file.getFileFromOffset() + this.mappedFileSize;
	if (fileTailOffset > offset) {
	if (offset >= file.getFileFromOffset()) {
	file.setWrotePosition((int) (offset % this.mappedFileSize));
	file.setCommittedPosition((int) (offset % this.mappedFileSize));
	file.setFlushedPosition((int) (offset % this.mappedFileSize));
	} else {
	file.destroy(1000);
	willRemoveFiles.add(file);
	}
	}
	}
	this.deleteExpiredFile(willRemoveFiles);

After truncating to offset T, the state is:

In-memory wrotePosition     = T
In-memory committedPosition = T
In-memory flushedPosition   = T

Physical bytes in [T, oldMaxOffset) remain unchanged

This also applies when T is exactly the beginning of a MappedFile.

Because the condition is:

offset >= file.getFileFromOffset()

the target file is retained with position zero instead of being deleted.

5. Existing MappedFiles are loaded as fully written after restart

mappedFile.setWrotePosition(this.mappedFileSize);
mappedFile.setFlushedPosition(this.mappedFileSize);
mappedFile.setCommittedPosition(this.mappedFileSize);

Source:

rocketmq/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java

Lines 270 to 298 in a6fb9e2

	public boolean doLoad(List<File> files) {
	// ascending order
	files.sort(Comparator.comparing(File::getName));
	for (int i = 0; i < files.size(); i++) {
	File file = files.get(i);
	if (file.isDirectory()) {
	continue;
	}
	if (file.length() == 0 && i == files.size() - 1) {
	boolean ok = file.delete();
	log.warn("{} size is 0, auto delete. is_ok: {}", file, ok);
	continue;
	}
	if (file.length() != this.mappedFileSize) {
	log.warn(file + "\t" + file.length()
	+ " length not matched message store config value, please check it manually");
	return false;
	}
	try {
	MappedFile mappedFile = new DefaultMappedFile(file.getPath(), mappedFileSize, runningFlags, writeWithoutMmap);
	mappedFile.setWrotePosition(this.mappedFileSize);
	mappedFile.setFlushedPosition(this.mappedFileSize);
	mappedFile.setCommittedPosition(this.mappedFileSize);
	this.mappedFiles.add(mappedFile);

CommitLog recovery then scans the physical contents to determine the valid tail.

Because the discarded bytes were not erased or invalidated, the stale records after T may still contain:

• a valid total size;
• a valid magic code;
• a correct physical offset;
• a valid CRC.

As a result, recovery may advance beyond the intended HA truncation offset.

Failure Scenario

A possible failure sequence is:

1. The slave has CommitLog data up to offset S.

2. After a master switch or epoch divergence, AutoSwitchHAClient calculates an aligned consistent point T.

3. T is smaller than S:

   T < S
   

4. The slave executes:

   messageStore.truncateFiles(T);

5. The in-memory CommitLog maximum offset becomes T.

6. The physical bytes in [T, S) remain unchanged.

7. Before the master has durably overwritten the stale tail, the slave crashes again.

8. On restart, the MappedFile is initially loaded as fully written.

9. CommitLog recovery scans the stale records after T.

10. The recovered maximum physical offset may become greater than T again.

Steps to Reproduce

The storage behavior can be reproduced without running a complete Controller cluster.

1. Create a CommitLog MappedFile.

2. Append and flush at least two valid messages.

3. Record the physical start offset of the second message as T.

4. Call:

   messageStore.truncateFiles(T);

5. Verify:

   messageStore.getMaxPhyOffset() == T
   

6. Do not append new data from T.

7. Simulate an abrupt process termination.

8. Reload the same store directory.

9. Execute CommitLog recovery.

10. Check the recovered maximum physical offset.

Expected result:

recoveredMaxPhyOffset == T

Potential current result:

recoveredMaxPhyOffset > T

The old second message may be recovered because its physical bytes remain valid on disk.

What Did You Expect to See?

Expected Behavior

After AutoSwitchHAClient successfully truncates the slave to offset T, a subsequent restart must not recover any CommitLog record at or after T unless that record has been received again from the current master.

The HA truncation should provide a durable recovery boundary.

What Did You See Instead?

Slave commitLog keeps the stale tail.

Additional Context

No response

[RockteMQ-AI]

Issue Evaluation

Category: type/bug | Status: Confirmed

The reported crash-safety issue in AutoSwitchHAService truncation has been verified against the current codebase on the develop branch.

Root Cause:

MappedFileQueue.truncateDirtyFiles() (line 217) only resets in-memory positions (wrotePosition, committedPosition, flushedPosition) for the target MappedFile. The physical bytes after the truncation offset remain unchanged on disk — no zeroing or invalidation is performed.

// MappedFileQueue.java:224-228
file.setWrotePosition((int) (offset % this.mappedFileSize));
file.setCommittedPosition((int) (offset % this.mappedFileSize));
file.setFlushedPosition((int) (offset % this.mappedFileSize));

Why CRC check does not prevent recovery of stale data:

checkCRCOnRecover defaults to true, but the stale tail bytes are exact replicas of previously valid messages replicated from the master. Their magic code, total size, physical offset, and CRC are all intact. CommitLog.recoverNormally() (line 348) scans these bytes via checkMessageAndReturnSize() and accepts them as valid messages.

Failure scenario (confirmed):

1. Slave receives and persists messages up to offset X
2. Master truncates its log to offset T (T < X) due to a divergent epoch
3. Slave reconnects, AutoSwitchHAClient.doTruncate() calls truncateFiles(T) → only resets in-memory positions
4. Slave crashes before master overwrites bytes [T, X)
5. On restart, CommitLog.recoverNormally() scans the file and recovers stale messages in [T, X)
6. Slave now has divergent data that the master does not have → data inconsistency

Impact: Data inconsistency between master and slave in Controller mode. The slave may serve or replicate messages that the master has already truncated.

Severity: High — affects data integrity in Controller mode failover scenarios.

Suggested fix direction: In truncateDirtyFiles(), after resetting in-memory positions, physically zero out or invalidate the bytes from the truncation offset to the end of the mapped file (e.g., write a zero-filled buffer or set the magic code of the first stale message to an invalid value).

An automated fix proposal will be generated. Reply /approve to proceed with PR generation, /revise to request changes to the approach, or /reject to decline.

---

Automated evaluation by RockteMQ-AI

[RockteMQ-AI]

Fix Spec Summary

A repair plan has been generated for this issue.

Fix approach: Physically invalidate the stale CommitLog tail in MappedFileQueue.truncateDirtyFiles() after resetting in-memory positions.

Key changes:

1. Add MappedFile.eraseAndForce(startPosition, endPosition) — zero-fills the dirty region via positional FileChannel.write() and calls force(false) for durability
2. Modify MappedFileQueue.truncateDirtyFiles() to capture the old wrote position before reset, then call eraseAndForce() on the boundary file

Why zeros (not BLANK_MAGIC_CODE): Zeros yield totalSize == 0 + magicCode == 0 in checkMessageAndReturnSize(), hitting the "illegal magic code" branch → recovery stops exactly at the truncation offset. BLANK_MAGIC_CODE means "skip to next file" which is wrong mid-file.

Scope: Only MappedFileQueue.truncateDirtyFiles() and the MappedFile interface/DefaultMappedFile implementation need changes. No changes to AutoSwitchHAClient, DefaultMessageStore, or CommitLog truncation paths.

Testing: Crash-recovery simulation test (truncate → shutdown without flush → restart → verify stale messages are not recovered), plus unit tests for eraseAndForce and boundary cases.

Reply /approve to proceed with PR generation, /revise <feedback> to request changes, or /reject to decline.

---

Automated fix proposal by RockteMQ-AI