From 4dacddb3879477b4eb888c24cb063e572fa4a02a Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Fri, 8 May 2026 14:38:45 +0800 Subject: [PATCH 1/6] feat #369: add cas subcommand to client-usage-standalone-demo CLI --- .../bank_transfer_invariant_embedded.rs | 196 ++++++++++++++++++ d-engine-server/tests/cas_operations/mod.rs | 1 + examples/client-usage-standalone/src/main.rs | 19 ++ 3 files changed, 216 insertions(+) create mode 100644 d-engine-server/tests/cas_operations/bank_transfer_invariant_embedded.rs diff --git a/d-engine-server/tests/cas_operations/bank_transfer_invariant_embedded.rs b/d-engine-server/tests/cas_operations/bank_transfer_invariant_embedded.rs new file mode 100644 index 00000000..acfaac7b --- /dev/null +++ b/d-engine-server/tests/cas_operations/bank_transfer_invariant_embedded.rs @@ -0,0 +1,196 @@ +#![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> { + 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(¤t); + // 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(()) +} diff --git a/d-engine-server/tests/cas_operations/mod.rs b/d-engine-server/tests/cas_operations/mod.rs index c0fe6fec..be4946e3 100644 --- a/d-engine-server/tests/cas_operations/mod.rs +++ b/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; diff --git a/examples/client-usage-standalone/src/main.rs b/examples/client-usage-standalone/src/main.rs index 43bde9dc..9dbfe162 100644 --- a/examples/client-usage-standalone/src/main.rs +++ b/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>(bytes: K) -> crate::Result { let bytes = bytes.as_ref(); let expected_len = 8; From ac7591e39f5c97009fd3405ad53241a49b9fe3df Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Fri, 8 May 2026 14:38:50 +0800 Subject: [PATCH 2/6] feat #369: add bank/set Jepsen workloads and migrate CLI to client-usage-standalone-demo --- .dockerignore | 2 + .../three-nodes-standalone/docker/README.md | 4 +- .../docker/jepsen/Dockerfile | 22 ++-- .../docker/jepsen/Makefile | 6 +- .../docker/jepsen/src/jepsen/d_engine.clj | 95 ++++++++------ .../jepsen/src/jepsen/d_engine/bank.clj | 100 +++++++++++++++ .../docker/jepsen/src/jepsen/d_engine/set.clj | 69 +++++++++++ .../jepsen/vendor/dengine_ctl/Cargo.toml | 15 --- .../jepsen/vendor/dengine_ctl/README.md | 31 ----- .../jepsen/vendor/dengine_ctl/src/main.rs | 116 ------------------ 10 files changed, 243 insertions(+), 217 deletions(-) create mode 100644 .dockerignore create mode 100644 examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/bank.clj create mode 100644 examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/set.clj delete mode 100644 examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/Cargo.toml delete mode 100644 examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/README.md delete mode 100644 examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/src/main.rs diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..6db06ce3 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,2 @@ +**/target/ +.git/ diff --git a/examples/three-nodes-standalone/docker/README.md b/examples/three-nodes-standalone/docker/README.md index 17e5c031..9dac1076 100644 --- a/examples/three-nodes-standalone/docker/README.md +++ b/examples/three-nodes-standalone/docker/README.md @@ -21,8 +21,8 @@ docker build -t demo:1.3 -f docker/Dockerfile . # Build monitoring component docker build -t prometheus:1.0 -f docker/monitoring/prometheus/Dockerfile . -# Build Jepsen test container -docker build -t jepsen:1.0 -f docker/jepsen/Dockerfile . +# Build Jepsen test container (run from d-engine workspace root) +docker build -t jepsen:1.0 -f examples/three-nodes-standalone/docker/jepsen/Dockerfile . ``` ### 3. Start the Cluster diff --git a/examples/three-nodes-standalone/docker/jepsen/Dockerfile b/examples/three-nodes-standalone/docker/jepsen/Dockerfile index 660ecc97..cf821f21 100644 --- a/examples/three-nodes-standalone/docker/jepsen/Dockerfile +++ b/examples/three-nodes-standalone/docker/jepsen/Dockerfile @@ -1,18 +1,16 @@ -# Use the official Rust image (automatically matches the ARM64 architecture) +# Build the dengine CLI binary from workspace root FROM rust:latest AS builder -# Set the working directory -WORKDIR /usr/src/dengine_ctl +WORKDIR /build # Install the protobuf compiler RUN apt-get update && apt-get install -y protobuf-compiler -# Copy the Rust project files -COPY vendor vendor +# Copy workspace source (target/ excluded via .dockerignore) +COPY . . -# Build the Rust components -WORKDIR /usr/src/dengine_ctl/vendor/dengine_ctl -RUN cargo build --release +# Build only the CLI package +RUN cargo build --package client-usage-standalone-demo --release # Use the JDK 21 base image (official ARM64 image) FROM eclipse-temurin:21-jdk-jammy @@ -43,7 +41,7 @@ ENV PATH="/root/.cargo/bin:${PATH}" # Configure SSH Environment -COPY sshkeys /root/.ssh +COPY examples/three-nodes-standalone/docker/jepsen/sshkeys /root/.ssh RUN chmod 600 /root/.ssh/id_rsa && \ mkdir /var/run/sshd && \ echo 'root:root' | chpasswd && \ @@ -55,14 +53,14 @@ RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && \ # Copy project files WORKDIR /app -COPY src src -COPY project.clj project.clj +COPY examples/three-nodes-standalone/docker/jepsen/src src +COPY examples/three-nodes-standalone/docker/jepsen/project.clj project.clj # Install Clojure dependencies RUN lein deps # Copy the compiled Rust binary -COPY --from=builder /usr/src/dengine_ctl/vendor/dengine_ctl/target/release/dengine_ctl /usr/local/bin/ +COPY --from=builder /build/target/release/client-usage-standalone-demo /usr/local/bin/ EXPOSE 22 CMD ["/bin/sh", "-c", "/usr/sbin/sshd -D && eval $(ssh-agent -s) && ssh-add /root/.ssh/id_rsa"] diff --git a/examples/three-nodes-standalone/docker/jepsen/Makefile b/examples/three-nodes-standalone/docker/jepsen/Makefile index 6d7f9c97..ec0340d6 100644 --- a/examples/three-nodes-standalone/docker/jepsen/Makefile +++ b/examples/three-nodes-standalone/docker/jepsen/Makefile @@ -3,7 +3,8 @@ # Configurable parameters TIME_LIMIT ?= 60 -TEST_COMMAND ?= dengine_ctl +TEST_COMMAND ?= client-usage-standalone-demo +WORKLOAD ?= register NODE1 ?= node1 NODE2 ?= node2 NODE3 ?= node3 @@ -33,7 +34,8 @@ test: restart-stack --node '"${NODE3}"' \ --endpoints '"${ENDPOINTS}"' \ --time-limit '"${TIME_LIMIT}"' \ - --command '"${TEST_COMMAND}"'' + --command '"${TEST_COMMAND}"' \ + --workload '"${WORKLOAD}"'' @echo "Jepsen test finished, checking result..." docker exec docker-jepsen-1 bash -c '\ lein trampoline run -m clojure.main -e "\ diff --git a/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine.clj b/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine.clj index 337cfe68..c96454bf 100644 --- a/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine.clj +++ b/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine.clj @@ -17,7 +17,9 @@ [jepsen.os :as os] [knossos.model :as model] [slingshot.slingshot :refer [try+]] - [clojure.tools.cli :refer [parse-opts]])) + [clojure.tools.cli :refer [parse-opts]] + [jepsen.d_engine.bank :as bank] + [jepsen.d_engine.set :as set-workload])) ;; ========== Operation Definition ========== ;; Update operation commands to match v0.1.4 API @@ -132,51 +134,66 @@ :algorithm :linear}) :timeline (timeline/html)}))) +(defn register-workload + "Original linearizable register workload." + [opts] + {:client (client (:command opts) (:endpoints opts)) + :checker (independent/checker + (checker/compose + {:linear (checker/linearizable {:model (model/cas-register) + :algorithm :auto}) + :timeline (timeline/html)})) + :generator (independent/concurrent-generator + 3 + (range 3) + (fn [k] + (->> (gen/mix [r w]) + (gen/stagger 1/2) + (gen/limit 40))))}) + +(defn workload + [opts] + (case (:workload opts) + "bank" (bank/workload opts) + "set" (set-workload/workload opts) + (register-workload opts))) + (defn test-spec [opts] (println "opts:" opts) (println "Time limit set to:" (:time-limit opts)) - (merge tests/noop-test - {:name "d-engine" - :ssh {:private-key-path "/root/.ssh/id_rsa" - :strict-host-key-checking false} - :client (client (:command opts) (:endpoints opts)) - :nemesis (nemesis/partition-random-halves) - ;; Final checker definition - :checker (independent/checker - (checker/compose - {;; :perf (checker/perf) - :linear (checker/linearizable {:model (model/cas-register) - :algorithm :auto}) - :timeline (timeline/html) - })) - :generator (->> (independent/concurrent-generator - 3 ; Concurrent 3 independent key spaces - (range 3) ; Keyspace - (fn [k] - (->> (gen/mix [r w]) ; Mixed lread and write - (gen/stagger 1/2) - (gen/limit 40)))) - ; Fault injection: inject into a partition every 15 seconds - (gen/nemesis - (cycle [(gen/sleep 10) - {:type :info, :f :start} ; More aggressive partition triggering - (gen/sleep 5) - {:type :info, :f :stop}])) - ; Test time limit - (gen/time-limit (:time-limit opts)))} - opts)) + (let [wl (workload opts)] + (merge tests/noop-test + {:name (str "d-engine-" (:workload opts "register")) + :ssh {:private-key-path "/root/.ssh/id_rsa" + :strict-host-key-checking false} + :client (:client wl) + :nemesis (nemesis/partition-random-halves) + :checker (:checker wl) + :generator (->> (:generator wl) + (gen/stagger 1/10) + (gen/nemesis + (cycle [(gen/sleep 10) + {:type :info, :f :start} + (gen/sleep 5) + {:type :info, :f :stop}])) + (gen/time-limit (:time-limit opts)))} + opts))) (def cli-opts "Additional command line options." - [["-c" "--command CMD" "dengine_ctl command path" - :default "dengine_ctl" ;; Provide a default value if needed - :parse-fn identity ;; Use `identity` if you want to pass the string as it is - :validate [(complement empty?) "command-path cannot be empty."]] - ["-e" "--endpoints ENDPOINTS" "The endpoints for the client." - :default "http://node1:9080,http://node2:9080,http://node3:9080" ;; Provide a default value if needed - :parse-fn identity ;; Use `identity` if you want to pass the string as it is - :validate [(complement empty?) "Endpoints cannot be empty."]]]) + [["-c" "--command CMD" "CLI binary path" + :default "client-usage-standalone-demo" + :parse-fn identity + :validate [(complement empty?) "command cannot be empty."]] + ["-e" "--endpoints ENDPOINTS" "d-engine gRPC endpoints (comma-separated)" + :default "http://node1:9080,http://node2:9080,http://node3:9080" + :parse-fn identity + :validate [(complement empty?) "endpoints cannot be empty."]] + ["-w" "--workload NAME" "Workload to run: register (default), bank, set" + :default "register" + :parse-fn identity + :validate [#{"register" "bank" "set"} "must be one of: register, bank, set"]]]) (defn -main "handles command lien arguments" diff --git a/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/bank.clj b/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/bank.clj new file mode 100644 index 00000000..1be91235 --- /dev/null +++ b/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/bank.clj @@ -0,0 +1,100 @@ +(ns jepsen.d_engine.bank + "Bank workload: concurrent transfers under faults must preserve total balance. + Encodes 3 account balances as a single packed u64 for atomic CAS." + (:require [clojure.tools.logging :refer [info]] + [clojure.string :as str] + [clojure.java.shell :as shell] + [jepsen [checker :as checker] + [client :as client] + [generator :as gen]] + [slingshot.slingshot :refer [try+]])) + +(def bank-key 1) +(def n-accounts 3) +(def init-balance 1000) + +;; Pack 3 balances (21 bits each) into a single u64. +(defn pack [a b c] + (bit-or (bit-shift-left (long a) 42) + (bit-shift-left (long b) 21) + (long c))) + +(defn unpack [packed] + [(bit-and (bit-shift-right packed 42) 0x1FFFFF) + (bit-and (bit-shift-right packed 21) 0x1FFFFF) + (bit-and packed 0x1FFFFF)]) + +(def init-packed (pack init-balance init-balance init-balance)) + +(defn ctl! [cmd endpoints & args] + (let [result (apply shell/sh cmd "--endpoints" endpoints args)] + (if (zero? (:exit result)) + (str/trim (:out result)) + (throw (ex-info "ctl failed" {:err (:err result) :exit (:exit result)}))))) + +(defn parse-long-safe [s] + (when (and s (not (str/includes? s "not found"))) + (try (Long/parseLong (str/trim s)) + (catch NumberFormatException _ nil)))) + +(defrecord BankClient [cmd endpoints] + client/Client + + (open! [this test node] this) + + (setup! [this test] + (ctl! cmd endpoints "put" (str bank-key) (str init-packed))) + + (invoke! [this test op] + (try+ + (case (:f op) + :read + (let [packed (or (parse-long-safe (ctl! cmd endpoints "lget" (str bank-key))) + init-packed) + balances (unpack packed)] + (assoc op :type :ok :value (zipmap (range n-accounts) balances))) + + :transfer + (let [{:keys [from to amount]} (:value op)] + (loop [] + (let [packed (or (parse-long-safe (ctl! cmd endpoints "lget" (str bank-key))) + init-packed) + bals (vec (unpack packed)) + from-bal (nth bals from)] + (if (< from-bal amount) + (assoc op :type :fail :error :insufficient-funds) + (let [new-bals (-> bals (update from - amount) (update to + amount)) + new-packed (apply pack new-bals) + swapped (ctl! cmd endpoints "cas" + (str bank-key) (str packed) (str new-packed))] + (if (= swapped "true") + (assoc op :type :ok) + (recur)))))))) + + (catch Exception e + (assoc op :type :info :error (.getMessage e))))) + + (teardown! [this test]) + (close! [this test])) + +(defn bank-checker [] + (let [expected (* n-accounts init-balance)] + (reify checker/Checker + (check [_ test history opts] + (let [bad (->> history + (filter #(and (= :ok (:type %)) (= :read (:f %)))) + (remove #(= expected (reduce + (vals (:value %))))))] + {:valid? (empty? bad) + :bad-reads (take 10 bad)}))))) + +(defn r [_ _] {:type :invoke :f :read :value nil}) +(defn t [_ _] + {:type :invoke :f :transfer + :value {:from (rand-int n-accounts) + :to (rand-int n-accounts) + :amount (inc (rand-int 5))}}) + +(defn workload [opts] + {:client (BankClient. (:command opts) (:endpoints opts)) + :checker (bank-checker) + :generator (gen/mix [t r])}) diff --git a/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/set.clj b/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/set.clj new file mode 100644 index 00000000..e0bbda8d --- /dev/null +++ b/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/set.clj @@ -0,0 +1,69 @@ +(ns jepsen.d_engine.set + "Set workload: every acknowledged add must survive faults and appear in reads. + Encodes a set of integers 0-62 as a u64 bitmask; CAS ensures atomic updates." + (:require [clojure.tools.logging :refer [info]] + [clojure.string :as str] + [clojure.java.shell :as shell] + [jepsen [checker :as checker] + [client :as client] + [generator :as gen]] + [slingshot.slingshot :refer [try+]])) + +(def set-key 42) + +(defn ctl! [cmd endpoints & args] + (let [result (apply shell/sh cmd "--endpoints" endpoints args)] + (if (zero? (:exit result)) + (str/trim (:out result)) + (throw (ex-info "ctl failed" {:err (:err result) :exit (:exit result)}))))) + +(defn parse-long-safe [s] + (when (and s (not (str/includes? s "not found"))) + (try (Long/parseLong (str/trim s)) + (catch NumberFormatException _ nil)))) + +(defn decode-set + "Decode a u64 bitmask into a set of present element indices." + [packed] + (->> (range 63) + (filter #(not= 0 (bit-and packed (bit-shift-left 1 %)))) + set)) + +(defrecord SetClient [cmd endpoints] + client/Client + + (open! [this test node] this) + + (setup! [this test] + (ctl! cmd endpoints "put" (str set-key) "0")) + + (invoke! [this test op] + (try+ + (case (:f op) + :add + (loop [] + (let [current (or (parse-long-safe (ctl! cmd endpoints "lget" (str set-key))) 0) + new-val (bit-or current (bit-shift-left 1 (long (:value op)))) + swapped (ctl! cmd endpoints "cas" + (str set-key) (str current) (str new-val))] + (if (= swapped "true") + (assoc op :type :ok) + (recur)))) + + :read + (let [packed (or (parse-long-safe (ctl! cmd endpoints "lget" (str set-key))) 0)] + (assoc op :type :ok :value (decode-set packed)))) + + (catch Exception e + (assoc op :type :fail :error (.getMessage e))))) + + (teardown! [this test]) + (close! [this test])) + +(defn add-op [_ _] {:type :invoke :f :add :value (rand-int 30)}) +(defn read-op [_ _] {:type :invoke :f :read :value nil}) + +(defn workload [opts] + {:client (SetClient. (:command opts) (:endpoints opts)) + :checker (checker/set-full) + :generator (gen/mix [add-op read-op])}) diff --git a/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/Cargo.toml b/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/Cargo.toml deleted file mode 100644 index e587f4bf..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/Cargo.toml +++ /dev/null @@ -1,15 +0,0 @@ -[package] -name = "dengine_ctl" -version = "0.2.0" -edition = "2024" - -[dependencies] -d-engine = { git = "https://github.com/deventlab/d-engine.git", branch = "feature/79-snapshot" } - -clap = { version = "4.0", features = ["derive"] } -tokio = { version = "1.36.0", features = ["full"] } -env_logger = "0.9.0" -anyhow = "1.0" - -[profile.dev] -incremental = true diff --git a/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/README.md b/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/README.md deleted file mode 100644 index 5304c453..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/README.md +++ /dev/null @@ -1,31 +0,0 @@ -# DEngine Client Example - -Command-line interface demonstrating DEngine's CRUD operations with Raft consensus. - -## Features - -- Put/Update key-value pairs -- Delete keys -- Read values (eventual consistency) -- Linearizable reads (strong consistency) - -## Usage - -```bash -# Put key-value pair (creates or updates) -cargo run -- --endpoints "http://127.0.0.1:9083,http://127.0.0.1:9082,http://127.0.0.1:9081" put 42 100 - -# Get value (eventually consistent) -cargo run -- --endpoints "http://127.0.0.1:9083,http://127.0.0.1:9082,http://127.0.0.1:9081" get 42 - -# Linearizable read (strongly consistent) -cargo run -- --endpoints "http://127.0.0.1:9083,http://127.0.0.1:9082,http://127.0.0.1:9081" lget 42 - -# Delete key -cargo run -- --endpoints "http://127.0.0.1:9083,http://127.0.0.1:9082,http://127.0.0.1:9081" delete 42 - -# Join cluster -cargo run -- --endpoints "http://127.0.0.1:9083,http://127.0.0.1:9082,http://127.0.0.1:9081" join \ - --node-id 100 \ - --address 127.0.0.1:9084 -``` diff --git a/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/src/main.rs b/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/src/main.rs deleted file mode 100644 index 07b5888a..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/src/main.rs +++ /dev/null @@ -1,116 +0,0 @@ -use anyhow::Result; -use clap::Parser; -use clap::Subcommand; -use d_engine::client::ClientApiError; -use d_engine::ClientApi; -use d_engine::ClientBuilder; -use d_engine::ConvertError; - -#[derive(Parser)] -#[command(version, about, long_about = None)] -struct Cli { - #[arg(long, default_value = "http://127.0.0.1:9083")] - endpoints: String, - #[command(subcommand)] - command: Commands, -} - -#[derive(Subcommand)] -enum Commands { - Put { key: u64, value: u64 }, - Delete { key: u64 }, - Get { key: u64 }, - Lget { key: u64 }, -} - -#[tokio::main] -async fn main() -> Result<()> { - let cli = Cli::parse(); - - // Create client connection - let client = ClientBuilder::new(cli.endpoints.split(',').map(String::from).collect()) - .connect_timeout(std::time::Duration::from_secs(10)) - .request_timeout(std::time::Duration::from_secs(10)) - .build() - .await - .map_err(|e| anyhow::anyhow!("Failed to create client: {:?}", e))?; - - match cli.command { - Commands::Put { key, value } => handle_write(&client, key, value).await, - Commands::Delete { key } => handle_delete(&client, key).await, - Commands::Get { key } => handle_read(&client, key, false).await, - Commands::Lget { key } => handle_read(&client, key, true).await, - } -} - -async fn handle_write( - client: &d_engine::Client, - key: u64, - value: u64, -) -> Result<()> { - println!("Writing key-value({}-{}) pair", key, value); - client - .put(safe_kv(key), safe_kv(value)) - .await - .map(|_| println!("Success")) - .map_err(|e: ClientApiError| anyhow::anyhow!("Write error: {:?}", e)) -} - -async fn handle_delete( - client: &d_engine::Client, - key: u64, -) -> Result<()> { - println!("Deleting key({})", key); - client - .delete(safe_kv(key)) - .await - .map(|_| println!("Success")) - .map_err(|e: ClientApiError| anyhow::anyhow!("Delete error: {:?}", e)) -} - -async fn handle_read( - client: &d_engine::Client, - key: u64, - linearizable: bool, -) -> Result<()> { - let result = if linearizable { - client.get_linearizable(safe_kv(key)).await - } else { - client.get_eventual(safe_kv(key)).await - } - .map_err(|e: ClientApiError| anyhow::anyhow!("Read error: {:?}", e))?; - - match result { - Some(value) => { - let value = safe_vk(&value).unwrap(); - println!("{:?}", value); - } - None => println!("Key not found"), - } - - Ok(()) -} - -pub fn safe_vk>(bytes: K) -> crate::Result { - let bytes = bytes.as_ref(); - let expected_len = 8; - - if bytes.len() != expected_len { - return Err(ConvertError::InvalidLength(bytes.len()).into()); - } - let array: [u8; 8] = bytes.try_into().expect("Guaranteed safe after length check"); - Ok(u64::from_be_bytes(array)) -} - -/// Converts a `u64` to an 8-byte array in big-endian byte order. -/// -/// # Examples -/// ``` -/// use d_engine::convert::safe_kv; -/// -/// let bytes = safe_kv(0x1234_5678_9ABC_DEF0); -/// assert_eq!(bytes, [0x12, 0x34, 0x56, 0x78, 0x9A, 0xBC, 0xDE, 0xF0]); -/// ``` -pub const fn safe_kv(num: u64) -> [u8; 8] { - num.to_be_bytes() -} From a73c95b2ce7f3b4ea73888d260a0dbdd0a372331 Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Sat, 9 May 2026 18:49:21 +0800 Subject: [PATCH 3/6] fix #369: remove embedded jepsen, fix Dockerfile build and config mount - Delete docker/jepsen/: moved to d-engine-jepsen repo - Delete docker/config/n*.toml: config now mounted via ../config directly - Fix Dockerfile: workspace-root build context, bookworm base, BuildKit cache mounts - Fix docker-compose.yml: config volume default ./config -> ../config - Update README: remove symlink step, reference d-engine-jepsen for tests --- .../three-nodes-standalone/docker/Dockerfile | 48 ++-- .../three-nodes-standalone/docker/README.md | 66 ++---- .../docker/config/n1.toml | 53 ----- .../docker/config/n2.toml | 53 ----- .../docker/config/n3.toml | 53 ----- .../docker/config/n4.toml | 58 ----- .../docker/config/n5.toml | 60 ----- .../docker/docker-compose.yml | 6 +- .../docker/jepsen/Dockerfile | 66 ------ .../docker/jepsen/Makefile | 82 ------- .../docker/jepsen/README.md | 97 --------- .../docker/jepsen/entrypoint.sh | 10 - .../docker/jepsen/project.clj | 13 -- .../docker/jepsen/src/jepsen/d_engine.clj | 205 ------------------ .../jepsen/src/jepsen/d_engine/bank.clj | 100 --------- .../docker/jepsen/src/jepsen/d_engine/set.clj | 69 ------ 16 files changed, 49 insertions(+), 990 deletions(-) delete mode 100644 examples/three-nodes-standalone/docker/config/n1.toml delete mode 100644 examples/three-nodes-standalone/docker/config/n2.toml delete mode 100644 examples/three-nodes-standalone/docker/config/n3.toml delete mode 100644 examples/three-nodes-standalone/docker/config/n4.toml delete mode 100644 examples/three-nodes-standalone/docker/config/n5.toml delete mode 100644 examples/three-nodes-standalone/docker/jepsen/Dockerfile delete mode 100644 examples/three-nodes-standalone/docker/jepsen/Makefile delete mode 100644 examples/three-nodes-standalone/docker/jepsen/README.md delete mode 100644 examples/three-nodes-standalone/docker/jepsen/entrypoint.sh delete mode 100644 examples/three-nodes-standalone/docker/jepsen/project.clj delete mode 100644 examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine.clj delete mode 100644 examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/bank.clj delete mode 100644 examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/set.clj diff --git a/examples/three-nodes-standalone/docker/Dockerfile b/examples/three-nodes-standalone/docker/Dockerfile index 69a56bfa..ea11ec15 100644 --- a/examples/three-nodes-standalone/docker/Dockerfile +++ b/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,28 +12,38 @@ 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 \ @@ -65,16 +73,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"] diff --git a/examples/three-nodes-standalone/docker/README.md b/examples/three-nodes-standalone/docker/README.md index 9dac1076..c20b5f30 100644 --- a/examples/three-nodes-standalone/docker/README.md +++ b/examples/three-nodes-standalone/docker/README.md @@ -1,75 +1,47 @@ -# Distributed Engine Jepsen Testing with Docker +# Distributed Engine Docker Setup ## Project Setup ### 1. Configure Environment ```bash -# Create required directories (run in the project root directory) -mkdir -p docker/output/{logs,db} docker/jepsen/store +cd ../d-engine/ -# Create configuration symlink (run only once) -ln -sf $(pwd)/config docker/config +# Create required directories (run from d-engine workspace root) +mkdir -p examples/three-nodes-standalone/docker/output/{logs,db} ``` ### 2. Build Docker Images ```bash -# Build core component (run in the project root directory) -docker build -t demo:1.3 -f docker/Dockerfile . +# Build core component (run from d-engine workspace root) +docker build -t demo:v0.2.4 -f examples/three-nodes-standalone/docker/Dockerfile . # Build monitoring component -docker build -t prometheus:1.0 -f docker/monitoring/prometheus/Dockerfile . +docker build -t prometheus:1.0 -f examples/three-nodes-standalone/docker/monitoring/prometheus/Dockerfile . -# Build Jepsen test container (run from d-engine workspace root) -docker build -t jepsen:1.0 -f examples/three-nodes-standalone/docker/jepsen/Dockerfile . +# Build Jepsen test container (run from d-engine-jepsen directory) +cd ../d-engine-jepsen/ +docker compose build jepsen ``` ### 3. Start the Cluster ```bash -# Start the main cluster (run in the docker/ directory) -docker-compose -f docker/docker-compose.yml up -d +# Start the main cluster (from d-engine examples/three-nodes-standalone/docker/) +docker-compose -f examples/three-nodes-standalone/docker/docker-compose.yml up -d # Start the monitoring system -docker-compose -f docker/monitoring/docker-compose.yml up -d +docker-compose -f examples/three-nodes-standalone/docker/monitoring/docker-compose.yml up -d ``` -### 4. Run the Tests - -```bash -# Enter the Jepsen container -docker exec -it jepsen bash - -# Run the test inside the container -lein run test --time-limit 60 -``` - -## Jepsen Distributed Systems Testing - -### Running the Jepsen Test Suite - -To validate the linearizability and fault tolerance of the distributed engine, we include a Jepsen test setup. Follow these steps to execute the tests: - -#### 1. Build the Jepsen Test Container - -```bash -cd docker/jepsen -docker build -t jepsen:1.0 -f Dockerfile . -``` - -### 2. Run the Test Suite +### 4. Run Jepsen Tests ```bash +# From d-engine-jepsen directory +cd ../d-engine-jepsen/ make test -``` -This will: - -1. Start the Jepsen test container -2. Execute the linearizability test with fault injection -3. Output either **`✅ PASS`** or **`❌ FAIL`** based on test results - -### Test Configuration Options - -Customize tests by modifying **`docker/jepsen/src/jepsen/d_engine.clj`**: +# Run a specific workload (register / bank / set) +make test WORKLOAD=bank +``` diff --git a/examples/three-nodes-standalone/docker/config/n1.toml b/examples/three-nodes-standalone/docker/config/n1.toml deleted file mode 100644 index 5e5d0d87..00000000 --- a/examples/three-nodes-standalone/docker/config/n1.toml +++ /dev/null @@ -1,53 +0,0 @@ -[cluster] -node_id = 1 -listen_address = "192.168.0.11:9081" -initial_cluster = [ - { id = 1, address = "192.168.0.11:9081", role = 1, status = 3 }, - { id = 2, address = "192.168.0.12:9082", role = 1, status = 3 }, - { id = 3, address = "192.168.0.13:9083", role = 1, status = 3 }, -] -db_root_dir = "./db" -log_dir = "./logs" - -[raft] -general_raft_timeout_duration_in_ms = 100 - - -[raft.commit_handler] -batch_size = 5000 - -[raft.persistence] -strategy = "MemFirst" -# strategy = "DiskFirst" -flush_policy = { Batch = { idle_flush_interval_ms = 100 } } - -[raft.snapshot] -enable = true -max_log_entries_before_snapshot = 10000 -snapshot_cool_down_since_last_check = { secs = 60 } -snapshots_dir = "./snapshots/" -retained_log_entries = 3 - -# == Network Control Plane (voting, heartbeat, etc.) == -[network.control] -concurrency_limit = 20 # Lower control plane concurrency is recommended -max_concurrent_streams = 128 -connection_window_size = 2_097_152 # 2MB is sufficient (to prevent memory overflow) - -tcp_keepalive_in_secs = 60 # Closer to native behavior -http2_keep_alive_interval_in_secs = 10 -http2_keep_alive_timeout_in_secs = 5 -max_frame_size = 16_777_215 # Default value is acceptable - -# == Network Data Plane (append_entries, etc.) == -[network.data] -connect_timeout_in_ms = 100 # Slow is fine, ensure the connection is established -request_timeout_in_ms = 300 # Log synchronization may take a long time. -concurrency_limit = 30 # High concurrency supports data replication. -max_concurrent_streams = 256 # Sufficient; higher values may be counterproductive. -connection_window_size = 4_194_304 # Increase the buffer size to support batch data. - -tcp_keepalive_in_secs = 60 -http2_keep_alive_interval_in_secs = 10 -http2_keep_alive_timeout_in_secs = 5 -max_frame_size = 16_777_215 diff --git a/examples/three-nodes-standalone/docker/config/n2.toml b/examples/three-nodes-standalone/docker/config/n2.toml deleted file mode 100644 index ef73ff4a..00000000 --- a/examples/three-nodes-standalone/docker/config/n2.toml +++ /dev/null @@ -1,53 +0,0 @@ -[cluster] -node_id = 2 -listen_address = "192.168.0.12:9082" -initial_cluster = [ - { id = 1, address = "192.168.0.11:9081", role = 1, status = 3 }, - { id = 2, address = "192.168.0.12:9082", role = 1, status = 3 }, - { id = 3, address = "192.168.0.13:9083", role = 1, status = 3 }, -] -db_root_dir = "./db" -log_dir = "./logs" - - -[raft] -general_raft_timeout_duration_in_ms = 100 - -[raft.commit_handler] -batch_size = 5000 - - -[raft.persistence] -strategy = "MemFirst" -# strategy = "DiskFirst" -flush_policy = { Batch = { idle_flush_interval_ms = 100 } } - -[raft.snapshot] -enable = true -max_log_entries_before_snapshot = 10000 -snapshot_cool_down_since_last_check = { secs = 60 } -snapshots_dir = "./snapshots/" -retained_log_entries = 3 - - -# == Network Control Plane (voting, heartbeat, etc.) == -[network.control] -concurrency_limit = 20 # Lower control plane concurrency is recommended -max_concurrent_streams = 128 -tcp_keepalive_in_secs = 60 # Closer to native behavior -http2_keep_alive_interval_in_secs = 10 -http2_keep_alive_timeout_in_secs = 5 -max_frame_size = 16_777_215 # Default value is acceptable -connection_window_size = 2_097_152 # 2MB is sufficient (to prevent memory overflow) - -# == Network Data Plane (append_entries, etc.) == -[network.data] -connect_timeout_in_ms = 100 # Slow is fine, ensure the connection is established -request_timeout_in_ms = 300 # Log synchronization may take a long time. -concurrency_limit = 30 # High concurrency supports data replication. -max_concurrent_streams = 256 # Sufficient; higher values may be counterproductive. -tcp_keepalive_in_secs = 60 -http2_keep_alive_interval_in_secs = 10 -http2_keep_alive_timeout_in_secs = 5 -max_frame_size = 16_777_215 -connection_window_size = 4_194_304 # Increase the buffer size to support batch data. diff --git a/examples/three-nodes-standalone/docker/config/n3.toml b/examples/three-nodes-standalone/docker/config/n3.toml deleted file mode 100644 index a2ff2a3f..00000000 --- a/examples/three-nodes-standalone/docker/config/n3.toml +++ /dev/null @@ -1,53 +0,0 @@ -[cluster] -node_id = 3 -listen_address = "192.168.0.13:9083" -initial_cluster = [ - { id = 1, address = "192.168.0.11:9081", role = 1, status = 3 }, - { id = 2, address = "192.168.0.12:9082", role = 1, status = 3 }, - { id = 3, address = "192.168.0.13:9083", role = 1, status = 3 }, -] -db_root_dir = "./db" -log_dir = "./logs" - -[raft] -general_raft_timeout_duration_in_ms = 100 - - -[raft.commit_handler] -batch_size = 5000 - - -[raft.persistence] -strategy = "MemFirst" -# strategy = "DiskFirst" -flush_policy = { Batch = { idle_flush_interval_ms = 100 } } - - -[raft.snapshot] -enable = true -max_log_entries_before_snapshot = 10000 -snapshot_cool_down_since_last_check = { secs = 60 } -snapshots_dir = "./snapshots/" -retained_log_entries = 3 - -# == Network Control Plane (voting, heartbeat, etc.) == -[network.control] -concurrency_limit = 20 # Lower control plane concurrency is recommended -max_concurrent_streams = 128 -tcp_keepalive_in_secs = 60 # Closer to native behavior -http2_keep_alive_interval_in_secs = 10 -http2_keep_alive_timeout_in_secs = 5 -max_frame_size = 16_777_215 # Default value is acceptable -connection_window_size = 2_097_152 # 2MB is sufficient (to prevent memory overflow) - -# == Network Data Plane (append_entries, etc.) == -[network.data] -connect_timeout_in_ms = 100 # Slow is fine, ensure the connection is established -request_timeout_in_ms = 300 # Log synchronization may take a long time. -concurrency_limit = 30 # High concurrency supports data replication. -max_concurrent_streams = 256 # Sufficient; higher values may be counterproductive. -tcp_keepalive_in_secs = 60 -http2_keep_alive_interval_in_secs = 10 -http2_keep_alive_timeout_in_secs = 5 -max_frame_size = 16_777_215 -connection_window_size = 4_194_304 # Increase the buffer size to support batch data. diff --git a/examples/three-nodes-standalone/docker/config/n4.toml b/examples/three-nodes-standalone/docker/config/n4.toml deleted file mode 100644 index bafe0d9d..00000000 --- a/examples/three-nodes-standalone/docker/config/n4.toml +++ /dev/null @@ -1,58 +0,0 @@ -[cluster] -node_id = 4 -listen_address = "192.168.0.14:9084" -initial_cluster = [ - { id = 1, address = "192.168.0.11:9081", role = 1, status = 3 }, - { id = 2, address = "192.168.0.12:9082", role = 1, status = 3 }, - { id = 3, address = "192.168.0.13:9083", role = 1, status = 3 }, - { id = 4, address = "192.168.0.14:9084", role = 4, status = 1 }, -] -db_root_dir = "./db" -log_dir = "./logs" - -[raft] -general_raft_timeout_duration_in_ms = 100 - -[raft.commit_handler] -batch_size = 5000 - - -[raft.persistence] -strategy = "MemFirst" -# strategy = "DiskFirst" -flush_policy = { Batch = { idle_flush_interval_ms = 100 } } - - -[retry.join_cluster] -max_retries = 3 # Maximum number of retries (recommended: 3) -timeout_ms = 500 # Single request timeout (recommended: 500-1000ms) -base_delay_ms = 3000 # Initial backoff delay (recommended: 3000ms) -max_delay_ms = 60000 # Maximum backoff delay (recommended: <=1m) - -[raft.snapshot] -enable = false -max_log_entries_before_snapshot = 10000 -snapshot_cool_down_since_last_check = { secs = 60 } -snapshots_dir = "./snapshots/" -retained_log_entries = 3 - -[network.control] -concurrency_limit = 8192 #// Moderate concurrency -max_concurrent_streams = 8192 #// Stream limit for control operations -tcp_keepalive_in_secs = 3600 #// 5 minute TCP keepalive -http2_keep_alive_interval_in_secs = 300 #// Frequent pings -http2_keep_alive_timeout_in_secs = 20 #// Short timeout -max_frame_size = 16_777_215 -connection_window_size = 2_097_152 #// 2MB connection window - - -[network.data] -connect_timeout_in_ms = 20 #// Balance speed and reliability -request_timeout_in_ms = 200 #// Accommodate log batches -concurrency_limit = 8192 #// High concurrency for parallel logs -max_concurrent_streams = 500 #// More streams for pipelining -tcp_keepalive_in_secs = 3600 #// 10 minute TCP keepalive -http2_keep_alive_interval_in_secs = 300 #// Moderate ping interval -http2_keep_alive_timeout_in_secs = 30 #// Longer grace period -max_frame_size = 16_777_215 -connection_window_size = 6_291_456 #// 6MB connection window diff --git a/examples/three-nodes-standalone/docker/config/n5.toml b/examples/three-nodes-standalone/docker/config/n5.toml deleted file mode 100644 index 565fd9c4..00000000 --- a/examples/three-nodes-standalone/docker/config/n5.toml +++ /dev/null @@ -1,60 +0,0 @@ -[cluster] -node_id = 5 -listen_address = "192.168.0.15:9085" -initial_cluster = [ - { id = 1, address = "192.168.0.11:9081", role = 1, status = 3 }, - { id = 2, address = "192.168.0.12:9082", role = 1, status = 3 }, - { id = 3, address = "192.168.0.13:9083", role = 1, status = 3 }, - { id = 4, address = "192.168.0.14:9084", role = 4, status = 1 }, - { id = 5, address = "192.168.0.15:9085", role = 4, status = 1 }, -] -db_root_dir = "./db" -log_dir = "./logs" - - -[raft] -general_raft_timeout_duration_in_ms = 100 - -[raft.commit_handler] -batch_size = 5000 - - -[raft.persistence] -strategy = "MemFirst" -# strategy = "DiskFirst" -flush_policy = { Batch = { idle_flush_interval_ms = 100 } } - - -[retry.join_cluster] -max_retries = 3 # Maximum number of retries (recommended: 3) -timeout_ms = 500 # Single request timeout (recommended: 500-1000ms) -base_delay_ms = 3000 # Initial backoff delay (recommended: 3000ms) -max_delay_ms = 60000 # Maximum backoff delay (recommended: <=1m) - -[raft.snapshot] -enable = false -max_log_entries_before_snapshot = 10000 -snapshot_cool_down_since_last_check = { secs = 60 } -snapshots_dir = "./snapshots/" -retained_log_entries = 3 - -[network.control] -concurrency_limit = 8192 #// Moderate concurrency -max_concurrent_streams = 8192 #// Stream limit for control operations -tcp_keepalive_in_secs = 3600 #// 5 minute TCP keepalive -http2_keep_alive_interval_in_secs = 300 #// Frequent pings -http2_keep_alive_timeout_in_secs = 20 #// Short timeout -max_frame_size = 16_777_215 -connection_window_size = 2_097_152 #// 2MB connection window - - -[network.data] -connect_timeout_in_ms = 20 #// Balance speed and reliability -request_timeout_in_ms = 200 #// Accommodate log batches -concurrency_limit = 8192 #// High concurrency for parallel logs -max_concurrent_streams = 500 #// More streams for pipelining -tcp_keepalive_in_secs = 3600 #// 10 minute TCP keepalive -http2_keep_alive_interval_in_secs = 300 #// Moderate ping interval -http2_keep_alive_timeout_in_secs = 30 #// Longer grace period -max_frame_size = 16_777_215 -connection_window_size = 6_291_456 #// 6MB connection window diff --git a/examples/three-nodes-standalone/docker/docker-compose.yml b/examples/three-nodes-standalone/docker/docker-compose.yml index df89204a..e5afb0c9 100644 --- a/examples/three-nodes-standalone/docker/docker-compose.yml +++ b/examples/three-nodes-standalone/docker/docker-compose.yml @@ -1,6 +1,6 @@ services: node1: &node-base - image: demo:1.3 + image: demo:v0.2.4 cap_add: - NET_ADMIN # Add the NET_ADMIN capability environment: @@ -20,7 +20,7 @@ services: - "9081:9081" volumes: - ${SSH_KEYS_PATH:-./jepsen/sshkeys}:/root/.ssh - - ${CONFIG_PATH:-./config}:/app/config + - ${CONFIG_PATH:-../config}:/app/config - ${LOGS_PATH:-./output/logs}:/app/logs - ${DB_PATH:-./output/db}:/app/db # - ${SNAPSHOTS_PATH:-./output/snapshots}:/app/snapshots @@ -62,7 +62,7 @@ services: - "9083:9083" jepsen: - image: jepsen:1.0 + image: jepsen:2.0 cap_add: - NET_ADMIN # Add the NET_ADMIN capability environment: diff --git a/examples/three-nodes-standalone/docker/jepsen/Dockerfile b/examples/three-nodes-standalone/docker/jepsen/Dockerfile deleted file mode 100644 index cf821f21..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/Dockerfile +++ /dev/null @@ -1,66 +0,0 @@ -# Build the dengine CLI binary from workspace root -FROM rust:latest AS builder - -WORKDIR /build - -# Install the protobuf compiler -RUN apt-get update && apt-get install -y protobuf-compiler - -# Copy workspace source (target/ excluded via .dockerignore) -COPY . . - -# Build only the CLI package -RUN cargo build --package client-usage-standalone-demo --release - -# Use the JDK 21 base image (official ARM64 image) -FROM eclipse-temurin:21-jdk-jammy - -# Install Leiningen -RUN apt-get update && \ - apt-get install -y \ - curl \ - git \ - gcc \ - build-essential \ - openssh-server \ - telnet \ - procps \ - net-tools \ - lsof \ - gnuplot \ - vim && \ - curl -L https://raw.githubusercontent.com/technomancy/leiningen/stable/bin/lein > /usr/local/bin/lein && \ - chmod +x /usr/local/bin/lein && \ - lein self-install && \ - apt-get clean && \ - rm -rf /var/lib/apt/lists/* - -# Install the Rust toolchain -RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y -ENV PATH="/root/.cargo/bin:${PATH}" - - -# Configure SSH Environment -COPY examples/three-nodes-standalone/docker/jepsen/sshkeys /root/.ssh -RUN chmod 600 /root/.ssh/id_rsa && \ - mkdir /var/run/sshd && \ - echo 'root:root' | chpasswd && \ - sed -i 's/#PermitRootLogin prohibit-password/PermitRootLogin yes/' /etc/ssh/sshd_config - -# Set the time zone (Asia/Shanghai) -RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && \ - echo "Asia/Shanghai" > /etc/timezone - -# Copy project files -WORKDIR /app -COPY examples/three-nodes-standalone/docker/jepsen/src src -COPY examples/three-nodes-standalone/docker/jepsen/project.clj project.clj - -# Install Clojure dependencies -RUN lein deps - -# Copy the compiled Rust binary -COPY --from=builder /build/target/release/client-usage-standalone-demo /usr/local/bin/ - -EXPOSE 22 -CMD ["/bin/sh", "-c", "/usr/sbin/sshd -D && eval $(ssh-agent -s) && ssh-add /root/.ssh/id_rsa"] diff --git a/examples/three-nodes-standalone/docker/jepsen/Makefile b/examples/three-nodes-standalone/docker/jepsen/Makefile deleted file mode 100644 index ec0340d6..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/Makefile +++ /dev/null @@ -1,82 +0,0 @@ -# docker/jepsen/Makefile -.PHONY: test view report clean - -# Configurable parameters -TIME_LIMIT ?= 60 -TEST_COMMAND ?= client-usage-standalone-demo -WORKLOAD ?= register -NODE1 ?= node1 -NODE2 ?= node2 -NODE3 ?= node3 -ENDPOINTS ?= http://node1:9081,http://node2:9082,http://node3:9083 -COMPOSE_FILE ?= ../docker-compose.yml - -# Restart Docker Compose stack -restart-stack: - @echo "Cleaning output directories..." - @rm -rf ../output/logs/* ../output/db/* - @echo "Restarting Docker Compose stack..." - @docker compose -f $(COMPOSE_FILE) down - @docker compose -f $(COMPOSE_FILE) up -d - @echo "Waiting for cluster to initialize (10 seconds)..." - @sleep 10 - - -# Main test target -test: restart-stack - @echo "Starting Jepsen test with time limit: ${TIME_LIMIT}s" - docker exec -e SSH_AUTH_SOCK=/ssh-agent docker-jepsen-1 bash -c '\ - eval "$$(ssh-agent -s)" && \ - ssh-add /root/.ssh/id_rsa && \ - lein run test \ - --node '"${NODE1}"' \ - --node '"${NODE2}"' \ - --node '"${NODE3}"' \ - --endpoints '"${ENDPOINTS}"' \ - --time-limit '"${TIME_LIMIT}"' \ - --command '"${TEST_COMMAND}"' \ - --workload '"${WORKLOAD}"'' - @echo "Jepsen test finished, checking result..." - docker exec docker-jepsen-1 bash -c '\ - lein trampoline run -m clojure.main -e "\ - (require '\''[knossos.model :as model])\ - (println \ - (if (-> (clojure.edn/read-string \ - {:readers {\ - '\''knossos.model.Register model/->Register\ - '\''knossos.model.CASRegister model/->CASRegister\ - '\''knossos.model.Inconsistent model/->Inconsistent}\ - }\ - (slurp \"/app/store/latest/results.edn\"))\ - :valid?)\ - \"✅ PASS\" \"❌ FAIL\"))"' - -# Set up SSH agent inside container -ssh-setup: - @echo "Configuring SSH keys..." - @docker exec docker-jepsen-1 bash -c "\ - eval \$$(ssh-agent -s) && \ - ssh-add /root/.ssh/id_rsa" - -# View latest test results -view: - @latest=$$(readlink ./store/latest); \ - if [ -z "$$latest" ]; then \ - echo "No test results found"; \ - exit 1; \ - fi; \ - open "$$(pwd)/store/$$latest/index.html" 2>/dev/null || \ - echo "Open manually: file://$$(pwd)/$$latest/index.html" - -# Show path to latest report -report: - @latest=$$(readlink ./store/latest); \ - if [ -z "$$latest" ]; then \ - echo "No test results available"; \ - exit 1; \ - fi; \ - echo "Latest report: $$(pwd)/store/$$latest" - -# Clean test artifacts -clean: - rm -rf ./store/* diff --git a/examples/three-nodes-standalone/docker/jepsen/README.md b/examples/three-nodes-standalone/docker/jepsen/README.md deleted file mode 100644 index 2851d4a2..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/README.md +++ /dev/null @@ -1,97 +0,0 @@ -# Jepsen Tests for dengine - -This directory contains [Jepsen](https://jepsen.io/) tests for validating the correctness of the `d-engine` cluster under various failure scenarios. - -## Prerequisites - -- Docker & Docker Compose -- Java (for Leiningen / Clojure tooling) -- Rust toolchain (for building `dengine`) -- SSH keys available under `jepsen/sshkeys/` (mounted into the Jepsen container) -- **Example cluster(three-nodes-standalone) running:** - -## Usage - -The `Makefile` provides a set of common commands for running and managing Jepsen tests. - -### 1. Restart the Docker stack - -```bash -make restart-stack -``` - - • Cleans logs and database data in ../output/ - • Restarts the Docker Compose cluster defined in ../docker-compose.yml - • Waits for the cluster to initialize - -2. Run Jepsen tests - -```bash -make test TIME_LIMIT=120 TEST_COMMAND=dengine_ctl -``` - - • Spins up the Jepsen container - • Executes the configured Jepsen test (lein run test) - • Runs against nodes defined by NODE1, NODE2, NODE3 - • Default runtime is 60 seconds (override via TIME_LIMIT) - -Environment variables you can override: -• TIME_LIMIT (default: 60) -• TEST_COMMAND (default: dengine_ctl) -• NODE1, NODE2, NODE3 (default: node1, node2, node3) -• ENDPOINTS (default: http://node1:9081,http://node2:9082,http://node3:9083) - -3. Configure SSH inside container - -If you need to manually set up the SSH agent in the Jepsen container: - -```bash -make ssh-setup -``` - -4. View the latest report - -```bash -make view -``` - - • Opens the latest Jepsen HTML report in your browser - • If auto-open fails, it will print the file path - -5. Print report path - -```bash -make report -``` - - • Prints the absolute path to the latest Jepsen report - -6. Clean test artifacts - -```bash -make clean -``` - - • Removes all artifacts under jepsen/store/ - -Output -• Test results are stored under jepsen/store/ -• The latest symlink points to the most recent test run -• Reports include Jepsen logs and visualization of histories - -⸻ - -Example - -Run a 2-minute Jepsen test against a three-node cluster: - -```bash -make test -make view -``` - -⸻ - -Notes -• Make sure your cluster nodes (node1, node2, node3) are reachable from inside the Jepsen container. -• For deeper debugging, you can docker exec -it docker-jepsen-1 bash and run lein run directly. diff --git a/examples/three-nodes-standalone/docker/jepsen/entrypoint.sh b/examples/three-nodes-standalone/docker/jepsen/entrypoint.sh deleted file mode 100644 index cb5658c2..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/entrypoint.sh +++ /dev/null @@ -1,10 +0,0 @@ -#!/bin/bash - -# Start SSH agent for container lifetime -eval $(ssh-agent -a /ssh-agent) > /dev/null - -# Add SSH keys (silently) -ssh-add /root/.ssh/id_rsa > /dev/null 2>&1 - -# Execute main command -exec "$@" diff --git a/examples/three-nodes-standalone/docker/jepsen/project.clj b/examples/three-nodes-standalone/docker/jepsen/project.clj deleted file mode 100644 index 9933c3d2..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/project.clj +++ /dev/null @@ -1,13 +0,0 @@ -(defproject jepsen.dengine "0.1.4" - :description "A Jepsen test for d-engine" - :license {:name "MIT License" - :url "https://opensource.org/licenses/MIT"} - :main jepsen.d_engine - :dependencies [[org.clojure/clojure "1.11.3"] - [jepsen "0.3.5"] - [knossos "0.3.8"] - [verschlimmbesserung "0.1.3"] - [net.java.dev.jna/jna "5.12.1"]] - :jvm-opts ["-Djava.awt.headless=true" - "-server" - "-Xmx72g"]) diff --git a/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine.clj b/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine.clj deleted file mode 100644 index c96454bf..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine.clj +++ /dev/null @@ -1,205 +0,0 @@ -(ns jepsen.d_engine - (:require - [clojure.tools.logging :refer :all] - [clojure.string :as str] - [verschlimmbesserung.core :as v] - [jepsen [checker :as checker] - [cli :as cli] - [client :as client] - [control :as c] - [generator :as gen] - [independent :as independent] - [nemesis :as nemesis] - [tests :as tests]] - [jepsen.checker.timeline :as timeline] - [jepsen.control.util :as cu] - [clojure.java.shell :as shell] - [jepsen.os :as os] - [knossos.model :as model] - [slingshot.slingshot :refer [try+]] - [clojure.tools.cli :refer [parse-opts]] - [jepsen.d_engine.bank :as bank] - [jepsen.d_engine.set :as set-workload])) - -;; ========== Operation Definition ========== -;; Update operation commands to match v0.1.4 API -(defn r [_ _] {:type :invoke, :f :read, :value nil}) -(defn w [_ _] {:type :invoke, :f :write, :value (rand-int 5)}) - -;; Adjust mixed ratio: 70% linear reads + 30% normal reads -(def mixed-reads (gen/mix [{:weight 7, :gen r} {:weight 3, :gen r}])) - -;; ========== Command execution tool ========== -(defn ctl-command - "Execute the dengine_ctl command and process the output" - [cmd & args] - (let [command (concat [cmd] args) ; Ensure everything is a string - _ (info "Executing command: " (pr-str command)) ; Better logging - result (apply shell/sh command)] ; Add Rust debugging information - (println "Executing command:" command) ;; Log the command being executed - (info "Command output:" (:out result)) - (info "Command error:" (:err result)) - (if (zero? (:exit result)) - (do - (println "Success:" (:out result)) ;; Log success output - (:out result)) - (throw (ex-info "Command failed" - {:exit (:exit result) - :err (:err result) - :out (:out result)}))))) - -;; ========== Client Implementation ========== -(defn parse-long-nil - "Parses a string to a Long. Passes through `nil`." - [s] - (println "Parsing raw string:" (pr-str s)) - (when s - (try - (-> s - (clojure.string/trim) ; Clean leading and trailing whitespace (including newlines) - (Long/parseLong)) - (catch NumberFormatException _ - nil)))) ; Return nil if parsing fails - -(defn client - "A client for a single compare-and-set register" - [cmd endpoints] - (assert (string? endpoints) (str "ENDPOINTS MUST BE STRING. GOT: " endpoints)) - (reify client/Client - - (open! [this test node] - (info "Opening client for node:" node) - this) - - (invoke! [this test op] - (println "Received operation:" op) - (let [[k v] (:value op) - ; Read operation failures are marked as :fail, write operations are marked as :info - crash-type (if (#{:get :lget} (:f op)) :fail :info)] - (try+ - (case (:f op) - ; Linear read (lget) - :read (let [result (parse-long-nil - (ctl-command cmd "--endpoints" endpoints "lget" (str k)))] - (if result - (assoc op :type :ok, :value (independent/tuple k result)) - (assoc op :type :fail, :error :not-found))) - -; Write operation (put) - :write (do - (println "endpoints:" endpoints "; Putting value:" v "for key:" k) - (ctl-command cmd "--endpoints" endpoints "put" (str k) (str v)) - (assoc op :type :ok))) - -; ===== Error handling ===== - (catch java.net.SocketTimeoutException e - (assoc op :type crash-type, :error :timeout)) - - (catch [:exit 4005] e ; cluster not available - (assoc op :type crash-type :error :cluster-unavailable)) - - (catch Exception e - (let [err-msg (or (.getMessage e) - (some-> e ex-data :err) - (some-> e ex-data :body))] - (cond - (and err-msg (str/includes? (str/lower-case err-msg) "key not found")) - (assoc op :type :fail :error :not-found) - - (and err-msg (str/includes? (str/lower-case err-msg) "cluster unavailable")) - (assoc op :type crash-type :error :cluster-unavailable) - - :else - (assoc op :type crash-type :error (or err-msg "unknown error")))))))) - - (close! [_ _] - (info "Closing client")) - - (setup! [_ _]) - (teardown! [_ _]))) - -;; ========== Checker Implementation ========== -(defn split-history - "Split History: Linear Reads vs. Normal Operations" - [history] - (group-by (fn [op] (if (= :lget (:f op)) :lget-ops :other-ops)) history)) - -(defn checker - "Linearizability checker" - [test history opts] - (independent/checker - (checker/compose - {;; :perf (checker/perf) - :linear (checker/linearizable {:model (model/cas-register) - :algorithm :linear}) - :timeline (timeline/html)}))) - -(defn register-workload - "Original linearizable register workload." - [opts] - {:client (client (:command opts) (:endpoints opts)) - :checker (independent/checker - (checker/compose - {:linear (checker/linearizable {:model (model/cas-register) - :algorithm :auto}) - :timeline (timeline/html)})) - :generator (independent/concurrent-generator - 3 - (range 3) - (fn [k] - (->> (gen/mix [r w]) - (gen/stagger 1/2) - (gen/limit 40))))}) - -(defn workload - [opts] - (case (:workload opts) - "bank" (bank/workload opts) - "set" (set-workload/workload opts) - (register-workload opts))) - -(defn test-spec - [opts] - (println "opts:" opts) - (println "Time limit set to:" (:time-limit opts)) - (let [wl (workload opts)] - (merge tests/noop-test - {:name (str "d-engine-" (:workload opts "register")) - :ssh {:private-key-path "/root/.ssh/id_rsa" - :strict-host-key-checking false} - :client (:client wl) - :nemesis (nemesis/partition-random-halves) - :checker (:checker wl) - :generator (->> (:generator wl) - (gen/stagger 1/10) - (gen/nemesis - (cycle [(gen/sleep 10) - {:type :info, :f :start} - (gen/sleep 5) - {:type :info, :f :stop}])) - (gen/time-limit (:time-limit opts)))} - opts))) - -(def cli-opts - "Additional command line options." - [["-c" "--command CMD" "CLI binary path" - :default "client-usage-standalone-demo" - :parse-fn identity - :validate [(complement empty?) "command cannot be empty."]] - ["-e" "--endpoints ENDPOINTS" "d-engine gRPC endpoints (comma-separated)" - :default "http://node1:9080,http://node2:9080,http://node3:9080" - :parse-fn identity - :validate [(complement empty?) "endpoints cannot be empty."]] - ["-w" "--workload NAME" "Workload to run: register (default), bank, set" - :default "register" - :parse-fn identity - :validate [#{"register" "bank" "set"} "must be one of: register, bank, set"]]]) - -(defn -main - "handles command lien arguments" - [& args] - (cli/run! - (merge (cli/single-test-cmd {:test-fn test-spec - :opt-spec cli-opts}) - (cli/serve-cmd)) - args)) diff --git a/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/bank.clj b/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/bank.clj deleted file mode 100644 index 1be91235..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/bank.clj +++ /dev/null @@ -1,100 +0,0 @@ -(ns jepsen.d_engine.bank - "Bank workload: concurrent transfers under faults must preserve total balance. - Encodes 3 account balances as a single packed u64 for atomic CAS." - (:require [clojure.tools.logging :refer [info]] - [clojure.string :as str] - [clojure.java.shell :as shell] - [jepsen [checker :as checker] - [client :as client] - [generator :as gen]] - [slingshot.slingshot :refer [try+]])) - -(def bank-key 1) -(def n-accounts 3) -(def init-balance 1000) - -;; Pack 3 balances (21 bits each) into a single u64. -(defn pack [a b c] - (bit-or (bit-shift-left (long a) 42) - (bit-shift-left (long b) 21) - (long c))) - -(defn unpack [packed] - [(bit-and (bit-shift-right packed 42) 0x1FFFFF) - (bit-and (bit-shift-right packed 21) 0x1FFFFF) - (bit-and packed 0x1FFFFF)]) - -(def init-packed (pack init-balance init-balance init-balance)) - -(defn ctl! [cmd endpoints & args] - (let [result (apply shell/sh cmd "--endpoints" endpoints args)] - (if (zero? (:exit result)) - (str/trim (:out result)) - (throw (ex-info "ctl failed" {:err (:err result) :exit (:exit result)}))))) - -(defn parse-long-safe [s] - (when (and s (not (str/includes? s "not found"))) - (try (Long/parseLong (str/trim s)) - (catch NumberFormatException _ nil)))) - -(defrecord BankClient [cmd endpoints] - client/Client - - (open! [this test node] this) - - (setup! [this test] - (ctl! cmd endpoints "put" (str bank-key) (str init-packed))) - - (invoke! [this test op] - (try+ - (case (:f op) - :read - (let [packed (or (parse-long-safe (ctl! cmd endpoints "lget" (str bank-key))) - init-packed) - balances (unpack packed)] - (assoc op :type :ok :value (zipmap (range n-accounts) balances))) - - :transfer - (let [{:keys [from to amount]} (:value op)] - (loop [] - (let [packed (or (parse-long-safe (ctl! cmd endpoints "lget" (str bank-key))) - init-packed) - bals (vec (unpack packed)) - from-bal (nth bals from)] - (if (< from-bal amount) - (assoc op :type :fail :error :insufficient-funds) - (let [new-bals (-> bals (update from - amount) (update to + amount)) - new-packed (apply pack new-bals) - swapped (ctl! cmd endpoints "cas" - (str bank-key) (str packed) (str new-packed))] - (if (= swapped "true") - (assoc op :type :ok) - (recur)))))))) - - (catch Exception e - (assoc op :type :info :error (.getMessage e))))) - - (teardown! [this test]) - (close! [this test])) - -(defn bank-checker [] - (let [expected (* n-accounts init-balance)] - (reify checker/Checker - (check [_ test history opts] - (let [bad (->> history - (filter #(and (= :ok (:type %)) (= :read (:f %)))) - (remove #(= expected (reduce + (vals (:value %))))))] - {:valid? (empty? bad) - :bad-reads (take 10 bad)}))))) - -(defn r [_ _] {:type :invoke :f :read :value nil}) -(defn t [_ _] - {:type :invoke :f :transfer - :value {:from (rand-int n-accounts) - :to (rand-int n-accounts) - :amount (inc (rand-int 5))}}) - -(defn workload [opts] - {:client (BankClient. (:command opts) (:endpoints opts)) - :checker (bank-checker) - :generator (gen/mix [t r])}) diff --git a/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/set.clj b/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/set.clj deleted file mode 100644 index e0bbda8d..00000000 --- a/examples/three-nodes-standalone/docker/jepsen/src/jepsen/d_engine/set.clj +++ /dev/null @@ -1,69 +0,0 @@ -(ns jepsen.d_engine.set - "Set workload: every acknowledged add must survive faults and appear in reads. - Encodes a set of integers 0-62 as a u64 bitmask; CAS ensures atomic updates." - (:require [clojure.tools.logging :refer [info]] - [clojure.string :as str] - [clojure.java.shell :as shell] - [jepsen [checker :as checker] - [client :as client] - [generator :as gen]] - [slingshot.slingshot :refer [try+]])) - -(def set-key 42) - -(defn ctl! [cmd endpoints & args] - (let [result (apply shell/sh cmd "--endpoints" endpoints args)] - (if (zero? (:exit result)) - (str/trim (:out result)) - (throw (ex-info "ctl failed" {:err (:err result) :exit (:exit result)}))))) - -(defn parse-long-safe [s] - (when (and s (not (str/includes? s "not found"))) - (try (Long/parseLong (str/trim s)) - (catch NumberFormatException _ nil)))) - -(defn decode-set - "Decode a u64 bitmask into a set of present element indices." - [packed] - (->> (range 63) - (filter #(not= 0 (bit-and packed (bit-shift-left 1 %)))) - set)) - -(defrecord SetClient [cmd endpoints] - client/Client - - (open! [this test node] this) - - (setup! [this test] - (ctl! cmd endpoints "put" (str set-key) "0")) - - (invoke! [this test op] - (try+ - (case (:f op) - :add - (loop [] - (let [current (or (parse-long-safe (ctl! cmd endpoints "lget" (str set-key))) 0) - new-val (bit-or current (bit-shift-left 1 (long (:value op)))) - swapped (ctl! cmd endpoints "cas" - (str set-key) (str current) (str new-val))] - (if (= swapped "true") - (assoc op :type :ok) - (recur)))) - - :read - (let [packed (or (parse-long-safe (ctl! cmd endpoints "lget" (str set-key))) 0)] - (assoc op :type :ok :value (decode-set packed)))) - - (catch Exception e - (assoc op :type :fail :error (.getMessage e))))) - - (teardown! [this test]) - (close! [this test])) - -(defn add-op [_ _] {:type :invoke :f :add :value (rand-int 30)}) -(defn read-op [_ _] {:type :invoke :f :read :value nil}) - -(defn workload [opts] - {:client (SetClient. (:command opts) (:endpoints opts)) - :checker (checker/set-full) - :generator (gen/mix [add-op read-op])}) From b31ffc4f2585bb054acb90bf5b9a18814b429650 Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Sun, 10 May 2026 10:11:36 +0800 Subject: [PATCH 4/6] feat #369: add iproute2 and wget to node Docker image for Jepsen nemesis - iproute2: partition nemesis needs `ip` to enumerate network interfaces - wget: nc/nemesis-packages setup! downloads bitflip binary via wget --- examples/three-nodes-standalone/docker/Dockerfile | 2 ++ 1 file changed, 2 insertions(+) diff --git a/examples/three-nodes-standalone/docker/Dockerfile b/examples/three-nodes-standalone/docker/Dockerfile index ea11ec15..df52fc4b 100644 --- a/examples/three-nodes-standalone/docker/Dockerfile +++ b/examples/three-nodes-standalone/docker/Dockerfile @@ -49,9 +49,11 @@ RUN apt-get update && \ openssh-server \ ca-certificates \ curl \ + wget \ unzip \ sudo \ iptables \ + iproute2 \ libc6 \ tzdata && \ rm -rf /var/lib/apt/lists/* && \ From 3c4751bb7a9b66651ac5d06e133e310c021a99bc Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Sun, 10 May 2026 15:19:27 +0800 Subject: [PATCH 5/6] style #369: fix rustfmt formatting in bank_transfer_invariant_embedded --- .../bank_transfer_invariant_embedded.rs | 23 +++++++++++-------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/d-engine-server/tests/cas_operations/bank_transfer_invariant_embedded.rs b/d-engine-server/tests/cas_operations/bank_transfer_invariant_embedded.rs index acfaac7b..d7e66711 100644 --- a/d-engine-server/tests/cas_operations/bank_transfer_invariant_embedded.rs +++ b/d-engine-server/tests/cas_operations/bank_transfer_invariant_embedded.rs @@ -26,7 +26,11 @@ fn pack(balances: [u64; 3]) -> [u8; 8] { 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] + [ + (state >> 42) & 0x1FFFFF, + (state >> 21) & 0x1FFFFF, + state & 0x1FFFFF, + ] } /// Concurrent CAS transfers maintain bank total invariant under follower failure. @@ -44,8 +48,8 @@ fn unpack(bytes: &[u8]) -> [u64; 3] { #[tokio::test] #[traced_test] #[serial] -async fn test_concurrent_transfers_preserve_bank_invariant( -) -> Result<(), Box> { +async fn test_concurrent_transfers_preserve_bank_invariant() +-> Result<(), Box> { let temp_dir = tempfile::tempdir()?; let db_root_dir = temp_dir.path().join("db"); let log_dir = temp_dir.path().join("logs"); @@ -136,14 +140,10 @@ async fn test_concurrent_transfers_preserve_bank_invariant( let new_bytes = pack(balances); match client - .compare_and_swap( - BANK_KEY, - Some(current.as_ref()), - new_bytes.as_slice(), - ) + .compare_and_swap(BANK_KEY, Some(current.as_ref()), new_bytes.as_slice()) .await { - Ok(true) => break, // committed + Ok(true) => break, // committed Ok(false) => { // Concurrent update; retry without delay. tokio::task::yield_now().await; @@ -162,7 +162,10 @@ async fn test_concurrent_transfers_preserve_bank_invariant( // 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); + info!( + "Stopping follower node {} (2/3 quorum maintained)", + follower_idx + 1 + ); engines[follower_idx].stop().await?; // Wait for all workers to finish. From 85adb4da57ffdf05f2e2f456e6a233d8c527b17f Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Sun, 10 May 2026 15:58:41 +0800 Subject: [PATCH 6/6] docs #369: clarify docker-compose working dir and exclude AI rules from git - Exclude .rules and .github/copilot-instructions.md from version control (consistent with existing CLAUDE.md / .claude/ policy) - Fix README ambiguity: docker-compose must run from docker/ so that relative volume paths (../config) resolve correctly --- .gitignore | 2 ++ examples/three-nodes-standalone/docker/README.md | 10 +++++++--- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/.gitignore b/.gitignore index b587a1ec..b91857fb 100644 --- a/.gitignore +++ b/.gitignore @@ -51,3 +51,5 @@ CONTEXT safe-split-commits.sh CLAUDE.md .claude/ +.rules +.github/copilot-instructions.md diff --git a/examples/three-nodes-standalone/docker/README.md b/examples/three-nodes-standalone/docker/README.md index c20b5f30..eae25948 100644 --- a/examples/three-nodes-standalone/docker/README.md +++ b/examples/three-nodes-standalone/docker/README.md @@ -28,11 +28,15 @@ docker compose build jepsen ### 3. Start the Cluster ```bash -# Start the main cluster (from d-engine examples/three-nodes-standalone/docker/) -docker-compose -f examples/three-nodes-standalone/docker/docker-compose.yml up -d +# All docker-compose commands must be run from the docker/ directory so that +# relative volume paths (e.g. ../config) resolve correctly. +cd examples/three-nodes-standalone/docker/ + +# Start the main cluster +docker-compose up -d # Start the monitoring system -docker-compose -f examples/three-nodes-standalone/docker/monitoring/docker-compose.yml up -d +docker-compose -f monitoring/docker-compose.yml up -d ``` ### 4. Run Jepsen Tests