feat #369: prepare d-engine node image and CLI for Jepsen integration

.dockerignore

@@ -0,0 +1,2 @@
+**/target/
+.git/

.gitignore

@@ -51,3 +51,5 @@ CONTEXT
 safe-split-commits.sh
 CLAUDE.md
 .claude/
+.rules
+.github/copilot-instructions.md

d-engine-server/tests/cas_operations/bank_transfer_invariant_embedded.rs

@@ -0,0 +1,199 @@
+#![cfg(feature = "rocksdb")]
+
+use d_engine_core::ClientApi;
+use d_engine_core::ClientApiError;
+use d_engine_server::RocksDBUnifiedEngine;
+use d_engine_server::api::EmbeddedEngine;
+use serial_test::serial;
+use std::sync::Arc;
+use std::time::Duration;
+use tracing::info;
+use tracing_test::traced_test;
+
+use crate::common::create_node_config;
+use crate::common::get_available_ports;
+use crate::common::node_config;
+
+const BANK_KEY: &[u8] = b"bank";
+const INITIAL_BALANCE: u64 = 1_000;
+const TRANSFERS_PER_WORKER: usize = 50;
+const TRANSFER_AMOUNT: u64 = 1;
+
+// Pack 3 account balances into a single u64 (21 bits each, max ~2M per account).
+fn pack(balances: [u64; 3]) -> [u8; 8] {
+    ((balances[0] << 42) | (balances[1] << 21) | balances[2]).to_be_bytes()
+}
+
+fn unpack(bytes: &[u8]) -> [u64; 3] {
+    let state = u64::from_be_bytes(bytes.try_into().expect("bank value must be 8 bytes"));
+    [
+        (state >> 42) & 0x1FFFFF,
+        (state >> 21) & 0x1FFFFF,
+        state & 0x1FFFFF,
+    ]
+}
+
+/// Concurrent CAS transfers maintain bank total invariant under follower failure.
+///
+/// Setup: 3 accounts, each INITIAL_BALANCE. Total = INITIAL_BALANCE * 3.
+///
+/// Load: 3 concurrent workers do circular transfers A→B, B→C, C→A.
+/// Each transfer is a CAS retry loop on the packed u64 bank key — atomic,
+/// no intermediate state visible to readers.
+///
+/// Fault: a follower node is stopped mid-test. Cluster keeps quorum (2/3),
+/// workers continue without interruption.
+///
+/// Assert: total balance is invariant throughout; no account goes negative.
+#[tokio::test]
+#[traced_test]
+#[serial]
+async fn test_concurrent_transfers_preserve_bank_invariant()
+-> Result<(), Box<dyn std::error::Error>> {
+    let temp_dir = tempfile::tempdir()?;
+    let db_root_dir = temp_dir.path().join("db");
+    let log_dir = temp_dir.path().join("logs");
+
+    let mut port_guard = get_available_ports(3).await;
+    port_guard.release_listeners();
+    let ports = port_guard.as_slice();
+
+    info!("Starting 3-node cluster for bank invariant test");
+    let mut engines = Vec::new();
+
+    for i in 0..3 {
+        let node_id = (i + 1) as u64;
+        let config_str = create_node_config(
+            node_id,
+            ports[i],
+            ports,
+            db_root_dir.to_str().unwrap(),
+            log_dir.to_str().unwrap(),
+        )
+        .await;
+
+        let config = node_config(&config_str);
+        let node_db_root = config.cluster.db_root_dir.join(format!("node{node_id}"));
+        let db_path = node_db_root.join("db");
+        tokio::fs::create_dir_all(&db_path).await?;
+
+        let (storage, state_machine) = RocksDBUnifiedEngine::open(&db_path)?;
+        let config_path = format!("/tmp/d-engine-bank-node{node_id}.toml");
+        tokio::fs::write(&config_path, &config_str).await?;
+
+        let engine = EmbeddedEngine::start_custom(
+            Arc::new(storage),
+            Arc::new(state_machine),
+            Some(&config_path),
+        )
+        .await?;
+        engines.push(engine);
+    }
+
+    let leader_info = engines[0]
+        .wait_ready(Duration::from_secs(15))
+        .await
+        .expect("Failed to elect leader");
+    info!("Leader elected: node {}", leader_info.leader_id);
+
+    let leader_idx = (leader_info.leader_id - 1) as usize;
+    let leader_client = engines[leader_idx].client();
+
+    // Initialize bank: 3 accounts, each INITIAL_BALANCE.
+    let initial_state = pack([INITIAL_BALANCE; 3]);
+    leader_client.put(BANK_KEY, initial_state.as_slice()).await?;
+    let expected_total = INITIAL_BALANCE * 3;
+    info!("Bank initialized: 3 × {INITIAL_BALANCE} = {expected_total}");
+
+    // Spawn 3 workers with circular transfers: worker i moves TRANSFER_AMOUNT
+    // from account i to account (i+1)%3, repeated TRANSFERS_PER_WORKER times.
+    // Circular pattern ensures no account drains to zero (net change = 0).
+    let mut worker_handles = Vec::new();
+    for worker_id in 0..3usize {
+        let client = leader_client.clone();
+        let from = worker_id;
+        let to = (worker_id + 1) % 3;
+
+        worker_handles.push(tokio::spawn(async move {
+            for _ in 0..TRANSFERS_PER_WORKER {
+                // CAS retry loop: read current state → compute new → atomic swap.
+                loop {
+                    let current = match client.get_linearizable(BANK_KEY).await {
+                        Ok(Some(v)) => v,
+                        Ok(None) => panic!("bank key disappeared"),
+                        Err(_) => {
+                            // Transient error (e.g. replication lag); retry.
+                            tokio::time::sleep(Duration::from_millis(10)).await;
+                            continue;
+                        }
+                    };
+
+                    let mut balances = unpack(&current);
+                    // Circular transfers always have sufficient funds,
+                    // but guard defensively.
+                    if balances[from] < TRANSFER_AMOUNT {
+                        tokio::task::yield_now().await;
+                        continue;
+                    }
+                    balances[from] -= TRANSFER_AMOUNT;
+                    balances[to] += TRANSFER_AMOUNT;
+                    let new_bytes = pack(balances);
+
+                    match client
+                        .compare_and_swap(BANK_KEY, Some(current.as_ref()), new_bytes.as_slice())
+                        .await
+                    {
+                        Ok(true) => break, // committed
+                        Ok(false) => {
+                            // Concurrent update; retry without delay.
+                            tokio::task::yield_now().await;
+                        }
+                        Err(_) => {
+                            // Transient error; brief backoff.
+                            tokio::time::sleep(Duration::from_millis(10)).await;
+                        }
+                    }
+                }
+            }
+            Ok::<_, ClientApiError>(())
+        }));
+    }
+
+    // Fault injection: stop a follower node midway through the transfers.
+    let follower_idx = (leader_idx + 1) % 3;
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    info!(
+        "Stopping follower node {} (2/3 quorum maintained)",
+        follower_idx + 1
+    );
+    engines[follower_idx].stop().await?;
+
+    // Wait for all workers to finish.
+    for handle in worker_handles {
+        handle.await??;
+    }
+
+    // Final invariant check.
+    let final_bytes = leader_client
+        .get_linearizable(BANK_KEY)
+        .await?
+        .expect("bank key must exist after all transfers");
+
+    let [a, b, c] = unpack(&final_bytes);
+    let final_total = a + b + c;
+    info!("Final balances: A={a}, B={b}, C={c}, total={final_total}");
+
+    assert_eq!(
+        final_total, expected_total,
+        "Bank total invariant violated: expected {expected_total}, got {final_total} \
+         (A={a}, B={b}, C={c})"
+    );
+
+    // Cleanup.
+    for engine in engines.iter_mut() {
+        let _ = engine.stop().await;
+    }
+
+    info!("Bank invariant test passed");
+    Ok(())
+}

