diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
index 1c46f9e2ce5..90140ab1361 100644
--- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
@@ -22,6 +22,7 @@
 import java.net.Inet6Address;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.LinkedList;
@@ -899,6 +900,75 @@ public void truncateDirtyFiles(long phyOffset) {
         }
     }
 
+    void persistTruncateMarker(long phyOffset) {
+        MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(phyOffset, false);
+        if (mappedFile == null) {
+            throw new IllegalStateException("Cannot find mapped file when persisting truncate marker, offset="
+                + phyOffset);
+        }
+
+        int relativeOffset = (int) (phyOffset - mappedFile.getFileFromOffset());
+        if (relativeOffset < 0 || relativeOffset + 2 * Integer.BYTES > mappedFile.getFileSize()) {
+            throw new IllegalStateException("Invalid truncate marker position, offset=" + phyOffset
+                + ", file=" + mappedFile.getFileName() + ", relativeOffset=" + relativeOffset);
+        }
+
+        if (!mappedFile.hold()) {
+            throw new IllegalStateException("Cannot hold mapped file when persisting truncate marker, offset="
+                + phyOffset + ", file=" + mappedFile.getFileName());
+        }
+
+        try {
+            MappedByteBuffer mappedByteBuffer = mappedFile.getMappedByteBuffer();
+            validateTruncateMarkerPosition(mappedByteBuffer, mappedFile, phyOffset, relativeOffset);
+            // An invalid zero header makes recovery stop here instead of rolling to the next mapped file.
+            // Do not use appendMessage because persisting the marker must not change the logical wrote position.
+            mappedByteBuffer.putLong(relativeOffset, 0L);
+            mappedByteBuffer.force();
+            log.info("Persisted CommitLog truncate marker, offset={}, file={}, relativeOffset={}",
+                phyOffset, mappedFile.getFileName(), relativeOffset);
+        } catch (RuntimeException e) {
+            throw new IllegalStateException("Failed to persist CommitLog truncate marker, offset=" + phyOffset
+                + ", file=" + mappedFile.getFileName(), e);
+        } finally {
+            mappedFile.release();
+        }
+    }
+
+    private void validateTruncateMarkerPosition(ByteBuffer buffer, MappedFile mappedFile, long phyOffset,
+        int relativeOffset) {
+        int currentPosition = 0;
+        while (currentPosition < relativeOffset) {
+            if (currentPosition + 2 * Integer.BYTES > relativeOffset) {
+                throw invalidTruncateMarkerPosition(mappedFile, phyOffset, relativeOffset, currentPosition);
+            }
+
+            int totalSize = buffer.getInt(currentPosition);
+            int magicCode = buffer.getInt(currentPosition + Integer.BYTES);
+            boolean invalidMagicCode = magicCode != MessageDecoder.MESSAGE_MAGIC_CODE
+                && magicCode != MessageDecoder.MESSAGE_MAGIC_CODE_V2;
+            if (invalidMagicCode
+                || totalSize <= MessageDecoder.MESSAGE_PHYSIC_OFFSET_POSITION + Long.BYTES
+                || (long) currentPosition + totalSize > relativeOffset
+                || buffer.getLong(currentPosition + MessageDecoder.MESSAGE_PHYSIC_OFFSET_POSITION)
+                != mappedFile.getFileFromOffset() + currentPosition) {
+                throw invalidTruncateMarkerPosition(mappedFile, phyOffset, relativeOffset, currentPosition);
+            }
+            currentPosition += totalSize;
+        }
+
+        if (currentPosition != relativeOffset) {
+            throw invalidTruncateMarkerPosition(mappedFile, phyOffset, relativeOffset, currentPosition);
+        }
+    }
+
+    private IllegalStateException invalidTruncateMarkerPosition(MappedFile mappedFile, long phyOffset,
+        int relativeOffset, int currentPosition) {
+        return new IllegalStateException("Truncate marker offset is not a CommitLog message boundary, offset="
+            + phyOffset + ", file=" + mappedFile.getFileName() + ", relativeOffset=" + relativeOffset
+            + ", validationPosition=" + currentPosition);
+    }
+
     protected void onCommitLogAppend(MessageExtBrokerInner msg, AppendMessageResult result, MappedFile commitLogFile) {
         this.getMessageStore().onCommitLogAppend(msg, result, commitLogFile);
     }
diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
index aee767dae2f..96e96c10314 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -793,17 +793,26 @@ public void truncateDirtyFiles(long offsetToTruncate) throws RocksDBException {
             return;
         }
 
+        long markerOffset = Math.max(offsetToTruncate, this.getMinPhyOffset());
         this.reputMessageService.shutdown();
 
         long oldReputFromOffset = this.reputMessageService.getReputFromOffset();
 
