Fix idle maintenance CPU and release v0.3.3

.github/workflows/ci.yml

@@ -0,0 +1,57 @@
+name: ci
+
+on:
+  pull_request:
+  workflow_call:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  tests:
+    runs-on: ubuntu-24.04
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          path: marekvs
+      - name: resolve tested ondaDB revision
+        id: ondadb
+        working-directory: marekvs
+        run: |
+          python3 - <<'PY' >> "$GITHUB_OUTPUT"
+          import tomllib
+          with open("Cargo.lock", "rb") as f:
+              lock = tomllib.load(f)
+          package = next(p for p in lock["package"] if p["name"] == "ondadb")
+          revision = package["source"].rsplit("#", 1)[1]
+          assert len(revision) == 40 and all(c in "0123456789abcdef" for c in revision)
+          print(f"ref={revision}")
+          PY
+      - uses: actions/checkout@v4
+        with:
+          repository: ${{ github.repository_owner }}/ondadb
+          ref: ${{ steps.ondadb.outputs.ref }}
+          path: ondadb
+      - uses: dtolnay/rust-toolchain@1.97.1
+        with:
+          components: rustfmt, clippy
+      - uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: |
+            marekvs
+            ondadb
+      - name: marekvs formatting, lint and regression suites
+        working-directory: marekvs
+        run: |
+          cargo fmt --all --check
+          cargo clippy --locked --workspace --all-targets -- -D warnings
+          cargo test --locked --workspace
+          python3 tests/chaos/grudge.py --test
+      - name: ondaDB regression suites (including range mask reuse)
+        working-directory: ondadb
+        run: |
+          cargo clippy --locked --all-targets -- -D warnings
+          cargo test --locked
+          cargo clippy --locked --all-targets --features unsafe-fastpath -- -D warnings
+          cargo test --locked --features unsafe-fastpath

.github/workflows/docker.yml

@@ -21,16 +21,18 @@ on:
     branches: [main]
     tags: ["v*"]
   workflow_dispatch:
-    inputs:
-      ondadb-ref:
-        description: ondadb ref to build against
-        default: main
 
 env:
   IMAGE: ghcr.io/${{ github.repository }}
 
 jobs:
+  quality:
+    uses: ./.github/workflows/ci.yml
+    permissions:
+      contents: read
+
   build:
+    needs: quality
     strategy:
       fail-fast: true
       matrix:
@@ -49,11 +51,25 @@ jobs:
         with:
           path: marekvs
 
+      - name: resolve pinned ondaDB revision
+        id: ondadb
+        working-directory: marekvs
+        run: |
+          python3 - <<'PYTHON' >> "$GITHUB_OUTPUT"
+          import tomllib
+          with open("Cargo.lock", "rb") as f:
+              lock = tomllib.load(f)
+          package = next(p for p in lock["package"] if p["name"] == "ondadb")
+          revision = package["source"].rsplit("#", 1)[1]
+          assert len(revision) == 40 and all(c in "0123456789abcdef" for c in revision)
+          print(f"ref={revision}")
+          PYTHON
+
       - name: checkout ondadb (path dependency)
         uses: actions/checkout@v4
         with:
           repository: ${{ github.repository_owner }}/ondadb
-          ref: ${{ inputs.ondadb-ref || 'main' }}
+          ref: ${{ steps.ondadb.outputs.ref }}
           path: ondadb
 
       - name: point cargo at the checked-out ondadb

Cargo.toml

@@ -13,7 +13,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.3.2"
+version = "0.3.3"
 edition = "2021"
 # 1.89: ondaDB ≥0.7 uses std::fs::File advisory locking for its DB LOCK file.
 rust-version = "1.89"
@@ -22,7 +22,7 @@ rust-version = "1.89"
 [workspace.dependencies]
 # Canonical source; a sibling checkout at ../ondadb overrides it via the
 # [patch] in .cargo/config.toml (auto-generated by `just` — see _cargo-config).
-ondadb = { git = "https://github.com/yannick/ondadb.git" }
+ondadb = { git = "https://github.com/yannick/ondadb.git", rev = "0f4ebc67434551c9d3d4cba899829234760f4215" }
 tokio = { version = "1", features = ["full"] }
 bytes = "1"
 postcard = { version = "1", features = ["alloc"] }

crates/marekvs-cluster/src/lib.rs

@@ -372,6 +372,13 @@ impl Cluster {
         self.view.read().clone()
     }
 
