Add HW 7

homework/07-kv-sharding/tests/.dockerignore

@@ -0,0 +1,5 @@
+target
+**/target
+**/__pycache__
+**/.pytest_cache
+*.pyc

homework/07-kv-sharding/tests/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "distsys-kv-sharding"
+version = "0.1.0"
+edition = "2021"
+resolver = "3"
+
+[dependencies]
+anysystem = "=0.3.0"
+indexmap = "2.12"
+assertables = "3.2.2"
+clap = { version = "3.2.17", features = ["cargo", "derive"] }
+decorum = "0.3.1"
+env_logger = "0.9.0"
+log = "0.4.17"
+pyo3 = { version = "=0.29.2", features = ["auto-initialize"] }
+rand = "0.8.5"
+rand_pcg = "0.3.1"
+serde = { version = "1.0.142", features = ["derive"] }
+serde_json = "1.0.83"
+sugars = "3.0.1"

homework/07-kv-sharding/tests/Dockerfile

@@ -0,0 +1,19 @@
+# syntax=docker/dockerfile:1
+FROM rust:1.97.1-bookworm@sha256:0e2bcaef56d041a486784e54104a81aebe0da44bd03019bd70bc0401e42e4a97 AS builder
+
+RUN apt-get update && apt-get install -y build-essential python3-dev
+
+WORKDIR /tests
+COPY . .
+RUN --mount=type=cache,id=distsys-course-cargo-registry,target=/usr/local/cargo/registry,sharing=locked \
+    --mount=type=cache,id=distsys-course-cargo-target-rust-1-97-1,target=/tests/target,sharing=locked \
+    cargo install --locked --path . --root /opt/course
+
+FROM debian:bookworm-slim
+
+RUN apt-get update && apt-get install -y python3-dev && rm -rf /var/lib/apt/lists/*
+
+COPY --from=builder /opt/course/bin/distsys-kv-sharding /usr/local/bin/distsys-kv-sharding
+
+WORKDIR /solution
+ENTRYPOINT ["timeout", "--verbose", "-k", "10", "600", "distsys-kv-sharding", "-i", "node.py"]

homework/07-kv-sharding/tests/src/common.rs

@@ -0,0 +1,125 @@
+use std::collections::HashSet;
+
+use rand::distributions::WeightedIndex;
+use rand::prelude::*;
+use rand_pcg::{Lcg128Xsl64, Pcg64};
+use serde::{Deserialize, Serialize};
+use sugars::boxed;
+
+use anysystem::python::PyProcessFactory;
+use anysystem::System;
+
+#[derive(Serialize)]
+pub struct GetReqMessage<'a> {
+    pub key: &'a str,
+}
+
+#[derive(Deserialize)]
+pub struct GetRespMessage<'a> {
+    pub key: &'a str,
+    pub value: Option<&'a str>,
+}
+
+#[derive(Serialize)]
+pub struct PutReqMessage<'a> {
+    pub key: &'a str,
+    pub value: &'a str,
+}
+
+#[derive(Deserialize)]
+pub struct PutRespMessage<'a> {
+    pub key: &'a str,
+    pub value: &'a str,
+}
+
+#[derive(Serialize)]
+pub struct DeleteReqMessage<'a> {
+    pub key: &'a str,
+}
+
+#[derive(Deserialize)]
+pub struct DeleteRespMessage<'a> {
+    pub key: &'a str,
+    pub value: Option<&'a str>,
+}
+
+#[derive(Serialize)]
+pub struct EmptyMessage {}
+
+#[derive(Serialize)]
+pub struct NodeMessage<'a> {
+    pub id: &'a str,
+}
+
+#[derive(Deserialize)]
+pub struct DumpKeysRespMessage {
+    pub keys: HashSet<String>,
+}
+
+#[derive(Deserialize)]
+pub struct CountRecordsRespMessage {
+    pub count: u64,
+}
+
+#[derive(Copy, Clone)]
+pub struct TestConfig<'a> {
+    pub process_factory: &'a PyProcessFactory,
+    pub proc_count: u32,
+    pub seed: u64,
+}
+
+pub fn build_system(config: &TestConfig, measure_max_size: bool) -> System {
+    let mut sys = System::new(config.seed);
+    sys.network().set_delays(0.01, 0.1);
+    let mut proc_names = Vec::new();
+    for n in 0..config.proc_count {
+        proc_names.push(format!("{n}"));
+    }
+    for n in 0..config.proc_count {
+        let proc_name = proc_names[n as usize].clone();
+        let mut proc = config
+            .process_factory
+            .build((proc_name.clone(), proc_names.clone()), config.seed);
+        if measure_max_size {
+            proc.set_max_size_freq(1000000);
+        }
+        // process and node on which it runs have the same name
+        let node_name = proc_name.clone();
+        sys.add_node(&node_name);
+        sys.add_process(&proc_name, boxed!(proc), &node_name);
+    }
+    sys
+}
+
+pub fn add_node(name: &str, sys: &mut System, config: &TestConfig) {
+    let proc_name = name.to_string();
+    let mut proc_names = sys.process_names();
+    proc_names.push(proc_name.clone());
+    let proc = config
+        .process_factory
+        .build((proc_name.clone(), proc_names), config.seed);
+    let node_name = proc_name.clone();
+    sys.add_node(&node_name);
+    sys.add_process(&proc_name, boxed!(proc), &node_name);
+}
+
+pub fn random_proc(sys: &System, rand: &mut Lcg128Xsl64) -> String {
+    sys.process_names().choose(rand).unwrap().clone()
+}
+
+const SYMBOLS: [char; 36] = [
+    'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's',
+    't', 'u', 'v', 'w', 'x', 'y', 'z', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9',
+];
+const WEIGHTS: [usize; 36] = [
+    13, 16, 3, 8, 8, 5, 6, 23, 4, 8, 24, 12, 2, 1, 1, 10, 5, 8, 10, 1, 24, 3, 1, 8, 12, 22, 5, 20,
+    18, 5, 5, 2, 1, 3, 16, 22,
+];
+
+pub fn random_string(length: usize, rand: &mut Pcg64) -> String {
+    let dist = WeightedIndex::new(WEIGHTS).unwrap();
+    rand.sample_iter(&dist)
+        .take(length)
+        .map(|x| SYMBOLS[x])
+        .collect()
+}

homework/07-kv-sharding/tests/src/main.rs

@@ -0,0 +1,254 @@
+mod common;
+mod python_seed;
+mod shared_state;
+mod tests;
+mod tests_mc;
+
+use indexmap::IndexMap;
+use std::io::Write;
+
+use clap::Parser;
+use env_logger::Builder;
+use log::LevelFilter;
+
+use anysystem::python::PyProcessFactory;
+use anysystem::test::{TestResult, TestSuite};
+
+use crate::common::TestConfig;
+use crate::tests::*;
+use crate::tests_mc::*;
+
+/// Sharded KV Store Homework Tests
+#[derive(Parser, Debug)]
+#[clap(about, long_about = None)]
+struct Args {
+    /// Path to Python file with solution
+    #[clap(long = "impl", short = 'i', default_value = "solution/node.py")]
+    solution_path: String,
+
+    /// Test to run (optional)
+    #[clap(long = "test", short)]
+    test: Option<String>,
+
+    /// Print execution trace
+    #[clap(long, short)]
+    debug: bool,
+
+    /// Number of nodes used in tests
+    #[clap(long, short, default_value = "10")]
+    node_count: u32,
+
+    /// Random seed used in tests
+    #[clap(long, short, default_value = "123")]
+    seed: u64,
+}
+
+fn main() {
+    let args = Args::parse();
+    python_seed::configure(args.seed);
+    if !shared_state::validate_or_report(&args.solution_path, &["StorageNode"]) {
+        return;
+    }
+    if args.debug {
+        Builder::new()
+            .filter(Some("anysystem"), LevelFilter::Debug)
+            .format(|buf, record| writeln!(buf, "{}", record.args()))
+            .init();
+    }
+    let process_factory = PyProcessFactory::new(&args.solution_path, "StorageNode");
+    let config = TestConfig {
+        process_factory: &process_factory,
+        proc_count: args.node_count,
+        seed: args.seed,
+    };
+    let mut single_config = config;
+    single_config.proc_count = 1;
+    let mut mc_config = config;
+    mc_config.proc_count = 3;
+    let mut tests = TestSuite::new();
+
+    tests.add("SINGLE NODE", test_single_node, single_config);
+    tests.add("INSERTS", test_inserts, config);
+    tests.add("DELETES", test_deletes, config);
+    tests.add("MEMORY GROWTH", test_memory_growth, config);
+    tests.add("ZERO HOP", test_zero_hop, config);
+    tests.add("MC NORMAL", test_mc_normal, mc_config);
+    tests.add("NODE ADDED", test_node_added, config);
+    tests.add(
+        "NODE ADDED DURING INSERTS",
+        test_node_added_during_inserts,
+        config,
+    );
+    tests.add(
+        "NODE ADDED DURING DELETIONS",
+        test_node_added_during_deletions,
+        config,
+    );
+    tests.add(
+        "NODE ADDED DURING READS",
+        test_node_added_during_reads,
+        config,
+    );
+    tests.add("NODE REMOVED", test_node_removed, config);
+    tests.add(
+        "NODE REMOVED AFTER CRASH",
+        test_node_removed_after_crash,
+        config,
+    );
+    tests.add("MC NODE REMOVED", test_mc_node_removed, mc_config);
+    tests.add("MIGRATION", test_migration, config);
+    tests.add("SCALE UP DOWN", test_scale_up_down, config);
+    tests.add("DISTRIBUTION", test_distribution, config);
+    tests.add(
+        "DISTRIBUTION NODE ADDED",
+        test_distribution_node_added,
+        config,
+    );
+    tests.add(
+        "DISTRIBUTION NODE REMOVED",
+        test_distribution_node_removed,
+        config,
+    );
+
+    if let Some(test) = &args.test {
+        tests.run_test(test);
+    } else {
+        let (_, results) = tests.run();
+        let score = score(results);
+        println!("\nSCORE: {score}\n");
+    }
+}
+
+const BASIC_TESTS: &[&str] = &[
+    "SINGLE NODE",
+    "INSERTS",
+    "DELETES",
+    "MEMORY GROWTH",
+    "ZERO HOP",
+    "MC NORMAL",
+];
+
+const ADDITIONAL_GROUPS: &[(&str, &[&str])] = &[
+    (
+        "Node changes",
+        &[
+            "NODE ADDED",
+            "NODE REMOVED",
+            "NODE REMOVED AFTER CRASH",
+            "MC NODE REMOVED",
+            "MIGRATION",
+            "SCALE UP DOWN",
+        ],
+    ),
+    (
+        "Requests during node addition",
+        &[
+            "NODE ADDED DURING INSERTS",
+            "NODE ADDED DURING DELETIONS",
+            "NODE ADDED DURING READS",
+        ],
+    ),
+    (
+        "Distribution",
+        &[
+            "DISTRIBUTION",
+            "DISTRIBUTION NODE ADDED",
+            "DISTRIBUTION NODE REMOVED",
+        ],
+    ),
+];
+
+fn group_passed(name: &str, tests: &[&str], results: &mut IndexMap<String, TestResult>) -> bool {
+    let mut failed = Vec::new();
+    for &test in tests {
+        let result = results
+            .swap_remove(test)
+            .unwrap_or_else(|| panic!("missing result for {test}"));
+        if result.is_err() {
+            failed.push(test);
+        }
+    }
+    if failed.is_empty() {
+        println!("{name} tests: passed");
+        true
+    } else {
+        println!("{name} tests: not passed {failed:?}");
+        false
+    }
+}
+
+fn score(mut results: IndexMap<String, TestResult>) -> u32 {
+    let basic_passed = group_passed("Basic", BASIC_TESTS, &mut results);
+    let mut points = if basic_passed { 3 } else { 0 };
+    for &(name, tests) in ADDITIONAL_GROUPS {
+        if group_passed(name, tests, &mut results) && basic_passed {
+            points += 2;
+        }
+    }
+    assert!(results.is_empty(), "unscored tests: {:?}", results.keys());
+    points
+}
+
+#[cfg(test)]
+mod scoring_tests {
+    use super::*;
+
+    fn score_with_failures(failed: &[&str]) -> u32 {
+        let mut results: IndexMap<String, TestResult> = BASIC_TESTS
+            .iter()
+            .chain(ADDITIONAL_GROUPS.iter().flat_map(|(_, tests)| tests.iter()))
+            .map(|&name| (name.to_string(), Ok(true)))
+            .collect();
+        for &name in failed {
+            *results.get_mut(name).unwrap() = Err("failed".to_string());
+        }
+        score(results)
+    }
+
+    #[test]
+    fn every_basic_test_is_required() {
+        for name in [
+            "SINGLE NODE",
+            "INSERTS",
+            "DELETES",
+            "MC NORMAL",
+            "ZERO HOP",
+            "MEMORY GROWTH",
+        ] {
+            assert_eq!(score_with_failures(&[name]), 0, "{name}");
+        }
+    }
+
+    #[test]
+    fn additional_groups_earn_two_points_each() {
+        let change = "NODE ADDED";
+        let read = "NODE ADDED DURING READS";
+        let balance = "DISTRIBUTION NODE ADDED";
+        let cases: &[(&[&str], u32)] = &[
+            (&[], 9),
+            (&[change], 7),
+            (&[read], 7),
+            (&[balance], 7),
+            (&[change, read], 5),
+            (&[change, balance], 5),
+            (&[read, balance], 5),
+            (&[change, read, balance], 3),
+        ];
+        for &(failures, expected) in cases {
+            assert_eq!(score_with_failures(failures), expected, "{failures:?}");
+        }
+    }
+
+    #[test]
+    fn concurrent_requests_are_scored_as_one_group() {
+        let requests = [
+            "NODE ADDED DURING INSERTS",
+            "NODE ADDED DURING DELETIONS",
+            "NODE ADDED DURING READS",
+        ];
+        for name in requests {
+            assert_eq!(score_with_failures(&[name]), 7, "{name}");
+        }
+        assert_eq!(score_with_failures(&requests), 7);
+    }
+}

homework/07-kv-sharding/tests/src/python_seed.rs

@@ -0,0 +1,33 @@
+//! Set Python's hash seed before the interpreter (including the validator) starts.
+use std::env;
+use std::process::Command;
+
+pub fn configure(seed: u64) {
+    // Python accepts only 32-bit hash seeds; the simulator keeps the full u64 seed.
+    let hash_seed = (seed as u32).to_string();
+    if env::var("PYTHONHASHSEED").as_deref() == Ok(hash_seed.as_str()) {
+        return;
+    }
+
+    // Restart with an inherited environment instead of changing it at runtime:
+    // on Windows, Rust and CPython can use different copies of the environment.
+    let mut command = Command::new(env::current_exe().expect("cannot locate test executable"));
+    command
+        .args(env::args_os().skip(1))
+        .env("PYTHONHASHSEED", hash_seed);
+
+    // Preserve the PID so timeout/signals still reach the test process directly.
+    #[cfg(unix)]
+    {
+        use std::os::unix::process::CommandExt;
+        let error = command.exec();
+        panic!("cannot restart tests with PYTHONHASHSEED: {error}");
+    }
+    #[cfg(not(unix))]
+    {
+        let status = command
+            .status()
+            .expect("cannot restart tests with PYTHONHASHSEED");
+        std::process::exit(status.code().unwrap_or(1));
+    }
+}

homework/07-kv-sharding/tests/src/shared_state.rs

@@ -0,0 +1,78 @@
+use std::ffi::CString;
+
+use pyo3::prelude::*;
+use pyo3::types::{PyList, PyModule};
+
+const VALIDATOR_CODE: &str = include_str!("../shared_state_validator.py");
+
+fn validator_module<'py>(py: Python<'py>) -> PyResult<Bound<'py, PyModule>> {
+    let code = CString::new(VALIDATOR_CODE).expect("validator source contains a null byte");
+    PyModule::from_code(
+        py,
+        &code,
+        c"shared_state_validator.py",
+        c"shared_state_validator",
+    )
+}
+
+#[cfg(test)]
+fn validate_source(
+    source: &str,
+    filename: &str,
+    class_names: &[&str],
+) -> Result<Vec<String>, String> {
+    Python::attach(|py| {
+        let module = validator_module(py).map_err(|error| error.to_string())?;
+        let class_names = PyList::new(py, class_names).map_err(|error| error.to_string())?;
+        module
+            .getattr("validate_source")
+            .and_then(|function| function.call1((source, filename, class_names)))
+            .and_then(|result| result.extract::<Vec<String>>())
+            .map_err(|error| error.to_string())
+    })
+}
+
+fn validate_solution(path: &str, class_names: &[&str]) -> Result<Vec<String>, String> {
+    Python::attach(|py| {
+        let module = validator_module(py).map_err(|error| error.to_string())?;
+        let class_names = PyList::new(py, class_names).map_err(|error| error.to_string())?;
+        module
+            .getattr("validate_solution")
+            .and_then(|function| function.call1((path, class_names)))
+            .and_then(|result| result.extract::<Vec<String>>())
+            .map_err(|error| error.to_string())
+    })
+}
+
+pub fn validate_or_report(path: &str, class_names: &[&str]) -> bool {
+    match validate_solution(path, class_names) {
+        Ok(violations) if violations.is_empty() => true,
+        Ok(violations) => {
+            println!("ERROR: shared state between AnySystem processes is forbidden");
+            for violation in violations {
+                println!("{violation}");
+            }
+            println!("\nSCORE: 0\n");
+            false
+        }
+        Err(error) => panic!("failed to validate solution for shared state: {error}"),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::validate_source;
+
+    #[test]
+    fn embeds_validator_and_rejects_shared_state() {
+        let violations = validate_source(
+            "class ProcessImpl:\n    shared = {}\n",
+            "solution.py",
+            &["ProcessImpl"],
+        )
+        .unwrap();
+
+        assert_eq!(violations.len(), 1);
+        assert!(violations[0].contains("ProcessImpl.shared"));
+    }
+}

homework/07-kv-sharding/tests/src/tests_mc.rs

@@ -0,0 +1,324 @@
+use std::collections::HashSet;
+
+use indexmap::IndexMap;
+use rand::prelude::SliceRandom;
+use rand::SeedableRng;
+use rand_pcg::Pcg64;
+use sugars::boxed;
+
+use anysystem::logger::LogEntry;
+use anysystem::mc::{
+    predicates::{collects, goals, invariants},
+    strategies::Bfs,
+    InvariantFn, McState, ModelChecker, StrategyConfig,
+};
+use anysystem::test::TestResult;
+use anysystem::Message;
+
+use crate::common::*;
+
+pub fn test_mc_normal(config: &TestConfig) -> TestResult {
+    let sys = build_system(config, false);
+    let mut mc = ModelChecker::new(&sys);
+    let mut rand = Pcg64::seed_from_u64(config.seed);
+
+    let key = random_string(8, &mut rand).to_uppercase();
+    let value = random_string(8, &mut rand);
+    let max_steps = 10;
+
+    let mut proc = random_proc(&sys, &mut rand);
+    let achieved_states = check_mc_get(&mut mc, &proc, &key, None, max_steps, None)?;
+    proc = random_proc(&sys, &mut rand);
+    let achieved_states = check_mc_put(
+        &mut mc,
+        &proc,
+        &key,
+        &value,
+        max_steps,
+        Some(achieved_states),
+    )?;
+    proc = random_proc(&sys, &mut rand);
+    let achieved_states = check_mc_get(
+        &mut mc,
+        &proc,
+        &key,
+        Some(&value),
+        max_steps,
+        Some(achieved_states),
+    )?;
+    proc = random_proc(&sys, &mut rand);
+    let achieved_states = check_mc_delete(
+        &mut mc,
+        &proc,
+        &key,
+        Some(&value),
+        max_steps,
+        Some(achieved_states),
+    )?;
+    proc = random_proc(&sys, &mut rand);
+    let achieved_states =
+        check_mc_get(&mut mc, &proc, &key, None, max_steps, Some(achieved_states))?;
+    proc = random_proc(&sys, &mut rand);
+    check_mc_delete(&mut mc, &proc, &key, None, max_steps, Some(achieved_states))?;
+    Ok(true)
+}
+
+pub fn test_mc_node_removed(config: &TestConfig) -> TestResult {
+    let sys = build_system(config, false);
+    let mut mc = ModelChecker::new(&sys);
+    let mut rand = Pcg64::seed_from_u64(config.seed);
+
+    // insert random key-value pairs
+    let keys_count = 40;
+    let mut achieved_states = None;
+    let mut kv = IndexMap::new();
+    for _ in 0..keys_count {
+        let k = random_string(8, &mut rand).to_uppercase();
+        let v = random_string(8, &mut rand);
+        let proc = random_proc(&sys, &mut rand);
+        let res = check_mc_put(&mut mc, &proc, &k, &v, 10, achieved_states)?;
+        achieved_states = Some(res);
+        kv.insert(k, v);
+    }
+
+    // remove a node from the system
+    let removed_proc = random_proc(&sys, &mut rand);
+    achieved_states = Some(check_mc_node_removed(
+        &mut mc,
+        sys.process_names(),
+        &removed_proc,
+        15,
+        achieved_states.unwrap_or_else(|| panic!("no states found after {keys_count} GET queries")),
+    )?);
+    let alive_proc = sys
+        .process_names()
+        .into_iter()
+        .filter(|proc| *proc != removed_proc)
+        .collect::<Vec<String>>();
+
+    // check that all data is still in the storage
+    for (k, v) in kv {
+        let proc = alive_proc.choose(&mut rand).unwrap().clone();
+        achieved_states = Some(check_mc_get(
+            &mut mc,
+            &proc,
+            &k,
+            Some(&v),
+            10,
+            achieved_states,
+        )?);
+    }
+    Ok(true)
+}
+
+// UTILS ---------------------------------------------------------------------------------------------------------------
+
+fn check_mc_get<S>(
+    mc: &mut ModelChecker,
+    proc: S,
+    key: S,
+    expected: Option<S>,
+    max_steps: u64,
+    start_states: Option<HashSet<McState>>,
+) -> Result<HashSet<McState>, String>
+where
+    S: Into<String> + Clone,
+{
+    let proc = proc.into();
+    let key = key.into();
+    let expected = expected.map(|s| s.into());
+    let msg = Message::new("GET", &format!(r#"{{"key": "{key}"}}"#));
+    let proc_name = proc.clone();
+    let invariant = boxed!(move |state: &McState| {
+        for entry in state.current_run_trace().iter() {
+            if let LogEntry::McLocalMessageSent {
+                msg: message,
+                proc: proc_tmp,
+            } = entry
+            {
+                if *proc_tmp != proc {
+                    return Err("local message received on wrong process".to_string());
+                }
+                if message.tip != "GET_RESP" {
+                    return Err(format!("wrong type {}", message.tip));
+                }
+                let data: GetRespMessage =
+                    serde_json::from_str(&message.data).map_err(|err| err.to_string())?;
+                if data.key != key {
+                    return Err(format!("wrong key {}", data.key));
+                }
+                if data.value.map(|s| s.to_string()) != expected {
+                    return Err(format!("wrong value {:?}", data.value));
+                }
+            }
+        }
+        Ok(())
+    });
+    mc_check_query(mc, proc_name, invariant, msg, start_states, max_steps)
+}
+
+fn check_mc_put<S>(
+    mc: &mut ModelChecker,
+    proc: S,
+    key: S,
+    value: S,
+    max_steps: u64,
+    start_states: Option<HashSet<McState>>,
+) -> Result<HashSet<McState>, String>
+where
+    S: Into<String>,
+{
+    let proc = proc.into();
+    let key = key.into();
+    let value = value.into();
+    let proc_name = proc.clone();
+    let msg = Message::new("PUT", &format!(r#"{{"key": "{key}", "value": "{value}"}}"#));
+    let invariant = Box::new(move |state: &McState| {
+        for entry in state.current_run_trace().iter() {
+            if let LogEntry::McLocalMessageSent {
+                msg: message,
+                proc: proc_tmp,
+            } = entry
+            {
+                if *proc_tmp != proc {
+                    return Err("local message received on wrong process".to_string());
+                }
+                if message.tip != "PUT_RESP" {
+                    return Err(format!("wrong type {}", message.tip));
+                }
+                let data: PutRespMessage =
+                    serde_json::from_str(&message.data).map_err(|err| err.to_string())?;
+                if data.key != key {
+                    return Err(format!("wrong key {}", data.key));
+                }
+                if data.value != value {
+                    return Err(format!("wrong value {:?}", data.value));
+                }
+            }
+        }
+        Ok(())
+    });
+    mc_check_query(mc, proc_name, invariant, msg, start_states, max_steps)
+}
+
+fn check_mc_delete<S>(
+    mc: &mut ModelChecker,
+    proc: S,
+    key: S,
+    expected: Option<S>,
+    max_steps: u64,
+    start_states: Option<HashSet<McState>>,
+) -> Result<HashSet<McState>, String>
+where
+    S: Into<String> + Clone,
+{
+    let proc = proc.into();
+    let key = key.into();
+    let expected = expected.map(|s| s.into());
+    let msg = Message::new("DELETE", &format!(r#"{{"key": "{key}"}}"#));
+    let proc_name = proc.clone();
+    let invariant = Box::new(move |state: &McState| {
+        for entry in state.current_run_trace().iter() {
+            if let LogEntry::McLocalMessageSent {
+                msg: message,
+                proc: proc_tmp,
+            } = entry
+            {
+                if *proc_tmp != proc {
+                    return Err("local message received on wrong process".to_string());
+                }
+                if message.tip != "DELETE_RESP" {
+                    return Err(format!("wrong type {}", message.tip));
+                }
+                let data: DeleteRespMessage =
+                    serde_json::from_str(&message.data).map_err(|err| err.to_string())?;
+                if data.key != key {
+                    return Err(format!("wrong key {}", data.key));
+                }
+                if data.value.map(|s| s.to_string()) != expected {
+                    return Err(format!("wrong value {:?}", data.value));
+                }
+            }
+        }
+        Ok(())
+    });
+    mc_check_query(mc, proc_name, invariant, msg, start_states, max_steps)
+}
+
+fn check_mc_node_removed(
+    mc: &mut ModelChecker,
+    process_names: Vec<String>,
+    removed_proc: &str,
+    max_steps: u64,
+    start_states: HashSet<McState>,
+) -> Result<HashSet<McState>, String> {
+    let strategy_config = StrategyConfig::default()
+        .invariant(invariants::state_depth_current_run(max_steps))
+        .goal(goals::no_events())
+        .collect(collects::no_events());
+
+    let res = mc.run_from_states_with_change::<Bfs>(strategy_config, start_states, move |sys| {
+        let msg = Message::new("NODE_REMOVED", &format!(r#"{{"id": "{removed_proc}"}}"#));
+        for proc in process_names.clone() {
+            sys.send_local_message(proc.clone(), proc, msg.clone());
+        }
+    });
+    match res {
+        Ok(stats) => Ok(stats
+            .collected_states
+            .into_iter()
+            .map(|mut state| {
+                state.network.disconnect_node(removed_proc);
+                state
+            })
+            .collect::<HashSet<McState>>()),
+        Err(err) => {
+            err.print_trace();
+            Err(err.message())
+        }
+    }
+}
+
+fn mc_check_query<S>(
+    mc: &mut ModelChecker,
+    proc: S,
+    invariant: InvariantFn,
+    msg: Message,
+    start_states: Option<HashSet<McState>>,
+    max_depth: u64,
+) -> Result<HashSet<McState>, String>
+where
+    S: Into<String>,
+{
+    let proc = proc.into();
+    let strategy_config = StrategyConfig::default()
+        .goal(goals::all_goals(vec![
+            goals::event_happened_n_times_current_run(LogEntry::is_mc_local_message_sent, 1),
+            goals::no_events(),
+        ]))
+        .invariant(invariants::all_invariants(vec![
+            invariant,
+            invariants::state_depth_current_run(max_depth),
+        ]))
+        .collect(collects::event_happened_n_times_current_run(
+            LogEntry::is_mc_local_message_sent,
+            1,
+        ));
+
+    let res = if let Some(start_states) = start_states {
+        mc.run_from_states_with_change::<Bfs>(strategy_config, start_states, |sys| {
+            sys.send_local_message(proc.clone(), proc.clone(), msg.clone());
+        })
+    } else {
+        mc.run_with_change::<Bfs>(strategy_config, |sys| {
+            sys.send_local_message(proc.clone(), proc.clone(), msg);
+        })
+    };
+    match res {
+        Ok(stats) => Ok(stats.collected_states),
+        Err(err) => {
+            err.print_trace();
+            Err(err.message())
+        }
+    }
+}