feat(wal): add LogRecord, Begin and WalCorruptionException

src/main/kotlin/kestrel/store/wal/LogRecord.kt

@@ -0,0 +1,50 @@
+package kestrel.store.wal
+
+/**
+ * A single write-ahead log entry. [Update] carries whole-page before/after
+ * images rather than a delta, so redoing it is just overwriting a page with
+ * [Update.after]; undoing it is overwriting with [Update.before]. `prevLsn`
+ * threads a transaction's updates into a backward-traversable chain, the
+ * same role it plays in ARIES.
+ */
+sealed class LogRecord {
+    class Begin(val txnId: Long) : LogRecord()
+
+    class Commit(val txnId: Long) : LogRecord()
+
+    class Abort(val txnId: Long) : LogRecord()
+
+    class Update(
+        val txnId: Long,
+        val prevLsn: Long,
+        val pageId: Long,
+        val before: ByteArray,
+        val after: ByteArray
+    ) : LogRecord()
+
+    /**
+     * A compensation log record (CLR), ARIES terminology for "this page was just
+     * restored to an earlier image by an undo". Written by [kestrel.store.wal.RecoveryManager.rollbackTo]
+     * for every [Update] it undoes on the way to a savepoint, so that:
+     * - forward redo ([kestrel.store.wal.RecoveryManager.recover]) can reproduce the undo's effect
+     *   just like any other logged write, by reapplying [undoImage].
+     * - a later undo walk (a full [kestrel.store.wal.RecoveryManager.rollback], or a rollback to an
+     *   earlier savepoint) that reaches this record never re-undoes the [Update] it compensates for;
+     *   it jumps straight to [undoNextLsn] instead, which is that [Update]'s own `prevLsn` - the
+     *   transaction's undo chain skips the already-undone range as if it had never happened, exactly
+     *   the property that makes partial rollback safe to repeat and safe to crash through.
+     * Chains into the transaction's `prevLsn` thread the same way [Update] does, via [prevLsn] - so a
+     * write logged after a savepoint rollback correctly points back at this CLR, not at the undone
+     * [Update] it replaced.
+     */
+    class CompensationUpdate(
+        val txnId: Long,
+        val prevLsn: Long,
+        val pageId: Long,
+        val undoImage: ByteArray,
+        val undoNextLsn: Long
+    ) : LogRecord()
+
+    /** Marks a quiescent checkpoint: written only when no transaction is active. */
+    class Checkpoint(val activeTxnIds: List<Long>) : LogRecord()
+}

src/main/kotlin/kestrel/store/wal/LogRecordCodec.kt

@@ -0,0 +1,85 @@
+package kestrel.store.wal
+
+import kestrel.KestrelException
+import kestrel.store.codec.ByteReader
+import kestrel.store.codec.ByteWriter
+
+class WalCorruptionException(message: String) : KestrelException(message)
+
+/** Serializes [LogRecord]s to and from the byte payload stored in each log frame. */
+object LogRecordCodec {
+    private const val TYPE_BEGIN = 1
+    private const val TYPE_COMMIT = 2
+    private const val TYPE_ABORT = 3
+    private const val TYPE_UPDATE = 4
+    private const val TYPE_CHECKPOINT = 5
+    private const val TYPE_COMPENSATION_UPDATE = 6
+
+    fun encode(record: LogRecord): ByteArray {
+        val writer = ByteWriter()
+        when (record) {
+            is LogRecord.Begin -> {
+                writer.writeByte(TYPE_BEGIN)
+                writer.writeFixedLong(record.txnId)
+            }
+            is LogRecord.Commit -> {
+                writer.writeByte(TYPE_COMMIT)
+                writer.writeFixedLong(record.txnId)
+            }
+            is LogRecord.Abort -> {
+                writer.writeByte(TYPE_ABORT)
+                writer.writeFixedLong(record.txnId)
+            }
+            is LogRecord.Update -> {
+                writer.writeByte(TYPE_UPDATE)
+                writer.writeFixedLong(record.txnId)
+                writer.writeFixedLong(record.prevLsn)
+                writer.writeFixedLong(record.pageId)
+                writer.writeLengthPrefixed(record.before)
+                writer.writeLengthPrefixed(record.after)
+            }
+            is LogRecord.Checkpoint -> {
+                writer.writeByte(TYPE_CHECKPOINT)
+                writer.writeVarInt(record.activeTxnIds.size.toLong())
+                for (id in record.activeTxnIds) writer.writeFixedLong(id)
+            }
+            is LogRecord.CompensationUpdate -> {
+                writer.writeByte(TYPE_COMPENSATION_UPDATE)
+                writer.writeFixedLong(record.txnId)
+                writer.writeFixedLong(record.prevLsn)
+                writer.writeFixedLong(record.pageId)
+                writer.writeLengthPrefixed(record.undoImage)
+                writer.writeFixedLong(record.undoNextLsn)
+            }
+        }
+        return writer.toByteArray()
+    }
+
+    fun decode(bytes: ByteArray): LogRecord {
+        val reader = ByteReader(bytes)
+        return when (val type = reader.readByte()) {
+            TYPE_BEGIN -> LogRecord.Begin(reader.readFixedLong())
+            TYPE_COMMIT -> LogRecord.Commit(reader.readFixedLong())
+            TYPE_ABORT -> LogRecord.Abort(reader.readFixedLong())
+            TYPE_UPDATE -> LogRecord.Update(
+                txnId = reader.readFixedLong(),
+                prevLsn = reader.readFixedLong(),
+                pageId = reader.readFixedLong(),
+                before = reader.readLengthPrefixed(),
+                after = reader.readLengthPrefixed()
+            )
+            TYPE_CHECKPOINT -> {
+                val count = reader.readVarInt().toInt()
+                LogRecord.Checkpoint((0 until count).map { reader.readFixedLong() })
+            }
+            TYPE_COMPENSATION_UPDATE -> LogRecord.CompensationUpdate(
+                txnId = reader.readFixedLong(),
+                prevLsn = reader.readFixedLong(),
+                pageId = reader.readFixedLong(),
+                undoImage = reader.readLengthPrefixed(),
+                undoNextLsn = reader.readFixedLong()
+            )
+            else -> throw WalCorruptionException("unknown log record type $type")
+        }
+    }
+}