+    /// Execute a short synchronous operation while placement cannot change.
+    /// Used for the final cold-purge eligibility check and local range commit.
+    pub fn with_view<T>(&self, f: impl FnOnce(&View) -> T) -> T {
+        let view = self.view.read();
+        f(&view)
+    }
+
     /// Subscribe to view changes (value = epoch).
     pub fn watch(&self) -> watch::Receiver<u64> {
         self.view_tx.subscribe()

crates/marekvs-engine/examples/idle_maintenance.rs

@@ -0,0 +1,105 @@
+//! Disposable, release-mode idle CPU fixture; see tests/idle_maintenance/README.md.
+use marekvs_core::{
+    envelope::{Envelope, RecordType},
+    ikey,
+};
+use marekvs_engine::{
+    store::{Store, StoreConfig},
+    Engine,
+};
+use std::time::{Duration, Instant};
+
+fn cpu_seconds() -> f64 {
+    let mut usage = std::mem::MaybeUninit::<libc::rusage>::uninit();
+    // SAFETY: getrusage initializes the correctly sized output on success.
+    assert_eq!(
+        unsafe { libc::getrusage(libc::RUSAGE_SELF, usage.as_mut_ptr()) },
+        0
+    );
+    let usage = unsafe { usage.assume_init() };
+    usage.ru_utime.tv_sec as f64
+        + usage.ru_stime.tv_sec as f64
+        + (usage.ru_utime.tv_usec + usage.ru_stime.tv_usec) as f64 / 1_000_000.0
+}
+
+fn main() {
+    let args: Vec<usize> = std::env::args()
+        .skip(1)
+        .map(|s| s.parse().unwrap())
+        .collect();
+    assert_eq!(
+        args.len(),
+        6,
+        "usage: idle_maintenance SHARDS RANGES DISTINCT_RANGES KEYS WARM_SECONDS SAMPLE_SECONDS"
+    );
+    let (shards, ranges, distinct, keys, warm, seconds) =
+        (args[0], args[1], args[2], args[3], args[4], args[5]);
+    assert!(
+        shards > 0
+            && shards <= 4096
+            && distinct <= 4096
+            && (ranges == 0 || distinct > 0)
+            && seconds > 0
+    );
+    let dir = tempfile::tempdir().unwrap();
+    let store = Store::open(&StoreConfig {
+        data_dir: dir.path().to_string_lossy().into_owned(),
+        shard_threads: shards,
+        ..StoreConfig::default()
+    })
+    .unwrap();
+    // Seed the live memtables without closing/flushing them. All writes use
+    // their owning shard, exactly like production commands and repair.
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap()
+        .block_on(async {
+            for shard in 0..shards {
+                store
+                    .run(shard as u16, move |ctx| {
+                        for i in 0..ranges {
+                            let pid = (i % distinct) as u16;
+                            if pid as usize % shards != shard {
+                                continue;
+                            }
+                            marekvs_engine::store::delete_partition_range(ctx, pid).unwrap();
+                        }
+                        for i in 0..keys {
+                            let key = format!("idle-key-{i}");
+                            if marekvs_core::pid_of(key.as_bytes()) as usize % shards != shard {
+                                continue;
+                            }
+                            let env = Envelope {
+                                flags: RecordType::String as u8,
+                                hlc: 1 << 16,
+                                origin: 1,
+                                ttl_deadline_ms: 0,
+                            };
+                            marekvs_engine::store::put_raw(
+                                ctx,
+                                &ikey::string_key(key.as_bytes()),
+                                &env.encode_with(b"value"),
+                            );
+                        }
+                    })
+                    .await;
+            }
+        });
+    let engine = Engine::new(store);
+    std::thread::sleep(Duration::from_secs(warm as u64));
+    println!(
+        "BEGIN_METRICS\n{}",
+        engine.metrics.render(engine.started_at_ms, 0)
+    );
+    let wall = Instant::now();
+    let cpu = cpu_seconds();
+    std::thread::sleep(Duration::from_secs(seconds as u64));
+    let cpu = cpu_seconds() - cpu;
+    let wall = wall.elapsed().as_secs_f64();
+    println!(
+        "END_METRICS\n{}",
+        engine.metrics.render(engine.started_at_ms, 0)
+    );
+    println!("RESULT shards={shards} ranges={ranges} distinct={distinct} keys={keys} warm_seconds={warm} sample_seconds={wall:.3} cpu_seconds={cpu:.6} cpu_percent={:.3}", cpu / wall * 100.0);
+}

crates/marekvs-engine/src/cmd/generic.rs

@@ -124,10 +124,11 @@ pub async fn type_cmd(engine: &Arc<Engine>, args: &[Vec<u8>]) -> Reply {
 }
 
 /// The envelope currently carrying this key's TTL (string, list, or head).
-fn ttl_envelope(ctx: &ShardCtx, key: &[u8]) -> Option<Envelope> {
+pub(crate) fn ttl_envelope(ctx: &ShardCtx, key: &[u8]) -> Option<Envelope> {
     // Lists carry their TTL on the collection head now (ctype 5 → `_` arm).
     match key_type(ctx, key)? {
         b's' => read_lww(ctx, &ikey::string_key(key), 0).map(|(e, _)| e),
+        b'l' => read_lww(ctx, &ikey::list_key(key), 0).map(|(e, _)| e),
         _ => get_head(ctx, key).map(|(e, _, _)| e),
     }
 }