diff --git a/src/main/kotlin/kestrel/store/wal/LogRecord.kt b/src/main/kotlin/kestrel/store/wal/LogRecord.kt new file mode 100644 index 0000000..04f4f0a --- /dev/null +++ b/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) : LogRecord() +} diff --git a/src/main/kotlin/kestrel/store/wal/LogRecordCodec.kt b/src/main/kotlin/kestrel/store/wal/LogRecordCodec.kt new file mode 100644 index 0000000..7f0f555 --- /dev/null +++ b/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") + } + } +}