Fix CI campaign counting and lifecycle races

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]) == []
 

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, [

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

test/jepsen/src/group/jepsen/db.clj

@@ -6,17 +6,19 @@
 (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"])]
       (when-not (= :ok (:status response))
         (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]

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")))))

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"]])))

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