d-engine-server/tests/cas_operations/mod.rs

@@ -81,6 +81,7 @@
 //! }
 //! ```
 
+mod bank_transfer_invariant_embedded;
 mod distributed_lock_embedded;
 mod distributed_lock_standalone;
 mod leader_failover_cas_embedded;

examples/client-usage-standalone/src/main.rs

@@ -22,6 +22,8 @@ enum Commands {
     Get { key: u64 },
     Sget { key: u64 },
     Lget { key: u64 },
+    /// Atomic compare-and-swap. Prints "true" if swapped, "false" if expected didn't match.
+    Cas { key: u64, expected: u64, new_value: u64 },
 }
 
 #[tokio::main]
@@ -42,6 +44,9 @@ async fn main() -> Result<()> {
         Commands::Get { key } => handle_read(&client, key, "e").await,
         Commands::Sget { key } => handle_read(&client, key, "s").await,
         Commands::Lget { key } => handle_read(&client, key, "l").await,
+        Commands::Cas { key, expected, new_value } => {
+            handle_cas(&client, key, expected, new_value).await
+        }
     }
 }
 
@@ -98,6 +103,20 @@ async fn handle_read(
     Ok(())
 }
 
+async fn handle_cas(
+    client: &Client,
+    key: u64,
+    expected: u64,
+    new_value: u64,
+) -> Result<()> {
+    let swapped = client
+        .compare_and_swap(safe_kv(key), Some(safe_kv(expected)), safe_kv(new_value))
+        .await
+        .map_err(|e: ClientApiError| anyhow::anyhow!("CAS error: {e:?}"))?;
+    println!("{}", swapped);
+    Ok(())
+}
+
 pub fn safe_vk<K: AsRef<[u8]>>(bytes: K) -> crate::Result<u64> {
     let bytes = bytes.as_ref();
     let expected_len = 8;

examples/three-nodes-standalone/docker/Dockerfile

@@ -1,11 +1,9 @@
 # ------------------------
 # Builder Stage (Optimized for ARM64)
 # ------------------------
-FROM rust:slim-bullseye AS builder
+FROM rust:slim-bookworm AS builder
 
-WORKDIR /usr/src/demo
-
-# Install build dependencies with cleanup
+# Install build dependencies
 RUN apt-get update && \
     apt-get install -y --no-install-recommends \
     pkg-config \
@@ -14,36 +12,48 @@ RUN apt-get update && \
     build-essential \
     ca-certificates \
     curl \
+    clang \
+    libclang-dev \
+    cmake \
     && rm -rf /var/lib/apt/lists/*
 
-# Cache dependencies first
-COPY Cargo.toml Cargo.lock ./
-RUN mkdir src && \
-    echo "fn main() {}" > src/main.rs && \
-    echo "" > src/lib.rs && \
-    cargo build --release && \
-    rm -rf src target/release/deps/demo*
+# Build context is the d-engine workspace root.
+# Run: docker build -t demo:v0.2.4 -f examples/three-nodes-standalone/docker/Dockerfile .
+WORKDIR /build
 
-# Copy actual source code
+# Disable sccache: .cargo/config.toml sets rustc-wrapper=sccache for local dev,
+# but sccache is not available in CI/Docker.
+ENV RUSTC_WRAPPER=""
+
+# Copy workspace source (.dockerignore excludes **/target/)
 COPY . .
 
