feat(block-wal): use version V1 and support sequential return (#2682)

* feat(s3stream/block-wal): complete appends sequentially (#2665)

* feat(s3stream/block-wal): complete appends sequentially



* fix: use a lock to ensure there is at most one callback thread



---------



* feat(s3stream/wal): write a padding record when no space at the end of device (#2673)

* refactor(RecordHeader): remove useless methods



* feat(SlidingWindowService): write a padding block when not enough space



* feat(recovery): handle padding records



* fix: fix incorrect assertion



---------



* feat(s3stream/wal): defaults to using version V1 and forward compatible (#2676)

* feat: introduce the Block WAL V1



* feat: impl `RecoverIteratorV1` which only recovers continuous records



* feat: wal forward compatibility



* test: fix tests



* test: test recover from WAL V1



* test: test upgrade



---------



---------

Signed-off-by: Ning Yu <[email protected]>

Author: Chillax-0v0

s3stream/src/main/java/com/automq/stream/s3/wal/common/RecordHeader.java

@@ -26,62 +26,58 @@
 public class RecordHeader {
     public static final int RECORD_HEADER_SIZE = 4 + 4 + 8 + 4 + 4;
     public static final int RECORD_HEADER_WITHOUT_CRC_SIZE = RECORD_HEADER_SIZE - 4;
-    public static final int RECORD_HEADER_MAGIC_CODE = 0x87654321;
-
-    private int magicCode0 = RECORD_HEADER_MAGIC_CODE;
-    private int recordBodyLength1;
-    private long recordBodyOffset2;
-    private int recordBodyCRC3;
+    public static final int RECORD_HEADER_DATA_MAGIC_CODE = 0x87654321;
+    /**
+     * Magic code for record header indicating that the record body is empty (used for padding).
+     */
+    public static final int RECORD_HEADER_EMPTY_MAGIC_CODE = 0x76543210;
+
+    private final int magicCode0;
+    private final int recordBodyLength1;
+    private final long recordBodyOffset2;
+    private final int recordBodyCRC3;
     private int recordHeaderCRC4;
 
-    public static RecordHeader unmarshal(ByteBuf byteBuf) {
-        RecordHeader recordHeader = new RecordHeader();
+    public RecordHeader(long offset, int length, int crc) {
+        this.magicCode0 = RECORD_HEADER_DATA_MAGIC_CODE;
+        this.recordBodyLength1 = length;
+        this.recordBodyOffset2 = offset + RECORD_HEADER_SIZE;
+        this.recordBodyCRC3 = crc;
+    }
+
+    public RecordHeader(long offset, int length) {
+        this.magicCode0 = RECORD_HEADER_EMPTY_MAGIC_CODE;
+        this.recordBodyLength1 = length;
+        this.recordBodyOffset2 = offset + RECORD_HEADER_SIZE;
+        this.recordBodyCRC3 = 0;
+    }
+
+    public RecordHeader(ByteBuf byteBuf) {
         byteBuf.markReaderIndex();
-        recordHeader.magicCode0 = byteBuf.readInt();
-        recordHeader.recordBodyLength1 = byteBuf.readInt();
-        recordHeader.recordBodyOffset2 = byteBuf.readLong();
-        recordHeader.recordBodyCRC3 = byteBuf.readInt();
-        recordHeader.recordHeaderCRC4 = byteBuf.readInt();
+        this.magicCode0 = byteBuf.readInt();
+        this.recordBodyLength1 = byteBuf.readInt();
+        this.recordBodyOffset2 = byteBuf.readLong();
+        this.recordBodyCRC3 = byteBuf.readInt();
+        this.recordHeaderCRC4 = byteBuf.readInt();
         byteBuf.resetReaderIndex();
-        return recordHeader;
     }
 
     public int getMagicCode() {
         return magicCode0;
     }
 
-    public RecordHeader setMagicCode(int magicCode) {
-        this.magicCode0 = magicCode;
-        return this;
-    }
-
     public int getRecordBodyLength() {
         return recordBodyLength1;
     }
 
-    public RecordHeader setRecordBodyLength(int recordBodyLength) {
-        this.recordBodyLength1 = recordBodyLength;
-        return this;
-    }
-
     public long getRecordBodyOffset() {
         return recordBodyOffset2;
     }
 
-    public RecordHeader setRecordBodyOffset(long recordBodyOffset) {
-        this.recordBodyOffset2 = recordBodyOffset;
-        return this;
-    }
-
     public int getRecordBodyCRC() {
         return recordBodyCRC3;
     }
 
-    public RecordHeader setRecordBodyCRC(int recordBodyCRC) {
-        this.recordBodyCRC3 = recordBodyCRC;
-        return this;
-    }
-
     public int getRecordHeaderCRC() {
         return recordHeaderCRC4;
     }
@@ -105,16 +101,10 @@ private ByteBuf marshalHeaderExceptCRC(ByteBuf buf) {
         return buf;
     }
 
-    public ByteBuf marshal(ByteBuf emptyBuf, boolean calculateCRC) {
+    public ByteBuf marshal(ByteBuf emptyBuf) {
         assert emptyBuf.writableBytes() == RECORD_HEADER_SIZE;
         ByteBuf buf = marshalHeaderExceptCRC(emptyBuf);
-
-        if (calculateCRC) {
-            buf.writeInt(WALUtil.crc32(buf, RECORD_HEADER_WITHOUT_CRC_SIZE));
-        } else {
-            buf.writeInt(-1);
-        }
-
+        buf.writeInt(WALUtil.crc32(buf, RECORD_HEADER_WITHOUT_CRC_SIZE));
         return buf;
     }
 }

s3stream/src/main/java/com/automq/stream/s3/wal/impl/object/ObjectWALService.java

@@ -113,7 +113,7 @@ public AppendResult append(TraceContext context, ByteBuf data, int crc) throws O
 
             long expectedWriteOffset = accumulator.append(recordSize, start -> {
                 CompositeByteBuf recordByteBuf = ByteBufAlloc.compositeByteBuffer();
-                Record record = WALUtil.generateRecord(data, header, 0, start, true);
+                Record record = WALUtil.generateRecord(data, header, 0, start);
                 recordByteBuf.addComponents(true, record.header(), record.body());
                 return recordByteBuf;
             }, appendResultFuture);
@@ -324,15 +324,15 @@ public RecoverResult next() {
             tryReadAhead();
 
             ByteBuf recordHeaderBuf = dataBuffer.readBytes(RECORD_HEADER_SIZE);
-            RecordHeader header = RecordHeader.unmarshal(recordHeaderBuf);
+            RecordHeader header = new RecordHeader(recordHeaderBuf);
 
             if (header.getRecordHeaderCRC() != WALUtil.crc32(recordHeaderBuf, RECORD_HEADER_WITHOUT_CRC_SIZE)) {
                 recordHeaderBuf.release();
                 throw new IllegalStateException("Record header crc check failed.");
             }
             recordHeaderBuf.release();
 
-            if (header.getMagicCode() != RecordHeader.RECORD_HEADER_MAGIC_CODE) {
+            if (header.getMagicCode() != RecordHeader.RECORD_HEADER_DATA_MAGIC_CODE) {
                 throw new IllegalStateException("Invalid magic code in record header.");
             }
 

s3stream/src/main/java/com/automq/stream/s3/wal/util/WALUtil.java

@@ -19,6 +19,7 @@
 
 package com.automq.stream.s3.wal.util;
 
+import com.automq.stream.s3.ByteBufAlloc;
 import com.automq.stream.s3.wal.common.Record;
 import com.automq.stream.s3.wal.common.RecordHeader;
 import com.automq.stream.utils.CommandResult;
@@ -38,7 +39,6 @@
 import jnr.posix.POSIX;
 import jnr.posix.POSIXFactory;
 
-import static com.automq.stream.s3.wal.common.RecordHeader.RECORD_HEADER_MAGIC_CODE;
 import static com.automq.stream.s3.wal.common.RecordHeader.RECORD_HEADER_SIZE;
 
 public class WALUtil {
@@ -51,17 +51,19 @@ public class WALUtil {
     private static final Logger LOGGER = LoggerFactory.getLogger(WALUtil.class);
 
     public static Record generateRecord(ByteBuf body, ByteBuf emptyHeader, int crc, long start) {
-        return generateRecord(body, emptyHeader, crc, start, true);
+        crc = 0 == crc ? WALUtil.crc32(body) : crc;
+        ByteBuf header = new RecordHeader(start, body.readableBytes(), crc).marshal(emptyHeader);
+        return new Record(header, body);
     }
 
-    public static Record generateRecord(ByteBuf body, ByteBuf emptyHeader, int crc, long start, boolean calculateCRC) {
-        crc = 0 == crc ? WALUtil.crc32(body) : crc;
-        ByteBuf header = new RecordHeader()
-            .setMagicCode(RECORD_HEADER_MAGIC_CODE)
-            .setRecordBodyLength(body.readableBytes())
-            .setRecordBodyOffset(start + RECORD_HEADER_SIZE)
-            .setRecordBodyCRC(crc)
-            .marshal(emptyHeader, calculateCRC);
+    public static Record generatePaddingRecord(ByteBuf emptyHeader, long start, int length) {
+        int bodyLength = length - RECORD_HEADER_SIZE;
+
+        ByteBuf header = new RecordHeader(start, bodyLength).marshal(emptyHeader);
+
+        ByteBuf body = ByteBufAlloc.byteBuffer(bodyLength);
+        body.writeZero(bodyLength);
+
         return new Record(header, body);
     }
 

s3stream/src/test/java/com/automq/stream/s3/wal/impl/object/ObjectWALServiceTest.java

@@ -331,7 +331,7 @@ private void writeV1Object(ObjectStorage objectStorage, ByteBuf data, long start
 
     private ByteBuf addRecordHeader(ByteBuf data, long startOffset) {
         ByteBuf header = ByteBufAlloc.byteBuffer(RECORD_HEADER_SIZE);
-        Record record = WALUtil.generateRecord(data, header, 0, startOffset, true);
+        Record record = WALUtil.generateRecord(data, header, 0, startOffset);
 
         CompositeByteBuf buffer = ByteBufAlloc.compositeByteBuffer();
         buffer.addComponents(true, record.header(), record.body());