-        // truncate consume queue
-        this.truncateDirtyLogicFiles(offsetToTruncate);
+        try {
+            this.commitLog.persistTruncateMarker(markerOffset);
 
-        // truncate commitLog
-        this.commitLog.truncateDirtyFiles(offsetToTruncate);
+            // truncate consume queue
+            this.truncateDirtyLogicFiles(offsetToTruncate);
 
-        this.recoverTopicQueueTable();
+            // truncate commitLog
+            this.commitLog.truncateDirtyFiles(offsetToTruncate);
+
+            this.recoverTopicQueueTable();
+        } catch (RocksDBException | RuntimeException e) {
+            LOGGER.error("Failed to truncate dirty files to {} after stopping ReputMessageService; "
+                + "keep the service stopped to avoid processing a partially truncated CommitLog", offsetToTruncate, e);
+            throw e;
+        }
 
         if (!messageStoreConfig.isEnableBuildConsumeQueueConcurrently()) {
             this.reputMessageService = new ReputMessageService();
diff --git a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
index 39d837e7bcd..99e8ad33673 100644
--- a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
@@ -65,6 +65,7 @@
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 import org.apache.rocketmq.store.config.StorePathConfigHelper;
 import org.apache.rocketmq.store.exception.ConsumeQueueException;
+import org.apache.rocketmq.store.logfile.MappedFile;
 import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
 import org.apache.rocketmq.store.queue.CqUnit;
 import org.apache.rocketmq.store.stats.BrokerStatsManager;
@@ -728,6 +729,90 @@ public void testRecover() throws Exception {
         }
     }
 
+    @Test
+    public void testTruncateMarkerSurvivesNormalRecovery() throws Exception {
+        verifyTruncateMarkerRecovery(false, false);
+    }
+
+    @Test
+    public void testTruncateMarkerSurvivesAbnormalRecovery() throws Exception {
+        verifyTruncateMarkerRecovery(true, false);
+    }
+
+    @Test
+    public void testTruncateMarkerAtFirstMessageSurvivesAbnormalRecovery() throws Exception {
+        verifyTruncateMarkerRecovery(true, true);
+    }
+
+    @Test
+    public void testPersistTruncateMarkerRejectsNonMessageBoundary() {
+        queueTotal = 1;
+        messageBody = storeMessage.getBytes(StandardCharsets.UTF_8);
+        PutMessageResult firstResult = messageStore.putMessage(buildMessage());
+        PutMessageResult secondResult = messageStore.putMessage(buildMessage());
+        assertThat(firstResult.getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK);
+        assertThat(secondResult.getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK);
+
+        long secondMessageOffset = secondResult.getAppendMessageResult().getWroteOffset();
+        IllegalStateException exception = Assert.assertThrows(IllegalStateException.class,
+            () -> getDefaultMessageStore().getCommitLog().persistTruncateMarker(secondMessageOffset + 1));
+        assertThat(exception.getMessage()).contains("Failed to persist CommitLog truncate marker");
+        assertThat(exception.getCause().getMessage()).contains("not a CommitLog message boundary");
+    }
+
+    private void verifyTruncateMarkerRecovery(boolean recoverAbnormally, boolean truncateFirstMessage)
+        throws Exception {
+        queueTotal = 1;
+        messageBody = storeMessage.getBytes(StandardCharsets.UTF_8);
+        PutMessageResult firstResult = messageStore.putMessage(buildMessage());
+        PutMessageResult secondResult = messageStore.putMessage(buildMessage());
+        PutMessageResult thirdResult = messageStore.putMessage(buildMessage());
+        assertThat(firstResult.getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK);
+        assertThat(secondResult.getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK);
+        assertThat(thirdResult.getPutMessageStatus()).isEqualTo(PutMessageStatus.PUT_OK);
+
+        long firstMessageOffset = firstResult.getAppendMessageResult().getWroteOffset();
+        long truncateOffset = truncateFirstMessage ? firstMessageOffset
+            : secondResult.getAppendMessageResult().getWroteOffset();
+        assertThat(messageStore.truncateFiles(truncateOffset)).isTrue();
+        assertThat(messageStore.getMaxPhyOffset()).isEqualTo(truncateOffset);
+
+        DefaultMessageStore defaultMessageStore = getDefaultMessageStore();
+        MappedFile mappedFile = defaultMessageStore.getCommitLog().getMappedFileQueue()
+            .findMappedFileByOffset(truncateOffset, false);
+        assertThat(mappedFile).isNotNull();
+        int relativeOffset = (int) (truncateOffset - mappedFile.getFileFromOffset());
+        assertThat(mappedFile.getMappedByteBuffer().getLong(relativeOffset)).isZero();
+
+        String storeRootDir = defaultMessageStore.getMessageStoreConfig().getStorePathRootDir();
+        messageStore.shutdown();
+        if (recoverAbnormally) {
+            File abortFile = new File(StorePathConfigHelper.getAbortFile(storeRootDir));
+            UtilAll.ensureDirOK(abortFile.getParent());
+            assertThat(abortFile.createNewFile()).isTrue();
+        }
+
+        MessageStore recoveredMessageStore = null;
+        try {
+            recoveredMessageStore = buildMessageStore(storeRootDir);
+            assertThat(recoveredMessageStore.load()).isTrue();
+            recoveredMessageStore.start();
+            assertThat(recoveredMessageStore.getMaxPhyOffset()).isEqualTo(truncateOffset);
+            if (!truncateFirstMessage) {
+                assertThat(recoveredMessageStore.lookMessageByOffset(firstMessageOffset)).isNotNull();
+            }
+            assertThat(recoveredMessageStore.lookMessageByOffset(truncateOffset)).isNull();
+        } finally {
+            if (recoveredMessageStore != null) {
+                try {
+                    recoveredMessageStore.shutdown();
+                } finally {
+                    recoveredMessageStore.destroy();
+                }
+            }
+        }
+    }
+
     @Test
     public void testStorePathOK() {
         if (messageStore instanceof DefaultMessageStore) {