From 5e7bac4787088db98d04959f5afab9b81b229530 Mon Sep 17 00:00:00 2001
From: ad hoc <postma.marin@protonmail.com>
Date: Fri, 20 Jan 2023 16:27:27 +0100
Subject: [PATCH 1/2] Add required dependencies

---
 Cargo.lock      | 168 +++++++-----------------------------------------
 sqld/Cargo.toml |   4 +-
 2 files changed, 27 insertions(+), 145 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 9bd88da4..c8245588 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -497,7 +497,7 @@ dependencies = [
  "http-body",
  "hyper",
  "itoa",
- "matchit 0.7.0",
+ "matchit",
  "memchr",
  "mime",
  "percent-encoding",
@@ -711,6 +711,26 @@ version = "0.6.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2c676a478f63e9fa2dd5368a42f28bba0d6c560b775f38583c8bbaa7fcd67c9c"
 
+[[package]]
+name = "bytemuck"
+version = "1.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c041d3eab048880cb0b86b256447da3f18859a163c3b8d8893f4e6368abe6393"
+dependencies = [
+ "bytemuck_derive",
+]
+
+[[package]]
+name = "bytemuck_derive"
+version = "1.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1aca418a974d83d40a0c1f0c5cba6ff4bc28d8df099109ca459a2118d40b6322"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "byteorder"
 version = "1.4.3"
@@ -813,28 +833,6 @@ dependencies = [
  "winapi",
 ]
 
-[[package]]
-name = "chrono-tz"
-version = "0.6.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "29c39203181991a7dd4343b8005bd804e7a9a37afb8ac070e43771e8c820bbde"
-dependencies = [
- "chrono",
- "chrono-tz-build",
- "phf",
-]
-
-[[package]]
-name = "chrono-tz-build"
-version = "0.0.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6f509c3a87b33437b05e2458750a0700e5bdd6956176773e6c7d6dd15a283a0c"
-dependencies = [
- "parse-zoneinfo",
- "phf",
- "phf_codegen",
-]
-
 [[package]]
 name = "clang-sys"
 version = "1.4.0"
@@ -2115,23 +2113,6 @@ version = "0.2.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "348108ab3fba42ec82ff6e9564fc4ca0247bdccdc68dd8af9764bbc79c3c8ffb"
 
-[[package]]
-name = "libsql-client"
-version = "0.4.1"
-dependencies = [
- "anyhow",
- "async-trait",
- "base64 0.21.0",
- "libsql-client",
- "num-traits",
- "rand",
- "reqwest",
- "serde_json",
- "tokio",
- "url",
- "worker",
-]
-
 [[package]]
 name = "libsql-wasmtime-bindings"
 version = "0.1.0"
@@ -2234,12 +2215,6 @@ dependencies = [
  "regex-automata",
 ]
 
-[[package]]
-name = "matchit"
-version = "0.4.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9376a4f0340565ad675d11fc1419227faf5f60cd7ac9cb2e7185a471f30af833"
-
 [[package]]
 name = "matchit"
 version = "0.7.0"
@@ -2655,15 +2630,6 @@ dependencies = [
  "windows-sys 0.45.0",
 ]
 
-[[package]]
-name = "parse-zoneinfo"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c705f256449c60da65e11ff6626e0c16a0a0b96aaa348de61376b249bc340f41"
-dependencies = [
- "regex",
-]
-
 [[package]]
 name = "password-hash"
 version = "0.4.2"
@@ -3254,7 +3220,6 @@ dependencies = [
  "http",
  "http-body",
  "hyper",
- "hyper-rustls",
  "hyper-tls",
  "ipnet",
  "js-sys",
@@ -3264,20 +3229,16 @@ dependencies = [
  "once_cell",
  "percent-encoding",
  "pin-project-lite",
- "rustls",
- "rustls-pemfile",
  "serde",
  "serde_json",
  "serde_urlencoded",
  "tokio",
  "tokio-native-tls",
- "tokio-rustls",
  "tower-service",
  "url",
  "wasm-bindgen",
  "wasm-bindgen-futures",
  "web-sys",
- "webpki-roots",
  "winreg",
 ]
 
@@ -3763,9 +3724,11 @@ dependencies = [
  "base64 0.21.0",
  "bincode",
  "bottomless",
+ "bytemuck",
  "byteorder",
  "bytes 1.4.0",
  "clap",
+ "crc",
  "crossbeam",
  "fallible-iterator",
  "futures",
@@ -4448,6 +4411,7 @@ source = "git+https://github.com/psarna/uuid?branch=stable_v7#044e1f812d9ac9e154
 dependencies = [
  "atomic",
  "getrandom",
+ "serde",
 ]
 
 [[package]]
@@ -4541,8 +4505,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "31f8dcbc21f30d9b8f2ea926ecb58f6b91192c17e9d33594b3df58b2007ca53b"
 dependencies = [
  "cfg-if",
- "serde",
- "serde_json",
  "wasm-bindgen-macro",
 ]
 
@@ -4611,19 +4573,6 @@ dependencies = [
  "leb128",
 ]
 
-[[package]]
-name = "wasm-streams"
-version = "0.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6bbae3363c08332cadccd13b67db371814cd214c2524020932f0804b8cf7c078"
-dependencies = [
- "futures-util",
- "js-sys",
- "wasm-bindgen",
- "wasm-bindgen-futures",
- "web-sys",
-]
-
 [[package]]
 name = "wasmparser"
 version = "0.92.0"
@@ -5046,75 +4995,6 @@ dependencies = [
  "winapi",
 ]
 
-[[package]]
-name = "worker"
-version = "0.0.12"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ca55ce51b3bf01da5a20598af220c5d3abf91f1978a50a0620ef966c39e3180"
-dependencies = [
- "async-trait",
- "chrono",
- "chrono-tz",
- "futures-channel",
- "futures-util",
- "http",
- "js-sys",
- "matchit 0.4.6",
- "pin-project 1.0.12",
- "serde",
- "serde_json",
- "url",
- "wasm-bindgen",
- "wasm-bindgen-futures",
- "wasm-streams",
- "web-sys",
- "worker-kv",
- "worker-macros",
- "worker-sys",
-]
-
-[[package]]
-name = "worker-kv"
-version = "0.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "682cbd728f179cc810b2ab77a2534da817b973e190ab184ab8efe1058b0dba84"
-dependencies = [
- "js-sys",
- "serde",
- "serde_json",
- "thiserror",
- "wasm-bindgen",
- "wasm-bindgen-futures",
-]
-
-[[package]]
-name = "worker-macros"
-version = "0.0.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c29ce5a41f5e7e644bc683681b62aed3adac838e01d18eb4e02a4dc30d4ac69"
-dependencies = [
- "async-trait",
- "proc-macro2",
- "quote",
- "syn",
- "wasm-bindgen",
- "wasm-bindgen-futures",
- "wasm-bindgen-macro-support",
- "worker-sys",
-]
-
-[[package]]
-name = "worker-sys"
-version = "0.0.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "825732b9b6360d6b1f5f614248317826cebf4878e36f61ccc71ca9dd53de8b41"
-dependencies = [
- "cfg-if",
- "js-sys",
- "wasm-bindgen",
- "web-sys",
-]
-
 [[package]]
 name = "xattr"
 version = "0.2.3"
diff --git a/sqld/Cargo.toml b/sqld/Cargo.toml
index eee7a773..1e84f69d 100644
--- a/sqld/Cargo.toml
+++ b/sqld/Cargo.toml
@@ -11,8 +11,10 @@ base64 = "0.21.0"
 bincode = "1.3.3"
 bottomless = { version = "0", path = "../bottomless", features = ["libsql_linked_statically"] }
 byteorder = "1.4.3"
+bytemuck = { version = "1.13.0", features = ["derive"] }
 bytes = { version = "1.2.1", features = ["serde"] }
 clap = { version = "4.0.23", features = [ "derive", "env"] }
+crc = "3.0.0"
 crossbeam = "0.8.2"
 fallible-iterator = "0.2.0"
 futures = "0.3.25"
@@ -49,7 +51,7 @@ tower = { version = "0.4.13", features = ["make"] }
 tower-http = { version = "0.3.5", features = ["compression-full", "cors"] }
 tracing = "0.1.37"
 tracing-subscriber = { version = "0.3.16", features = ["env-filter"] }
-uuid = { version = "1.3", features = ["v4"] }
+uuid = { version = "1.3", features = ["v4", "serde"] }
 
 [dev-dependencies]
 proptest = "1.0.0"

From 8a43dba39feadfe74f9203a706247165cbdd5524 Mon Sep 17 00:00:00 2001
From: ad hoc <postma.marin@protonmail.com>
Date: Fri, 20 Jan 2023 16:28:38 +0100
Subject: [PATCH 2/2] improve WAL log file structure and protocol

---
 sqld/proto/wal_log.proto                     |  23 +-
 sqld/src/database/write_proxy/replication.rs | 111 +++---
 sqld/src/rpc/wal_log.rs                      |  50 +--
 sqld/src/wal_logger.rs                       | 341 ++++++++++++++-----
 4 files changed, 322 insertions(+), 203 deletions(-)

diff --git a/sqld/proto/wal_log.proto b/sqld/proto/wal_log.proto
index 73976ef5..e814afbe 100644
--- a/sqld/proto/wal_log.proto
+++ b/sqld/proto/wal_log.proto
@@ -5,35 +5,18 @@ message LogOffset {
     uint64 startOffset = 1;
 }
 
-message HelloRequest {
-}
+message HelloRequest { }
 
 message HelloResponse {
     string generation_id = 1;
     uint64 first_frame_id_in_generation = 2;
 }
 
-message WalLogEntry {
-    uint64 index = 1;
-    oneof Payload {
-        Frame frame = 2;
-        Commit commit = 3;
-    }
-}
-
 message Frame {
-    uint32 page_no = 1;
-    bytes data = 2;
-}
-
-message Commit {
-    int32 page_size = 1;
-    uint32 size_after = 2;
-    bool is_commit = 3;
-    int32 sync_flags = 4;
+    bytes data = 1;
 }
 
 service WalLog {
-    rpc LogEntries(LogOffset) returns (stream WalLogEntry) {}
     rpc Hello(HelloRequest) returns (HelloResponse) {}
+    rpc LogEntries(LogOffset) returns (stream Frame) {}
 }
diff --git a/sqld/src/database/write_proxy/replication.rs b/sqld/src/database/write_proxy/replication.rs
index c95bc0dd..ab7c7b4b 100644
--- a/sqld/src/database/write_proxy/replication.rs
+++ b/sqld/src/database/write_proxy/replication.rs
@@ -33,9 +33,8 @@ use tonic::transport::Channel;
 use crate::libsql::ffi::{types::XWalFrameFn, PgHdr, Wal};
 use crate::libsql::open_with_regular_wal;
 use crate::libsql::wal_hook::WalHook;
-use crate::rpc::wal_log::wal_log_rpc::wal_log_entry::Payload;
-use crate::rpc::wal_log::wal_log_rpc::{wal_log_client::WalLogClient, LogOffset, WalLogEntry};
-use crate::rpc::wal_log::wal_log_rpc::{Commit, Frame};
+use crate::rpc::wal_log::wal_log_rpc::{wal_log_client::WalLogClient, LogOffset};
+use crate::wal_logger::{WalFrame, WAL_PAGE_SIZE};
 
 pub struct PeriodicDbUpdater {
     interval: Duration,
@@ -92,7 +91,7 @@ struct ReadReplicationHook {
     last_applied_index_file: File,
     last_applied_index: Option<u64>,
     /// Buffer for incoming frames
-    buffer: VecDeque<WalLogEntry>,
+    buffer: VecDeque<WalFrame>,
     rt: Handle,
 }
 
@@ -120,7 +119,7 @@ unsafe impl WalHook for ReadReplicationHook {
         _page_headers: *mut PgHdr,
         _size_after: u32,
         _is_commit: c_int,
-        _sync_flags: c_int,
+        sync_flags: c_int,
         orig: XWalFrameFn,
     ) -> c_int {
         let rt = self.rt.clone();
@@ -129,33 +128,21 @@ unsafe impl WalHook for ReadReplicationHook {
             return SQLITE_ERROR;
         }
 
-        while let Some((page_headers, truncate, commit)) = self.next_transaction() {
-            tracing::trace!(commit = ?commit, truncate = truncate);
+        while let Some((page_headers, frame_count)) = self.next_transaction() {
+            let size_after = self.buffer[frame_count - 1].header.size_after;
+            assert_ne!(size_after, 0, "commit index points to non commit frame");
+
+            tracing::trace!(commit = ?frame_count, size_after = size_after);
             let attempted_commit_index = self
                 .last_applied_index
-                .map(|x| x + truncate as u64 - 1)
+                .map(|x| x + frame_count as u64)
                 .unwrap_or_default();
             // pre-write index
             self.last_applied_index_file
                 .write_all_at(&attempted_commit_index.to_le_bytes(), 0)
                 .unwrap();
-            let Commit {
-                page_size,
-                size_after,
-                is_commit,
-                sync_flags,
-            } = commit;
 
-            let ret = unsafe {
-                orig(
-                    wal,
-                    page_size,
-                    page_headers,
-                    size_after,
-                    is_commit as _,
-                    sync_flags,
-                )
-            };
+            let ret = unsafe { orig(wal, WAL_PAGE_SIZE, page_headers, size_after, 1, sync_flags) };
 
             if ret == 0 {
                 debug_assert!(all_applied(page_headers));
@@ -165,7 +152,7 @@ unsafe impl WalHook for ReadReplicationHook {
                     .unwrap();
                 self.last_applied_index.replace(attempted_commit_index);
                 // remove commited entries.
-                self.buffer.drain(..truncate);
+                self.buffer.drain(..frame_count);
                 tracing::trace!("applied frame batch");
             } else {
                 // should we retry?
@@ -180,31 +167,29 @@ unsafe impl WalHook for ReadReplicationHook {
     }
 }
 
-/// Turn a list of `WalLogEntry` into a list of PgHdr.
+/// Turn a list of `WalFrame` into a list of PgHdr.
 /// The caller has the responsibility to free the returned headers.
-fn make_page_header<'a>(entries: impl Iterator<Item = &'a WalLogEntry>) -> *mut PgHdr {
+fn make_page_header<'a>(frames: impl Iterator<Item = &'a WalFrame>) -> *mut PgHdr {
     let mut current_pg = std::ptr::null_mut();
 
     let mut headers_count = 0;
-    for entry in entries {
-        if let Payload::Frame(Frame { page_no, data }) = entry.payload.as_ref().unwrap() {
-            let page = PgHdr {
-                pPage: std::ptr::null_mut(),
-                pData: data.as_ptr() as _,
-                pExtra: std::ptr::null_mut(),
-                pCache: std::ptr::null_mut(),
-                pDirty: current_pg,
-                pPager: std::ptr::null_mut(),
-                pgno: *page_no,
-                pageHash: 0,
-                flags: 0,
-                nRef: 0,
-                pDirtyNext: std::ptr::null_mut(),
-                pDirtyPrev: std::ptr::null_mut(),
-            };
-            headers_count += 1;
-            current_pg = Box::into_raw(Box::new(page));
-        }
+    for frame in frames {
+        let page = PgHdr {
+            pPage: std::ptr::null_mut(),
+            pData: frame.data.as_ptr() as _,
+            pExtra: std::ptr::null_mut(),
+            pCache: std::ptr::null_mut(),
+            pDirty: current_pg,
+            pPager: std::ptr::null_mut(),
+            pgno: frame.header.page_no,
+            pageHash: 0,
+            flags: 0,
+            nRef: 0,
+            pDirtyNext: std::ptr::null_mut(),
+            pDirtyPrev: std::ptr::null_mut(),
+        };
+        headers_count += 1;
+        current_pg = Box::into_raw(Box::new(page));
     }
 
     tracing::trace!("built {headers_count} page headers");
@@ -265,19 +250,22 @@ impl ReadReplicationHook {
     /// Note: It does not seem possible to batch transaction. I suspect that this is because the
     /// original implementation of the sqlite WAL overwrites when pages appear multiple times in
     /// the same transaction.
-    fn next_transaction(&self) -> Option<(*mut PgHdr, usize, Commit)> {
-        let (commit_idx, commit) =
-            self.buffer
-                .iter()
-                .enumerate()
-                .find_map(|(i, e)| match &e.payload {
-                    Some(Payload::Commit(commit)) => Some((i, commit.clone())),
-                    _ => None,
-                })?;
+    fn next_transaction(&self) -> Option<(*mut PgHdr, usize)> {
+        // nothing to do yet.
+        if self.buffer.is_empty() {
+            return None;
+        }
+
+        let frame_count = self
+            .buffer
+            .iter()
+            .enumerate()
+            .find_map(|(i, f)| (f.header.size_after != 0).then_some(i + 1))?; // early return if
+                                                                              // missing commit frame.
 
-        let headers = make_page_header(self.buffer.iter().take(commit_idx));
+        let headers = make_page_header(self.buffer.iter().take(frame_count));
 
-        Some((headers, commit_idx + 1, commit))
+        Some((headers, frame_count))
     }
 
     /// Asks the writer for new log frames to apply.
@@ -287,13 +275,14 @@ impl ReadReplicationHook {
         let req = LogOffset { start_offset };
 
         let mut stream = self.logger.log_entries(req).await?.into_inner();
-        while let Some(frame) = stream.next().await {
-            let frame = frame?;
+        while let Some(raw_frame) = stream.next().await {
+            let raw_frame = raw_frame?;
+            let frame = WalFrame::decode(raw_frame.data)?;
             debug_assert_eq!(
-                frame.index, self.fetch_frame_index,
+                frame.header.frame_id, self.fetch_frame_index,
                 "out of order log frame"
             );
-            self.fetch_frame_index = frame.index + 1;
+            self.fetch_frame_index = frame.header.frame_id + 1;
             self.buffer.push_back(frame);
         }
 
diff --git a/sqld/src/rpc/wal_log.rs b/sqld/src/rpc/wal_log.rs
index 711bcac1..773d3071 100644
--- a/sqld/src/rpc/wal_log.rs
+++ b/sqld/src/rpc/wal_log.rs
@@ -6,7 +6,7 @@ pub mod wal_log_rpc {
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use crate::wal_logger::{WalLogEntry, WalLogger};
+use crate::wal_logger::{FrameId, WalLogger};
 
 use std::collections::HashSet;
 use std::sync::RwLock;
@@ -14,10 +14,7 @@ use tokio_stream::wrappers::ReceiverStream;
 use tonic::Status;
 use wal_log_rpc::wal_log_server::WalLog;
 
-use self::wal_log_rpc::{
-    wal_log_entry::Payload, Frame, HelloRequest, HelloResponse, LogOffset,
-    WalLogEntry as RpcWalLogEntry,
-};
+use self::wal_log_rpc::{Frame, HelloRequest, HelloResponse, LogOffset};
 
 pub struct WalLogService {
     logger: Arc<WalLogger>,
@@ -26,29 +23,6 @@ pub struct WalLogService {
 
 const _NO_HELLO_ERROR_MSG: &str = "NO_HELLO";
 
-impl From<(u64, WalLogEntry)> for RpcWalLogEntry {
-    fn from((index, entry): (u64, WalLogEntry)) -> Self {
-        let payload = match entry {
-            WalLogEntry::Frame { page_no, data } => Payload::Frame(Frame { page_no, data }),
-            WalLogEntry::Commit {
-                page_size,
-                size_after,
-                is_commit,
-                sync_flags,
-            } => Payload::Commit(wal_log_rpc::Commit {
-                page_size,
-                size_after,
-                is_commit,
-                sync_flags,
-            }),
-        };
-        Self {
-            index,
-            payload: Some(payload),
-        }
-    }
-}
-
 impl WalLogService {
     pub fn new(logger: Arc<WalLogger>) -> Self {
         Self {
@@ -57,20 +31,22 @@ impl WalLogService {
         }
     }
 
-    fn stream_pages(&self, start_offset: usize) -> ReceiverStream<Result<RpcWalLogEntry, Status>> {
+    fn stream_pages(&self, start_id: FrameId) -> ReceiverStream<Result<Frame, Status>> {
         let logger = self.logger.clone();
         let (sender, receiver) = tokio::sync::mpsc::channel(64);
         tokio::task::spawn_blocking(move || {
-            let mut offset = start_offset;
+            let mut offset = start_id;
             loop {
-                match logger.get_entry(offset) {
+                match logger.frame_bytes(offset) {
                     Ok(None) => break,
-                    Ok(Some(entry)) => {
-                        let entry: RpcWalLogEntry = (offset as u64, entry).into();
-                        let _ = sender.blocking_send(Ok(entry));
+                    Ok(Some(data)) => {
+                        if let Err(e) = sender.blocking_send(Ok(Frame { data })) {
+                            tracing::error!("failed to send frame: {e}");
+                            break;
+                        }
                         offset += 1;
                     }
-                    Err(_) => todo!(),
+                    Err(e) => todo!("{e}"),
                 }
             }
         });
@@ -81,7 +57,8 @@ impl WalLogService {
 
 #[tonic::async_trait]
 impl WalLog for WalLogService {
-    type LogEntriesStream = ReceiverStream<Result<RpcWalLogEntry, Status>>;
+    type LogEntriesStream = ReceiverStream<Result<Frame, Status>>;
+
     async fn log_entries(
         &self,
         req: tonic::Request<LogOffset>,
@@ -102,6 +79,7 @@ impl WalLog for WalLogService {
         let stream = self.stream_pages(start_offset as _);
         Ok(tonic::Response::new(stream))
     }
+
     async fn hello(
         &self,
         req: tonic::Request<HelloRequest>,
diff --git a/sqld/src/wal_logger.rs b/sqld/src/wal_logger.rs
index c7e5b969..d0a3ce0c 100644
--- a/sqld/src/wal_logger.rs
+++ b/sqld/src/wal_logger.rs
@@ -1,14 +1,18 @@
 use std::ffi::{c_int, c_void};
 use std::fs::{File, OpenOptions};
-use std::io::{Cursor, Read, Write};
-use std::ops::DerefMut;
+use std::io::{Read, Write};
+use std::mem::size_of;
 use std::os::unix::prelude::FileExt;
 use std::path::Path;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 
-use bytes::{Bytes, BytesMut};
+use anyhow::ensure;
+use bytemuck::{cast_ref, try_from_bytes, try_pod_read_unaligned, Pod, Zeroable};
+use bytes::{BufMut, Bytes, BytesMut};
+use crc::Crc;
 use parking_lot::Mutex;
-use serde::{Deserialize, Serialize};
+use uuid::Uuid;
 
 use crate::libsql::ffi::{
     types::{XWalFrameFn, XWalUndoFn},
@@ -17,12 +21,18 @@ use crate::libsql::ffi::{
 use crate::libsql::{ffi::PageHdrIter, wal_hook::WalHook};
 use crate::rpc::wal_log::wal_log_rpc::HelloResponse;
 
+pub const WAL_PAGE_SIZE: i32 = 4096;
+const WAL_MAGIC: u64 = u64::from_le_bytes(*b"SQLDWAL\0");
+const CRC_64_GO_ISO: Crc<u64> = Crc::<u64>::new(&crc::CRC_64_GO_ISO);
+
+pub type FrameId = u64;
+
 // Clone is necessary only because opening a database may fail, and we need to clone the empty
 // struct.
 #[derive(Clone)]
 pub struct WalLoggerHook {
     /// Current frame index, updated on each commit.
-    buffer: Vec<WalLogEntry>,
+    buffer: Vec<WalPage>,
     logger: Arc<WalLogger>,
 }
 
@@ -58,7 +68,7 @@ unsafe impl WalHook for WalLoggerHook {
         }
 
         if is_commit != 0 {
-            self.commit(page_size, ntruncate, is_commit != 0, sync_flags);
+            self.commit(ntruncate);
         }
 
         rc
@@ -76,18 +86,40 @@ unsafe impl WalHook for WalLoggerHook {
     }
 }
 
-#[derive(Serialize, Deserialize, Clone)]
-pub enum WalLogEntry {
-    Frame {
-        page_no: u32,
-        data: Bytes,
-    },
-    Commit {
-        page_size: i32,
-        size_after: u32,
-        is_commit: bool,
-        sync_flags: i32,
-    },
+#[derive(Clone)]
+struct WalPage {
+    page_no: u32,
+    /// 0 for non-commit frames
+    size_after: u32,
+    data: Bytes,
+}
+
+#[derive(Clone)]
+/// A buffered WalFrame.
+/// Cloning this is cheap.
+pub struct WalFrame {
+    pub header: WalFrameHeader,
+    pub data: Bytes,
+}
+
+impl WalFrame {
+    fn encode<B: BufMut>(&self, mut buf: B) {
+        self.header.encode(&mut buf);
+        // FIXME: unnecessary data copy. (clone is ok, since it's Bytes)
+        buf.put(&mut self.data.clone());
+    }
+
+    pub fn decode(mut data: Bytes) -> anyhow::Result<Self> {
+        let header_bytes = data.split_to(size_of::<WalFrameHeader>());
+        ensure!(
+            data.len() == WAL_PAGE_SIZE as usize,
+            "invalid frame size, expected: {}, found: {}",
+            WAL_PAGE_SIZE,
+            data.len()
+        );
+        let header = WalFrameHeader::decode(&header_bytes)?;
+        Ok(Self { header, data })
+    }
 }
 
 impl WalLoggerHook {
@@ -99,22 +131,17 @@ impl WalLoggerHook {
     }
 
     fn write_frame(&mut self, page_no: u32, data: &[u8]) {
-        let entry = WalLogEntry::Frame {
+        let entry = WalPage {
             page_no,
+            size_after: 0,
             data: Bytes::copy_from_slice(data),
         };
         self.buffer.push(entry);
     }
 
-    fn commit(&mut self, page_size: i32, size_after: u32, is_commit: bool, sync_flags: i32) {
-        let entry = WalLogEntry::Commit {
-            page_size,
-            size_after,
-            is_commit,
-            sync_flags,
-        };
-        self.buffer.push(entry);
-        self.logger.append(&self.buffer);
+    fn commit(&mut self, size_after: u32) {
+        self.buffer.last_mut().unwrap().size_after = size_after;
+        self.logger.push_page(&self.buffer);
         self.buffer.clear();
     }
 
@@ -123,25 +150,77 @@ impl WalLoggerHook {
     }
 }
 
+#[derive(Debug, Clone, Copy, Zeroable, Pod)]
+#[repr(C)]
+struct WalLoggerFileHeader {
+    /// magic number: b"SQLDWAL\0" as u64
+    magic: u64,
+    /// Initial checksum value for the rolling CRC checksum
+    /// computed with the 64 bits CRC_64_GO_ISO
+    start_checksum: u64,
+    /// Uuid of the database associated with this log.
+    db_id: u128,
+    /// Frame index of the first frame in the log
+    start_frame_id: u64,
+    /// Wal file version number, currently: 1
+    version: u32,
+    /// page size: 4096
+    page_size: i32,
+}
+
+impl WalLoggerFileHeader {
+    fn decode(buf: &[u8]) -> anyhow::Result<Self> {
+        let this: Self =
+            try_pod_read_unaligned(buf).map_err(|_e| anyhow::anyhow!("invalid WAL log header"))?;
+        ensure!(this.magic == WAL_MAGIC, "invalid WAL log header");
+
+        Ok(this)
+    }
+
+    fn encode<B: BufMut>(&self, mut buf: B) {
+        buf.put(&cast_ref::<_, [u8; size_of::<Self>()]>(self)[..]);
+    }
+}
+
+/// The file header for the WAL log. All fields are represented in little-endian ordering.
+/// See `encode` and `decode` for actual layout.
+// repr C for stable sizing
+#[repr(C)]
+#[derive(Debug, Clone, Copy, Zeroable, Pod)]
+pub struct WalFrameHeader {
+    /// Incremental frame id
+    pub frame_id: u64,
+    /// Rolling checksum of all the previous frames, including this one.
+    pub checksum: u64,
+    /// page number, if frame_type is FrameType::Page
+    pub page_no: u32,
+    pub size_after: u32,
+}
+
+impl WalFrameHeader {
+    fn encode<B: BufMut>(&self, mut buf: B) {
+        buf.put(&cast_ref::<_, [u8; size_of::<Self>()]>(self)[..]);
+    }
+
+    fn decode(buf: &[u8]) -> anyhow::Result<Self> {
+        let this = try_from_bytes(buf).map_err(|_e| anyhow::anyhow!("invalid frame header"))?;
+        Ok(*this)
+    }
+}
+
 pub struct WalLogger {
-    current_offset: Mutex<usize>,
+    /// offset id of the next Frame to write into the log
+    next_frame_id: Mutex<FrameId>,
     /// first index present in the file
-    start_offset: usize,
+    start_frame_id: FrameId,
     log_file: File,
+    current_checksum: AtomicU64,
     hello_response: HelloResponse,
 }
 
-#[derive(Serialize, Deserialize)]
-struct WalLoggerFileHeader {
-    version: u8,
-    start_index: u64,
-}
-
 impl WalLogger {
     /// size of a single frame
-    pub const FRAME_SIZE: usize = 4112;
-    /// Size of the file header
-    pub const HEADER_SIZE: usize = 4096;
+    pub const FRAME_SIZE: usize = size_of::<WalFrameHeader>() + WAL_PAGE_SIZE as usize;
 
     pub fn open(path: impl AsRef<Path>) -> anyhow::Result<Self> {
         let path = path.as_ref().join("wallog");
@@ -150,32 +229,46 @@ impl WalLogger {
             .write(true)
             .read(true)
             .open(path)?;
-        let mut file_end = log_file.metadata()?.len();
-
-        let mut header_buf = [0; 4096];
+        let file_end = log_file.metadata()?.len();
+        let end_id;
+        let current_checksum;
         let header = if file_end == 0 {
             let header = WalLoggerFileHeader {
                 version: 1,
-                start_index: 0,
+                start_frame_id: 0,
+                magic: WAL_MAGIC,
+                page_size: WAL_PAGE_SIZE,
+                start_checksum: 0,
+                db_id: Uuid::new_v4().as_u128(),
             };
-            bincode::serialize_into(Cursor::new(&mut header_buf[..]), &header)?;
+
+            let mut header_buf = BytesMut::new();
+            header.encode(&mut header_buf);
+
+            assert_eq!(header_buf.len(), std::mem::size_of::<WalLoggerFileHeader>());
+
             log_file.write_all(&header_buf)?;
-            file_end = 4096;
+            end_id = 0;
+            current_checksum = AtomicU64::new(0);
             header
         } else {
+            let mut header_buf = BytesMut::zeroed(size_of::<WalLoggerFileHeader>());
             log_file.read_exact(&mut header_buf)?;
-            let header: WalLoggerFileHeader = bincode::deserialize(&header_buf)?;
+            let header = WalLoggerFileHeader::decode(&header_buf)?;
+            end_id = (file_end - size_of::<WalFrameHeader>() as u64) / Self::FRAME_SIZE as u64;
+            current_checksum = AtomicU64::new(Self::compute_checksum(&header, &log_file)?);
             header
         };
 
         Ok(Self {
-            current_offset: Mutex::new(file_end as usize),
-            start_offset: header.start_index as _,
+            next_frame_id: Mutex::new(end_id),
+            start_frame_id: header.start_frame_id,
             log_file,
             hello_response: HelloResponse {
                 generation_id: uuid::Uuid::new_v4().to_string(),
                 first_frame_id_in_generation: file_end,
             },
+            current_checksum,
         })
     }
 
@@ -183,48 +276,126 @@ impl WalLogger {
         &self.hello_response
     }
 
-    fn append(&self, frames: &[WalLogEntry]) {
-        let mut lock = self.current_offset.lock();
+    fn push_page(&self, pages: &[WalPage]) {
+        let mut lock = self.next_frame_id.lock();
         let mut current_offset = *lock;
-        for frame in frames.iter() {
-            #[cfg(any(debug_assertions, test))]
-            if let WalLogEntry::Frame { ref data, .. } = frame {
-                assert_eq!(data.len(), 4096);
-            }
+        let mut buffer = BytesMut::with_capacity(Self::FRAME_SIZE);
+        let mut current_checksum = self.current_checksum.load(Ordering::Relaxed);
+        for page in pages.iter() {
+            debug_assert_eq!(page.data.len(), WAL_PAGE_SIZE as usize);
+            let mut digest = CRC_64_GO_ISO.digest_with_initial(current_checksum);
+            digest.update(&page.data);
+            let checksum = digest.finalize();
+
+            let header = WalFrameHeader {
+                frame_id: current_offset,
+                checksum,
+                page_no: page.page_no,
+                size_after: page.size_after,
+            };
+
+            let frame = WalFrame {
+                header,
+                data: page.data.clone(),
+            };
+
+            frame.encode(&mut buffer);
 
-            let mut buffer = BytesMut::zeroed(Self::FRAME_SIZE);
-            bincode::serialize_into(Cursor::new(buffer.deref_mut()), frame).unwrap();
             self.log_file
-                .write_all_at(&buffer, current_offset as _)
-                // TODO: Handle write error
+                .write_all_at(
+                    &buffer,
+                    self.byte_offset(current_offset)
+                        .expect("attempt to write entry before first entry in the log"),
+                )
                 .unwrap();
-            current_offset += Self::FRAME_SIZE;
+
+            current_offset += 1;
+            current_checksum = checksum;
+
+            buffer.clear();
         }
 
+        self.current_checksum
+            .store(current_checksum, Ordering::Relaxed);
+
         *lock = current_offset;
     }
 
-    /// Returns frame at `index`.
+    /// Returns bytes represening a WalFrame for frame `id`
     ///
     /// If the requested frame is before the first frame in the log, or after the last frame,
     /// Ok(None) is returned.
-    // TODO: implement log compaction
-    // TODO: implement page cache
-    pub fn get_entry(&self, offset: usize) -> anyhow::Result<Option<WalLogEntry>> {
-        if offset < self.start_offset {
+    pub fn frame_bytes(&self, id: FrameId) -> anyhow::Result<Option<Bytes>> {
+        if id < self.start_frame_id {
             return Ok(None);
         }
-        let read_offset = Self::HEADER_SIZE + (offset - self.start_offset) * Self::FRAME_SIZE;
 
-        if read_offset >= *self.current_offset.lock() {
+        if id >= *self.next_frame_id.lock() {
             return Ok(None);
         }
 
         let mut buffer = BytesMut::zeroed(Self::FRAME_SIZE);
-        self.log_file.read_exact_at(&mut buffer, read_offset as _)?;
-        let entry: WalLogEntry = bincode::deserialize(&buffer)?;
+        self.log_file
+            .read_exact_at(&mut buffer, self.byte_offset(id).unwrap())?;
 
-        Ok(Some(entry))
+        Ok(Some(buffer.freeze()))
+    }
+
+    /// Returns the bytes position of the `nth` entry in the log
+    fn absolute_byte_offset(nth: u64) -> u64 {
+        std::mem::size_of::<WalLoggerFileHeader>() as u64 + nth * WalLogger::FRAME_SIZE as u64
+    }
+
+    fn byte_offset(&self, id: FrameId) -> Option<u64> {
+        if id < self.start_frame_id {
+            return None;
+        }
+        Self::absolute_byte_offset(id - self.start_frame_id).into()
+    }
+
+    /// Returns an iterator over the WAL frame headers
+    fn frames_iter(
+        file: &File,
+    ) -> anyhow::Result<impl Iterator<Item = anyhow::Result<WalFrame>> + '_> {
+        fn read_frame_offset(file: &File, offset: u64) -> anyhow::Result<Option<WalFrame>> {
+            let mut buffer = BytesMut::zeroed(WalLogger::FRAME_SIZE);
+            file.read_exact_at(&mut buffer, offset)?;
+            let mut buffer = buffer.freeze();
+            let header_bytes = buffer.split_to(size_of::<WalFrameHeader>());
+            let header = WalFrameHeader::decode(&header_bytes)?;
+            Ok(Some(WalFrame {
+                header,
+                data: buffer,
+            }))
+        }
+
+        let file_len = file.metadata()?.len();
+        let mut current_offset = 0;
+
+        Ok(std::iter::from_fn(move || {
+            let read_offset = Self::absolute_byte_offset(current_offset);
+            if read_offset >= file_len {
+                return None;
+            }
+            current_offset += 1;
+            read_frame_offset(file, read_offset).transpose()
+        }))
+    }
+
+    fn compute_checksum(wal_header: &WalLoggerFileHeader, log_file: &File) -> anyhow::Result<u64> {
+        tracing::debug!("computing WAL log running checksum...");
+        let mut iter = Self::frames_iter(log_file)?;
+        iter.try_fold(wal_header.start_frame_id, |sum, frame| {
+            let frame = frame?;
+            let mut digest = CRC_64_GO_ISO.digest_with_initial(sum);
+            digest.update(&frame.data);
+            let cs = digest.finalize();
+            ensure!(
+                cs == frame.header.checksum,
+                "invalid WAL file: invalid checksum"
+            );
+            Ok(cs)
+        })
     }
 }
 
@@ -237,34 +408,31 @@ mod test {
         let dir = tempfile::tempdir().unwrap();
         let logger = WalLogger::open(dir.path()).unwrap();
 
-        assert_eq!(*logger.current_offset.lock(), WalLogger::HEADER_SIZE);
+        assert_eq!(*logger.next_frame_id.lock(), 0);
 
         let frames = (0..10)
-            .map(|i| WalLogEntry::Frame {
-                data: Bytes::from(vec![i; 4096]),
-                page_no: i as _,
+            .map(|i| WalPage {
+                page_no: i,
+                size_after: 0,
+                data: Bytes::from(vec![i as _; 4096]),
             })
             .collect::<Vec<_>>();
-        logger.append(&frames);
+        logger.push_page(&frames);
 
         for i in 0..10 {
-            let frame = logger.get_entry(i).unwrap().unwrap();
-            let WalLogEntry::Frame{ page_no, data } =  frame else {panic!()};
-            assert_eq!(page_no, i as u32);
-            assert!(data.iter().all(|x| i as u8 == *x));
+            let frame = WalFrame::decode(logger.frame_bytes(i).unwrap().unwrap()).unwrap();
+            assert_eq!(frame.header.page_no, i as u32);
+            assert!(frame.data.iter().all(|x| i as u8 == *x));
         }
 
-        assert_eq!(
-            *logger.current_offset.lock(),
-            WalLogger::HEADER_SIZE + 10 * WalLogger::FRAME_SIZE
-        );
+        assert_eq!(*logger.next_frame_id.lock(), 10);
     }
 
     #[test]
     fn index_out_of_bounds() {
         let dir = tempfile::tempdir().unwrap();
         let logger = WalLogger::open(dir.path()).unwrap();
-        assert!(logger.get_entry(1).unwrap().is_none());
+        assert!(logger.frame_bytes(1).unwrap().is_none());
     }
 
     #[test]
@@ -272,10 +440,11 @@ mod test {
     fn incorrect_frame_size() {
         let dir = tempfile::tempdir().unwrap();
         let logger = WalLogger::open(dir.path()).unwrap();
-        let entry = WalLogEntry::Frame {
+        let entry = WalPage {
             page_no: 0,
+            size_after: 0,
             data: vec![0; 3].into(),
         };
-        logger.append(&[entry]);
+        logger.push_page(&[entry]);
     }
 }
