feat: add optional serde codec for log entries

Cargo.toml

@@ -20,6 +20,11 @@ path = "examples/append_compact_purge.rs"
 name = "raft-engine-fork"
 path = "examples/fork.rs"
 
+[[example]]
+name = "json-codec"
+path = "examples/append_with_json_codec.rs"
+required-features = ["serde-json"]
+
 [[test]]
 name = "failpoints"
 path = "tests/failpoints/mod.rs"
@@ -59,6 +64,8 @@ rayon = "1.5"
 rhai = { version = "1.7", features = ["sync"], optional = true }
 scopeguard = "1.1"
 serde = { version = "1.0", features = ["derive"] }
+serde_json = { version = "1.0", optional = true }
+bincode = { version = "1.3", optional = true }
 serde_repr = "0.1"
 strum = { version = "0.26.2", features = ["derive"] }
 thiserror = "1.0"
@@ -84,10 +91,13 @@ internals = []
 nightly = ["prometheus/nightly"]
 failpoints = ["fail/failpoints"]
 scripting = ["rhai"]
+serde-bincode = ["dep:bincode"]
+serde-json = ["dep:serde_json"]
 swap = ["nightly", "memmap2"]
 std_fs = []
 
 nightly_group = ["nightly", "swap"]
+serde_group = ["serde-bincode", "serde-json"]
 
 [patch.crates-io]
 raft-proto = { git = "https://github.com/tikv/raft-rs", branch = "master" }

Makefile

@@ -45,18 +45,18 @@ CLIPPY_WHITELIST += -A clippy::bool_assert_comparison
 ## Run clippy.
 clippy:
 ifdef WITH_NIGHTLY_FEATURES
-	cargo ${TOOLCHAIN_ARGS} clippy --all --features nightly_group,failpoints --all-targets -- -D clippy::all ${CLIPPY_WHITELIST}
+	cargo ${TOOLCHAIN_ARGS} clippy --all --features nightly_group,serde_group,failpoints --all-targets -- -D clippy::all ${CLIPPY_WHITELIST}
 else
-	cargo ${TOOLCHAIN_ARGS} clippy --all --features failpoints --all-targets -- -D clippy::all ${CLIPPY_WHITELIST}
+	cargo ${TOOLCHAIN_ARGS} clippy --all --features serde_group,failpoints --all-targets -- -D clippy::all ${CLIPPY_WHITELIST}
 endif
 
 ## Run tests.
 test:
 ifdef WITH_NIGHTLY_FEATURES
-	cargo ${TOOLCHAIN_ARGS} test --all --features nightly_group ${EXTRA_CARGO_ARGS} -- --nocapture
+	cargo ${TOOLCHAIN_ARGS} test --all --features nightly_group,serde_group ${EXTRA_CARGO_ARGS} -- --nocapture
 	cargo ${TOOLCHAIN_ARGS} test --test failpoints --features nightly_group,failpoints ${EXTRA_CARGO_ARGS} -- --test-threads 1 --nocapture
 else
-	cargo ${TOOLCHAIN_ARGS} test --all ${EXTRA_CARGO_ARGS} -- --nocapture
+	cargo ${TOOLCHAIN_ARGS} test --all --features serde_group ${EXTRA_CARGO_ARGS} -- --nocapture
 	cargo ${TOOLCHAIN_ARGS} test --test failpoints --features failpoints ${EXTRA_CARGO_ARGS} -- --test-threads 1 --nocapture
 endif
 
@@ -66,9 +66,9 @@ test_matrix:
 	$(error Must run test matrix with nightly features. Please reset WITH_STABLE_TOOLCHAIN.)
 else
 test_matrix: test
-	cargo ${TOOLCHAIN_ARGS} test --all ${EXTRA_CARGO_ARGS} -- --nocapture
+	cargo ${TOOLCHAIN_ARGS} test --all --features serde_group ${EXTRA_CARGO_ARGS} -- --nocapture
 	cargo ${TOOLCHAIN_ARGS} test --test failpoints --features failpoints ${EXTRA_CARGO_ARGS} -- --test-threads 1 --nocapture
-	cargo ${TOOLCHAIN_ARGS} test --all --features nightly_group,std_fs ${EXTRA_CARGO_ARGS} -- --nocapture
+	cargo ${TOOLCHAIN_ARGS} test --all --features nightly_group,std_fs,serde_group ${EXTRA_CARGO_ARGS} -- --nocapture
 	cargo ${TOOLCHAIN_ARGS} test --test failpoints --features nightly_group,std_fs,failpoints ${EXTRA_CARGO_ARGS} -- --test-threads 1 --nocapture
 endif
 

examples/append_with_json_codec.rs

