From a3adbd2fefd77af96cb1f0d651b20ec988b4a477 Mon Sep 17 00:00:00 2001 From: Jason Stiebs Date: Mon, 21 Sep 2026 15:19:26 -0500 Subject: [PATCH] Fix CI campaign counting and lifecycle synchronization --- test/anti_entropy_fault_regression_test.exs | 4 + test/distributed_test.exs | 33 +++---- test/jepsen/campaign.sh | 4 +- test/jepsen/src/group/jepsen/db.clj | 8 +- test/jepsen/test/group/jepsen/db_test.clj | 52 ++++++++++++ .../group/jepsen/retired_evidence_test.clj | 5 +- test/jepsen_campaign_test.exs | 85 +++++++++++++++++++ 7 files changed, 169 insertions(+), 22 deletions(-) create mode 100644 test/jepsen/test/group/jepsen/db_test.clj create mode 100644 test/jepsen_campaign_test.exs diff --git a/test/anti_entropy_fault_regression_test.exs b/test/anti_entropy_fault_regression_test.exs index d41ca2b..8d5fc58 100644 --- a/test/anti_entropy_fault_regression_test.exs +++ b/test/anti_entropy_fault_regression_test.exs @@ -3385,6 +3385,10 @@ defmodule Group.AntiEntropyFaultRegressionTest do end end) + # Name registration precedes init/1. Wait for startup repair to finish + # before reading ETS directly; the transport remains in :drop mode. + TestCluster.flush_shards(context.node_b, name) + assert TestCluster.rpc!(context.node_b, Group, :lookup, [name, registry_key]) == nil assert TestCluster.rpc!(context.node_b, Group, :members, [name, pg_key]) == [] diff --git a/test/distributed_test.exs b/test/distributed_test.exs index 34c7a5e..921509e 100644 --- a/test/distributed_test.exs +++ b/test/distributed_test.exs @@ -1140,23 +1140,24 @@ defmodule Group.DistributedTest do node_b in TestCluster.rpc!(node_a, Group, :nodes, [name, cluster]) end) - assert TestCluster.rpc!(node_a, Group, :lookup, [ - name, - registry_key, - [cluster: cluster] - ]) == nil - - assert TestCluster.rpc!(node_a, Group, :members, [name, pg_key, [cluster: cluster]]) == [] - + # Routing readiness is not a recovery barrier: replay can expose an old + # join before its leave. Require both removals and retained rows in the + # same converged observation, not immediately after route discovery. TestCluster.assert_eventually(fn -> - match?( - {^retained_registry_pid, %{from: :b, retained: true}}, - TestCluster.rpc!(node_a, Group, :lookup, [ - name, - retained_registry_key, - [cluster: cluster] - ]) - ) and + TestCluster.rpc!(node_a, Group, :lookup, [ + name, + registry_key, + [cluster: cluster] + ]) == nil and + TestCluster.rpc!(node_a, Group, :members, [name, pg_key, [cluster: cluster]]) == [] and + match?( + {^retained_registry_pid, %{from: :b, retained: true}}, + TestCluster.rpc!(node_a, Group, :lookup, [ + name, + retained_registry_key, + [cluster: cluster] + ]) + ) and match?( [{^retained_pg_pid, %{from: :b, retained: true}}], TestCluster.rpc!(node_a, Group, :members, [ diff --git a/test/jepsen/campaign.sh b/test/jepsen/campaign.sh index 4e0a12c..8b9729a 100755 --- a/test/jepsen/campaign.sh +++ b/test/jepsen/campaign.sh @@ -80,7 +80,9 @@ for transport in "${transports[@]}"; do --min-delta-run-records "${min_delta_run_records}" \ --transport "${transport}" \ --scenario "${scenario}" >"${log}" 2>&1; then - valid_count="$(rg -c "Everything looks good" "${log}" || true)" + # Use a standard runner utility, and let read/command errors fail the + # campaign instead of silently turning them into zero valid histories. + valid_count="$(awk '/^Everything looks good/ { count++ } END { print count + 0 }' "${log}")" if [[ "${valid_count}" != "${test_count}" ]]; then echo "Failed ${transport}/${scenario}: expected ${test_count} valid histories, found ${valid_count:-0}" >&2 diff --git a/test/jepsen/src/group/jepsen/db.clj b/test/jepsen/src/group/jepsen/db.clj index 8898077..ea41c7f 100644 --- a/test/jepsen/src/group/jepsen/db.clj +++ b/test/jepsen/src/group/jepsen/db.clj @@ -6,8 +6,10 @@ (defrecord DockerDB [] db/DB (setup! [_this test node] - (docker/heal! (:nodes test)) (docker/restart! node) + ;; Jepsen sets up nodes concurrently. Never exec firewall commands on a + ;; sibling while its setup thread may be restarting its container. + (docker/heal! [node]) (docker/reset-oracle! node) (group-client/wait-listening! node) (let [response (group-client/request! node ["reset-conflict-evidence"])] @@ -15,8 +17,8 @@ (throw (ex-info "conflict oracle reset failed" {:node node :response response})))) (group-client/wait-ready! node (count (:nodes test)))) - (teardown! [_this test _node] - (docker/heal! (:nodes test))) + (teardown! [_this _test node] + (docker/heal! [node])) db/Kill (kill! [_this _test node] diff --git a/test/jepsen/test/group/jepsen/db_test.clj b/test/jepsen/test/group/jepsen/db_test.clj new file mode 100644 index 0000000..1794b4b --- /dev/null +++ b/test/jepsen/test/group/jepsen/db_test.clj @@ -0,0 +1,52 @@ +(ns group.jepsen.db-test + (:require [clojure.test :refer :all] + [group.jepsen.client :as client] + [group.jepsen.db :as group-db] + [group.jepsen.docker :as docker] + [jepsen.db :as db]) + (:import (java.util.concurrent CountDownLatch TimeUnit))) + +(deftest concurrent-setup-heals-only-its-own-restarted-container + (let [nodes ["n1" "n2" "n3"] + test {:nodes nodes} + database (group-db/db) + restarting (CountDownLatch. 3) + running (atom #{}) + healed (atom [])] + (with-redefs [docker/restart! + (fn [node] + (.countDown restarting) + (when-not (.await restarting 5 TimeUnit/SECONDS) + (throw (ex-info "setups did not restart concurrently" {}))) + (swap! running conj node)) + docker/heal! + (fn [targets] + (is (= 1 (count targets))) + (is (every? @running targets)) + (swap! healed into targets)) + docker/reset-oracle! (fn [_]) + client/wait-listening! (fn [_]) + client/request! (fn [_ _] {:status :ok}) + client/wait-ready! (fn [_ size] (is (= 3 size)))] + (let [setups (mapv #(future (db/setup! database test %)) nodes)] + (try + (doseq [setup setups] + (is (not= ::timeout (deref setup 10000 ::timeout)))) + (is (= (sort nodes) (sort @healed))) + (finally + (doseq [setup setups] + (future-cancel setup)))))))) + +(deftest teardown-heals-each-container-once + (let [nodes ["n1" "n2" "n3"] + calls (atom [])] + (with-redefs [docker/heal! #(swap! calls conj %)] + (doseq [node nodes] + (db/teardown! (group-db/db) {:nodes nodes} node)) + (is (= [["n1"] ["n2"] ["n3"]] @calls))))) + +(deftest setup-does-not-hide-firewall-failures + (with-redefs [docker/restart! (fn [_]) + docker/heal! (fn [_] (throw (ex-info "firewall failed" {:exit 1})))] + (is (thrown-with-msg? Exception #"firewall failed" + (db/setup! (group-db/db) {:nodes ["n1"]} "n1"))))) diff --git a/test/jepsen/test/group/jepsen/retired_evidence_test.clj b/test/jepsen/test/group/jepsen/retired_evidence_test.clj index 5f738c5..2404233 100644 --- a/test/jepsen/test/group/jepsen/retired_evidence_test.clj +++ b/test/jepsen/test/group/jepsen/retired_evidence_test.clj @@ -40,7 +40,7 @@ (deftest workload-reset-clears-the-running-conflict-recorder (let [calls (atom [])] - (with-redefs [docker/heal! (fn [_]) + (with-redefs [docker/heal! #(swap! calls conj [:heal %]) docker/restart! #(swap! calls conj [:restart %]) docker/reset-oracle! #(swap! calls conj [:disk-reset %]) client/wait-listening! #(swap! calls conj [:listening %]) @@ -50,7 +50,8 @@ client/wait-ready! (fn [node _] (swap! calls conj [:ready node]))] (dotimes [_ 2] (db/setup! (group-db/db) {:nodes ["n1"]} "n1")) - (is (= (vec (mapcat identity (repeat 2 [[:restart "n1"] [:disk-reset "n1"] + (is (= (vec (mapcat identity (repeat 2 [[:restart "n1"] [:heal ["n1"]] + [:disk-reset "n1"] [:listening "n1"] ["n1" ["reset-conflict-evidence"]] [:ready "n1"]]))) diff --git a/test/jepsen_campaign_test.exs b/test/jepsen_campaign_test.exs new file mode 100644 index 0000000..784544b --- /dev/null +++ b/test/jepsen_campaign_test.exs @@ -0,0 +1,85 @@ +defmodule Group.JepsenCampaignTest do + use ExUnit.Case, async: true + + @moduletag :local + @moduletag :tmp_dir + + setup %{tmp_dir: directory} do + script_dir = Path.join(directory, "checkout/test/jepsen") + bin = Path.join(directory, "bin") + File.mkdir_p!(script_dir) + File.mkdir_p!(bin) + File.cp!("test/jepsen/campaign.sh", Path.join(script_dir, "campaign.sh")) + + # Match a minimal CI runner: no ripgrep and no real Docker/campaign work. + for command <- ~w(bash dirname mkdir mktemp awk tail) do + File.ln_s!(System.find_executable(command), Path.join(bin, command)) + end + + write_executable(Path.join(bin, "timeout"), """ + #!/bin/bash + shift 3 + exec "$@" + """) + + write_executable(Path.join(script_dir, "run.sh"), """ + #!/bin/bash + printf '%s\\n' "$CAMPAIGN_OUTPUT" + exit "$CAMPAIGN_STATUS" + """) + + {:ok, script: Path.join(script_dir, "campaign.sh"), bin: bin} + end + + test "counts every successful history without ripgrep", context do + {output, status} = run_campaign(context, 2) + assert status == 0, output + assert output =~ "Passed distribution/mixed: 2/2 valid histories" + end + + for count <- [0, 1, 3] do + @tag count: count + test "rejects #{count} successful histories when two are required", context do + {output, status} = run_campaign(context, context.count) + assert status == 1, output + assert output =~ "expected 2 valid histories, found #{context.count}" + end + end + + test "preserves a failed campaign status even after successful histories", context do + {output, status} = run_campaign(context, 2, 42) + assert status == 42, output + refute output =~ "Jepsen campaign passed" + end + + test "does not disguise a counter failure as missing histories", context do + awk = Path.join(context.bin, "awk") + File.rm!(awk) + write_executable(awk, "#!/bin/bash\nexit 42\n") + + {output, status} = run_campaign(context, 2) + assert status == 42, output + refute output =~ "valid histories, found" + end + + defp run_campaign(context, count, status \\ 0) do + System.cmd(System.find_executable("bash"), [context.script], + env: [ + {"PATH", context.bin}, + {"GROUP_JEPSEN_SKIP_CHECKER", "1"}, + {"GROUP_JEPSEN_CAMPAIGN_TRANSPORT", "distribution"}, + {"GROUP_JEPSEN_CAMPAIGN_SCENARIO", "mixed"}, + {"GROUP_JEPSEN_CAMPAIGN_COUNT", "2"}, + {"GROUP_JEPSEN_CAMPAIGN_ARTIFACT_DIR", ""}, + {"CAMPAIGN_OUTPUT", String.duplicate("Everything looks good!\n", count)}, + {"CAMPAIGN_STATUS", Integer.to_string(status)} + ], + stderr_to_stdout: true + ) + end + + defp write_executable(path, contents) do + File.write!(path, contents) + File.chmod!(path, 0o755) + end +end