-# Final build
-RUN cargo build --release # --locked
+# Build the three-nodes-standalone demo binary.
+# Cache mounts persist the cargo registry and compiled artifacts across builds:
+# - First build: ~10 min (RocksDB compiles from source)
+# - Subsequent builds: ~30s (only changed Rust code recompiles)
+WORKDIR /build/examples/three-nodes-standalone
+RUN --mount=type=cache,target=/usr/local/cargo/registry \
+    --mount=type=cache,target=/build/examples/three-nodes-standalone/target \
+    cargo build --release && \
+    cp target/release/demo /usr/local/bin/demo
 
 # ------------------------
 # Runtime Stage (ARM64 Optimized)
 # ------------------------
 FROM debian:bookworm-slim
 
-# Install runtime dependencies in single layer
+# Install runtime dependencies
 RUN apt-get update && \
     apt-get install -y --no-install-recommends \
     openssh-server \
     ca-certificates \
     curl \
+    wget \
     unzip \
     sudo \
     iptables \
+    iproute2 \
     libc6 \
     tzdata && \
     rm -rf /var/lib/apt/lists/* && \
@@ -65,16 +75,14 @@ RUN curl -LO "https://github.com/grafana/loki/releases/download/v3.0.0/promtail-
     chmod +x /usr/local/bin/promtail && \
     rm promtail-linux-arm64.zip
 
-# Copy built binary
-COPY --from=builder /usr/src/demo/target/release/demo /usr/local/bin/demo
+COPY --from=builder /usr/local/bin/demo /usr/local/bin/demo
 
 # Create directories
 RUN mkdir -p /app/{logs,db,snapshots} /etc/promtail
 
-# Copy configuration
-COPY docker/monitoring/promtail/config.yml /etc/promtail/config.yml
+# Promtail config path relative to workspace root
+COPY examples/three-nodes-standalone/docker/monitoring/promtail/config.yml /etc/promtail/config.yml
 
 WORKDIR /app
 
-# Start services
-CMD ["sh", "-c", "/usr/sbin/sshd -D & CONFIG_PATH=$CONFIG_PATH LOG_DIR=$LOG_DIR METRICS_PORT=$METRICS_PORT RUST_LOG=demo=$LOG_LEVEL,d_engine=$LOG_LEVEL,hyper=warn,sled=warn demo  & promtail --config.file=/etc/promtail/config.yml > /app/logs/promtail.log 2>&1"]
+CMD ["sh", "-c", "/usr/sbin/sshd -D & CONFIG_PATH=$CONFIG_PATH LOG_DIR=$LOG_DIR METRICS_PORT=$METRICS_PORT RUST_LOG=demo=$LOG_LEVEL,d_engine=$LOG_LEVEL,hyper=warn,sled=warn demo & promtail --config.file=/etc/promtail/config.yml > /app/logs/promtail.log 2>&1"]