@@ -0,0 +1,147 @@
+// Copyright (c) 2017-present, PingCAP, Inc. Licensed under Apache-2.0.
+
+use raft_engine::{Config, Engine, JsonCodec, LogBatch, MessageExt, ReadableSize};
+use rand::thread_rng;
+use rand_distr::{Distribution, Normal};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+struct JsonEntry {
+    index: u64,
+    term: u64,
+    payload: String,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
+struct RegionState {
+    last_index: u64,
+    last_round: u64,
+}
+
+struct JsonEntryExt;
+
+impl MessageExt<JsonCodec> for JsonEntryExt {
+    type Entry = JsonEntry;
+
+    fn index(e: &Self::Entry) -> u64 {
+        e.index
+    }
+}
+
+const DATA_DIR: &str = "append_with_json_codec";
+const ROUNDS: u64 = 256;
+const WRITES_PER_ROUND: u64 = 512;
+const COMPACT_OFFSET: u64 = 32;
+
+fn main() {
+    env_logger::init();
+
+    let config = Config {
+        dir: DATA_DIR.to_owned(),
+        // Small thresholds so that a short run actually rotates files, purges
+        // and exercises the rewrite path.
+        target_file_size: ReadableSize::kb(128),
+        purge_threshold: ReadableSize::mb(1),
+        batch_compression_threshold: ReadableSize::kb(0),
+        ..Default::default()
+    };
+    let engine = Engine::open(config).expect("Open raft engine");
+    let recovered = engine.raft_groups();
+    if recovered.is_empty() {
+        println!("[EXAMPLE] starting from an empty {DATA_DIR}/");
+    } else {
+        println!(
+            "[EXAMPLE] recovered {} raft groups written by a previous run",
+            recovered.len()
+        );
+    }
+
+    let mut rand_regions = Normal::new(8.0, 4.0)
+        .unwrap()
+        .sample_iter(thread_rng())
+        .map(|x: f64| (x as u64) % 16);
+    let mut rand_compacts = Normal::new(COMPACT_OFFSET as f64, 16.0)
+        .unwrap()
+        .sample_iter(thread_rng())
+        .map(|x: f64| x as u64);
+
+    let mut batch = LogBatch::with_capacity(256);
+    let payload = "x".repeat(1024);
+
+    for round in 1..=ROUNDS {
+        for _ in 0..WRITES_PER_ROUND {
+            let region = rand_regions.next().unwrap();
+            let mut state = engine
+                .get_value::<RegionState, JsonCodec>(region, b"state")
+                .unwrap()
+                .unwrap_or_default();
+
+            state.last_index += 1; // manually update the state
+            state.last_round = round;
+
+            let entry = JsonEntry {
+                index: state.last_index,
+                term: round,
+                payload: payload.clone(),
+            };
+            batch
+                .add_entries_with::<JsonEntryExt, JsonCodec>(region, &[entry])
+                .unwrap();
+            batch
+                .put_value::<JsonCodec, _>(region, b"state".to_vec(), &state)
+                .unwrap();
+            engine.write(&mut batch, false).unwrap();
+
+            if state.last_index % COMPACT_OFFSET == 0 {
+                let rand_compact_offset = rand_compacts.next().unwrap();
+                if state.last_index > rand_compact_offset {
+                    let compact_to = state.last_index - rand_compact_offset;
+                    engine.compact_to(region, compact_to);
+                }
+            }
+        }
+
+        for region in engine.purge_expired_files().unwrap() {
+            let state = engine
+                .get_value::<RegionState, JsonCodec>(region, b"state")
+                .unwrap()
+                .unwrap();
+            let compact_to = state.last_index.saturating_sub(7);
+            engine.compact_to(region, compact_to);
+            println!("[EXAMPLE] round {round}: force compact {region} to {compact_to}");
+        }
+    }
+    engine.sync().unwrap();
+
+    // Read everything back and check it survived the whole cycle.
+    let mut regions = engine.raft_groups();
+    regions.sort_unstable();
+    for &region in &regions {
+        let (first, last) = match (engine.first_index(region), engine.last_index(region)) {
+            (Some(f), Some(l)) => (f, l),
+            _ => continue,
+        };
+        let mut entries = Vec::new();
+        engine
+            .fetch_entries_to_with::<JsonEntryExt, JsonCodec>(
+                region,
+                first,
+                last + 1,
+                None,
+                &mut entries,
+            )
+            .unwrap();
+        assert_eq!(entries.len() as u64, last - first + 1);
+        for (offset, e) in entries.iter().enumerate() {
+            assert_eq!(e.index, first + offset as u64);
+            assert_eq!(e.payload, payload);
+        }
+        let state = engine
+            .get_value::<RegionState, JsonCodec>(region, b"state")
+            .unwrap()
+            .unwrap();
+        assert_eq!(state.last_index, last);
+        println!("[EXAMPLE] region {region}: entries [{first}, {last}] verified");
+    }
+    println!("[EXAMPLE] done, data left in {DATA_DIR}/");
+}