fix(wal): isolate raft-engine entry caches by engine

.gitattributes

@@ -0,0 +1,5 @@
+# Preserve the published third-party source, including its existing whitespace.
+vendor/raft-engine/** linguist-vendored -whitespace
+# Keep the provenance and complete local delta prominent in review.
+vendor/raft-engine/LANCE_CONTEXT_PATCH.md -linguist-vendored
+vendor/raft-engine/read-cache-identity.patch -linguist-vendored

.github/workflows/rust-test.yml

@@ -6,6 +6,7 @@ on:
   pull_request:
     paths:
       - crates/**
+      - vendor/raft-engine/**
       - .github/workflows/rust-test.yml
       - Cargo.toml
       - Cargo.lock
@@ -63,6 +64,7 @@ jobs:
         run: |
           cargo run -p lance-context-wal --features lab --bin context-wal-lab -- codec --payload-bytes 1024 --rows 8 --iterations 2
           cargo test -p lance-context-wal --features lab --bin context-wal-lab
+          cargo test -p lance-context-wal --features lab --test engine_cache
           cargo run -p lance-context-wal --features lab --bin context-wal-lab -- disk --directory "$RUNNER_TEMP/wal-disk-smoke" --payload-bytes 1024 --batch 4 --writers 2 --groups-per-writer 2 --rounds 8
       - name: WAL placement, admission, record, contract and IPC decoding tests
         run: |

.github/workflows/style.yml

@@ -6,6 +6,8 @@ on:
   pull_request:
     paths:
       - crates/**
+      - vendor/raft-engine/**
+      - .typos.toml
       - Cargo.toml
       - Cargo.lock
       - .github/workflows/style.yml

.typos.toml

@@ -0,0 +1,8 @@
+[files]
+# Preserve the vendored upstream source; its two-file correction is documented
+# separately instead of mixing unrelated upstream spelling changes into it.
+extend-exclude = ["vendor/raft-engine/**"]
+
+[default.extend-words]
+# Generalized advantage estimation in the rollout schema.
+GAE = "GAE"

Cargo.toml

@@ -12,6 +12,7 @@ members = [
     "python",
 ]
 resolver = "2"
+exclude = ["vendor/raft-engine"]
 
 [workspace.package]
 edition = "2021"

crates/lance-context-wal/Cargo.toml

@@ -15,7 +15,7 @@ record = ["dep:bytes", "bytes/serde", "dep:serde", "dep:sha2", "dep:uuid"]
 placement = ["dep:anyhow", "dep:serde", "dep:serde_json", "dep:sha2", "dep:uuid"]
 admission = ["dep:anyhow", "dep:serde"]
 ipc = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "dep:arrow-schema", "dep:bytes"]
-lab = ["ipc", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "dep:raft-proto", "dep:serde_json"]
+lab = ["ipc", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "raft-engine/internals", "dep:raft-proto", "dep:serde_json"]
 
 [dependencies]
 tokio = { version = "1", features = ["io-util"], optional = true }
@@ -28,7 +28,7 @@ bytes = { version = "1", optional = true }
 clap = { version = "4", features = ["derive"], optional = true }
 hdrhistogram = { version = "7", optional = true }
 libc = { version = "0.2", optional = true }
-raft-engine = { version = "=0.4.2", default-features = false, optional = true }
+raft-engine = { path = "../../vendor/raft-engine", version = "=0.4.2", default-features = false, optional = true }
 raft-proto = { version = "=0.7.0", optional = true }
 serde_json = { version = "1", optional = true }
 serde = { version = "1", features = ["derive"], optional = true }

crates/lance-context-wal/tests/engine_cache.rs

@@ -0,0 +1,135 @@
+#![cfg(feature = "lab")]
+
+//! Regression expectations for the actual pinned dependency and real files.
+//! Run unchanged against the original and any candidate dependency correction.
+use anyhow::{ensure, Result};
+use raft_engine::{internals::LogQueue, Config, Engine, LogBatch, MessageExt, ReadableSize};
+use raft_proto::eraftpb::Entry;
+use std::path::Path;
+use std::sync::Arc;
+
+struct Index;
+impl MessageExt for Index {
+    type Entry = Entry;
+    fn index(entry: &Entry) -> u64 {
+        entry.index
+    }
+}
+
+fn config(path: &Path, file_size: u64) -> Config {
+    Config {
+        dir: path.to_string_lossy().into_owned(),
+        target_file_size: ReadableSize(file_size),
+        purge_threshold: ReadableSize(2 * file_size),
+        purge_rewrite_threshold: Some(ReadableSize(file_size)),
+        enable_log_recycle: false,
+        batch_compression_threshold: ReadableSize(0),
+        ..Config::default()
+    }
+}
+
+fn write(engine: &Engine, through: u64, byte: u8) -> Result<()> {
+    for index in 1..=through {
+        let mut batch = LogBatch::default();
+        batch.add_entries::<Index>(
+            1,
+            &[Entry {
+                index,
+                term: 1,
+                data: vec![byte; 4096].into(),
+                ..Entry::default()
+            }],
+        )?;
+        engine.write(&mut batch, true)?;
+    }
+    Ok(())
+}
+
+fn verify(entry: Entry, byte: u8) -> Result<()> {
+    ensure!(entry.data.len() == 4096, "unexpected payload length");
+    ensure!(
+        entry.data.iter().all(|value| *value == byte),
+        "wrong engine payload at index {}: expected byte {}, actual first byte {:?}",
+        entry.index,
+        byte,
+        entry.data.first()
+    );
+    Ok(())
+}
+
+fn fresh_thread_verify(engine: Arc<Engine>, through: u64, byte: u8) -> Result<()> {
+    std::thread::spawn(move || {
+        for index in 1..=through {
+            verify(engine.get_entry::<Index>(1, index)?.unwrap(), byte)?;
+        }
+        Ok(())
+    })
+    .join()
+    .expect("fresh reader panicked")
+}
+
+#[test]
+fn scalar_reads_distinguish_two_live_engines() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let a = Arc::new(Engine::open(config(&root.path().join("a"), 64 * 1024))?);
+    let b = Arc::new(Engine::open(config(&root.path().join("b"), 64 * 1024))?);
+    write(&a, 1, 41)?;
+    write(&b, 1, 42)?;
+    fresh_thread_verify(a.clone(), 1, 41)?;
+    fresh_thread_verify(b.clone(), 1, 42)?;
+    verify(a.get_entry::<Index>(1, 1)?.unwrap(), 41)?;
+    verify(b.get_entry::<Index>(1, 1)?.unwrap(), 42)
+}
+
+#[test]
+fn batched_reads_distinguish_two_live_engines() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let a = Engine::open(config(&root.path().join("a"), 64 * 1024))?;
+    let b = Engine::open(config(&root.path().join("b"), 64 * 1024))?;
+    write(&a, 1, 51)?;
+    write(&b, 1, 52)?;
+    let mut entries = Vec::new();
+    a.fetch_entries_to::<Index>(1, 1, 2, None, &mut entries)?;
+    verify(entries.pop().unwrap(), 51)?;
+    b.fetch_entries_to::<Index>(1, 1, 2, None, &mut entries)?;
+    verify(entries.pop().unwrap(), 52)
+}
+
+#[test]
+fn closing_an_engine_does_not_leave_a_cache_alias_for_the_next() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let a = Engine::open(config(&root.path().join("a"), 64 * 1024))?;
+    write(&a, 1, 61)?;
+    verify(a.get_entry::<Index>(1, 1)?.unwrap(), 61)?;
+    drop(a);
+    let b = Engine::open(config(&root.path().join("b"), 64 * 1024))?;
+    write(&b, 1, 62)?;
+    verify(b.get_entry::<Index>(1, 1)?.unwrap(), 62)
+}
+
+#[test]
+fn rewrite_never_persists_another_engines_cached_payload() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let a = Arc::new(Engine::open(config(&root.path().join("a"), 4096))?);
+    let config_b = config(&root.path().join("b"), 4096);
+    let b = Arc::new(Engine::open(config_b.clone())?);
+    write(&a, 8, 71)?;
+    write(&b, 8, 72)?;
+    fresh_thread_verify(b.clone(), 8, 72)?;
+    let before = b.file_span(LogQueue::Append);
+    ensure!(
+        before.1 > before.0 + 2,
+        "fixture needs several closed segments"
+    );
+    verify(a.get_entry::<Index>(1, 1)?.unwrap(), 71)?;
+    b.purge_expired_files()?;
+    ensure!(
+        b.file_span(LogQueue::Append).0 > before.0,
+        "test did not exercise physical rewrite/reclamation"
+    );
+    drop(b);
+    let reopened = Arc::new(Engine::open(config_b)?);
+    // Reading on a new OS thread distinguishes persisted corruption from a
+    // second cache hit in the test's original thread.
+    fresh_thread_verify(reopened, 8, 72)
+}

vendor/raft-engine/.cargo_vcs_info.json

@@ -0,0 +1,6 @@
+{
+  "git": {
+    "sha1": "084c5ca0e4f8291ab67afbfe8b9df02747838c86"
+  },
+  "path_in_vcs": ""
+}

vendor/raft-engine/.github/dependabot.yml

@@ -0,0 +1,6 @@
+version: 2
+updates:
+  - package-ecosystem: "cargo"
+    directory: "/"
+    schedule:
+      interval: "daily"

vendor/raft-engine/.github/workflows/rust.yml

@@ -0,0 +1,114 @@
+name: Rust
+
+on: [push, pull_request]
+
+env:
+  CARGO_TERM_COLOR: always
+
+jobs:
+  nightly:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ ubuntu-latest, macos-latest ]
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}
+      - name: Install nightly toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: nightly-2023-12-31
+          override: true
+          components: rustfmt, clippy, rust-src
+      - uses: Swatinem/rust-cache@v1
+        with:
+          sharedKey: ${{ matrix.os }}
+      - name: Cache dependencies
+        if: ${{ matrix.os == 'ubuntu-latest' }}
+        run: if [[ ! -e ~/.cargo/bin/grcov ]]; then cargo install grcov; fi
+      - name: Format
+        run: |
+          make format
+          git diff --exit-code
+      - name: Clippy
+        run: make clippy
+        env:
+          EXTRA_CARGO_ARGS: '--fix'
+      - name: Run tests
+        run: make test
+        env:
+          RUST_BACKTRACE: 1
+          EXTRA_CARGO_ARGS: '--verbose'
+      - name: Run asan tests
+        if: ${{ matrix.os == 'ubuntu-latest' }}
+        run: make test
+        env:
+          RUST_BACKTRACE: 1
+          RUSTFLAGS: '-Zsanitizer=address'
+          RUSTDOCFLAGS: '-Zsanitizer=address'
+          EXTRA_CARGO_ARGS: '--verbose -Zbuild-std --target x86_64-unknown-linux-gnu'
+  stable:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ ubuntu-latest ]
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}
+      - name: Install stable toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: 1.75.0
+          override: true
+          components: rustfmt, clippy, rust-src
+      - uses: Swatinem/rust-cache@v1
+        with:
+          sharedKey: ${{ matrix.os }}-stable
+      - name: Clippy
+        run: make clippy
+        env:
+          WITH_STABLE_TOOLCHAIN: 'force'
+      - name: Run tests
+        run: make test
+        env:
+          RUST_BACKTRACE: 1
+          EXTRA_CARGO_ARGS: '--verbose'
+          WITH_STABLE_TOOLCHAIN: 'force'
+  coverage:
+    runs-on: ubuntu-latest
+    needs: nightly
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}
+      - name: Install nightly toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: nightly-2023-12-31
+          override: true
+          components: llvm-tools-preview
+      - uses: Swatinem/rust-cache@v1
+        with:
+          sharedKey: ubuntu-latest
+      - name: Install grcov
+        run: if [[ ! -e ~/.cargo/bin/grcov ]]; then cargo install --locked grcov; fi
+      - name: Run tests
+        run: |
+          make test_matrix
+        env:
+          RUSTFLAGS: '-Cinstrument-coverage'
+          LLVM_PROFILE_FILE: '%p-%m.profraw'
+          EXTRA_CARGO_ARGS: '--verbose'
+      - name: Run grcov
+        run: grcov `find . \( -name "*.profraw" \) -print` --binary-path target/debug/deps/ -s . -t lcov --branch --ignore-not-existing --ignore '../**' --ignore '/*' -o coverage.lcov
+      - name: Upload
+        uses: codecov/codecov-action@v3
+        env:
+          CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+        with:
+          file: coverage.lcov

vendor/raft-engine/.gitignore

@@ -0,0 +1,13 @@
+## File system
+.DS_Store
+Cargo.lock
+
+## Build
+/target
+
+## Temporary files
+*~
+\#*
+\#*\#
+.#*
+.idea