From b48d4a66e7bb03a19235eac74d6e8dfd535b087c Mon Sep 17 00:00:00 2001 From: Jason Stiebs Date: Wed, 9 Sep 2026 14:23:06 -0500 Subject: [PATCH 1/2] Add Elixir and OTP CI matrix and stabilize distributed tests --- .github/workflows/ci.yml | 62 ++++++++++++++++++++++++++++++++++++ README.md | 18 +++++++++++ test/README.md | 27 ++++++++++++++-- test/distributed_test.exs | 38 +++++++++++----------- test/support/test_cluster.ex | 37 ++++++++++++++++++--- test/test_helper.exs | 4 ++- 6 files changed, 159 insertions(+), 27 deletions(-) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..fd25d61 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,62 @@ +name: CI + +on: + push: + pull_request: + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + test: + name: ${{ matrix.latest && 'Latest stable Elixir / OTP' || format('Elixir {0} / OTP {1}', matrix.elixir, matrix.otp) }} + runs-on: ubuntu-24.04 + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + include: + - elixir: "1.19" + otp: "26" + - elixir: "1.19" + otp: "27" + - elixir: "1.19" + otp: "28" + - elixir: "1.20" + otp: "27" + - elixir: "1.20" + otp: "28" + # Unlike "latest", these ranges exclude prereleases. + - elixir: "> 0" + otp: "> 0" + latest: true + env: + MIX_ENV: test + # Inherited by :peer nodes; bound scheduler usage for distributed tests. + ERL_FLAGS: "+S 4:4" + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1 + with: + elixir-version: ${{ matrix.elixir }} + otp-version: ${{ matrix.otp }} + + - name: Install dependencies + run: mix deps.get + + - name: Check formatting + if: matrix.latest + run: mix format --check-formatted + + - name: Compile with warnings as errors + run: mix compile --warnings-as-errors + + - name: Run local and distributed tests + run: mix test --warnings-as-errors diff --git a/README.md b/README.md index c3d8fce..e605916 100644 --- a/README.md +++ b/README.md @@ -400,6 +400,24 @@ mix test See [`test/README.md`](test/README.md) for details on the distributed test infrastructure. +GitHub Actions runs the full local and distributed suite on pushes and pull +requests using Elixir 1.19 / OTP 26–28, Elixir 1.20 / OTP 27–28, and the latest +stable Elixir / OTP pair. Version ranges pick up new patch releases automatically; +the latest-stable job also picks up new minor and major releases, excluding +prereleases. Every job treats compilation and test warnings as errors, and the +latest-stable job checks formatting. + +To run the same checks locally: + +```bash +export MIX_ENV=test +export ERL_FLAGS="+S 4:4" +mix deps.get +mix format --check-formatted +mix compile --warnings-as-errors +mix test --warnings-as-errors +``` + ## Benchmarks ```bash diff --git a/test/README.md b/test/README.md index 58a6d51..b0aeba4 100644 --- a/test/README.md +++ b/test/README.md @@ -104,8 +104,10 @@ TestCluster.spawn_register_update_unregister(node_a, :test, "user/1", %{v: 1}, % `spawn_register` accepts `flush_shards: num_shards` which calls `:sys.get_state` on the target shard's GenServer after registration. This -blocks until all pending messages (nodedown, replicate, etc.) are processed -on that shard — useful in partition tests where you need to guarantee ordering. +synchronizes with that shard after the write; it is not a pre-write barrier +and does not flush buffered replication. Use `TestCluster.flush_shards/2` to +flush replication and `TestCluster.assert_group_nodes/3` to wait for peer +discovery or nodedown cleanup on every shard. ```elixir TestCluster.spawn_register(node_a, :test, "key", %{}, flush_shards: 4) @@ -136,6 +138,9 @@ TestCluster.monitor_nodes_on(node_a, self()) assert_receive {:nodedown_on_remote, ^node_b}, 5000 ``` +`monitor_nodes_on/2` returns only after the remote monitor is installed. +Receiving its notification does not mean Group's shards have processed nodedown. + #### Network partitions ```elixir @@ -148,6 +153,24 @@ partitions don't work reliably because the test node bridges them — Erlang distribution is fully meshed, so if the test node can reach both peers, they can reach each other through it. +Before disconnecting, wait for discovery on **every shard**, not just +`Node.list/0` connectivity. In-flight discovery sends can otherwise trigger +reconnect retries. After disconnecting, wait for every shard's peer state +before writing partition-local data: + +```elixir +TestCluster.assert_group_nodes(node_a, :test, [node_b, node_c]) +TestCluster.assert_group_nodes(node_b, :test, [node_a, node_c]) +TestCluster.assert_group_nodes(node_c, :test, [node_a, node_b]) + +TestCluster.disconnect_nodes(node_c, node_a) +TestCluster.disconnect_nodes(node_c, node_b) + +TestCluster.assert_group_nodes(node_a, :test, [node_b]) +TestCluster.assert_group_nodes(node_b, :test, [node_a]) +TestCluster.assert_group_nodes(node_c, :test, []) +``` + #### Polling for eventual consistency ```elixir diff --git a/test/distributed_test.exs b/test/distributed_test.exs index e62a4d9..dcb9e9c 100644 --- a/test/distributed_test.exs +++ b/test/distributed_test.exs @@ -574,14 +574,11 @@ defmodule Group.DistributedTest do start_group_on_peers(peers, opts) - # Wait for Erlang-level connectivity so disconnect_nodes actually works - TestCluster.assert_eventually( - fn -> - c_nodes = TestCluster.rpc!(node_c, Node, :list, []) - node_a in c_nodes and node_b in c_nodes - end, - timeout: 5000 - ) + # Finish discovery on every shard before cutting links. Erlang connectivity + # alone can leave handshake sends in flight that trigger reconnect retries. + TestCluster.assert_group_nodes(node_a, name, [node_b, node_c]) + TestCluster.assert_group_nodes(node_b, name, [node_a, node_c]) + TestCluster.assert_group_nodes(node_c, name, [node_a, node_b]) # Set up nodedown monitor on A TestCluster.monitor_nodes_on(node_a, self()) @@ -593,8 +590,12 @@ defmodule Group.DistributedTest do # Wait for A to confirm it saw C go down assert_receive {:nodedown_on_remote, ^node_c}, 5000 + # A separate node monitor does not establish that every shard handled DOWN. + TestCluster.assert_group_nodes(node_a, name, [node_b]) + TestCluster.assert_group_nodes(node_b, name, [node_a]) + TestCluster.assert_group_nodes(node_c, name, []) + # While partitioned: register keys on A, join groups on C - # flush_shards ensures nodedown is processed before registering TestCluster.spawn_register(node_a, name, "user/from_a", %{origin: :a}, flush_shards: 2) TestCluster.spawn_join(node_c, name, "room/from_c", %{origin: :c}) @@ -1970,14 +1971,10 @@ defmodule Group.DistributedTest do length(nodes) >= 1 end) - # Wait for Erlang-level connectivity so disconnect_nodes actually works - TestCluster.assert_eventually( - fn -> - c_nodes = TestCluster.rpc!(node_c, Node, :list, []) - node_a in c_nodes and node_b in c_nodes - end, - timeout: 5000 - ) + # Wait for Group discovery, not just Erlang connectivity, before partitioning. + TestCluster.assert_group_nodes(node_a, name, [node_b, node_c]) + TestCluster.assert_group_nodes(node_b, name, [node_a, node_c]) + TestCluster.assert_group_nodes(node_c, name, [node_a, node_b]) # Set up nodedown monitors on A before partitioning TestCluster.monitor_nodes_on(node_a, self()) @@ -1990,9 +1987,12 @@ defmodule Group.DistributedTest do assert_receive {:nodedown_on_remote, ^node_c}, 5000 # Wait for Group's own peer tables to reflect the partition before writing. + TestCluster.assert_group_nodes(node_a, name, [node_b]) + TestCluster.assert_group_nodes(node_b, name, [node_a]) + TestCluster.assert_group_nodes(node_c, name, []) + TestCluster.assert_eventually(fn -> - node_c not in TestCluster.rpc!(node_a, Group, :nodes, [name]) and - node_c not in TestCluster.rpc!(node_a, Group, :nodes, [name, "game"]) + node_c not in TestCluster.rpc!(node_a, Group, :nodes, [name, "game"]) end) # Register data during partition diff --git a/test/support/test_cluster.ex b/test/support/test_cluster.ex index f9b8c1e..c1014a5 100644 --- a/test/support/test_cluster.ex +++ b/test/support/test_cluster.ex @@ -11,7 +11,7 @@ defmodule Group.TestCluster do Enum.flat_map(code_paths, fn p -> [~c"-pa", p] end) for _i <- 1..count do - name = :"peer#{System.unique_integer([:positive])}" + name = :"peer_#{System.pid()}_#{System.unique_integer([:positive])}" {:ok, pid, node} = :peer.start(%{name: name, host: ~c"127.0.0.1", longnames: true, args: args}) @@ -302,10 +302,20 @@ defmodule Group.TestCluster do @doc "Monitor nodedown events from a remote node, forwarding to caller" def monitor_nodes_on(node, target_pid) do :erpc.call(node, fn -> - spawn(fn -> - :net_kernel.monitor_nodes(true) - forward_nodedown(target_pid) - end) + parent = self() + + pid = + spawn(fn -> + :net_kernel.monitor_nodes(true) + send(parent, {:monitor_ready, self()}) + forward_nodedown(target_pid) + end) + + receive do + {:monitor_ready, ^pid} -> pid + after + 5000 -> raise "monitor_nodes_on timed out" + end end) end @@ -397,6 +407,23 @@ defmodule Group.TestCluster do end) end + @doc "Wait for every shard and the shared peer table to agree on the expected Group peers." + def assert_group_nodes(node, name, expected_nodes) do + expected_nodes = Enum.sort(expected_nodes) + + assert_eventually(fn -> + :erpc.call(node, fn -> + num_shards = Group.get_config(name).num_shards + + Enum.sort(Group.nodes(name)) == expected_nodes and + Enum.all?(0..(num_shards - 1), fn shard -> + state = :sys.get_state(:"#{name}_replica_#{shard}") + Enum.sort(Map.keys(state.remote_shards)) == expected_nodes + end) + end) + end) + end + @doc "Returns the current message_queue_len for a shard on a remote node." def shard_message_queue_len(node, name, shard) do :erpc.call(node, __MODULE__, :do_shard_message_queue_len, [name, shard]) diff --git a/test/test_helper.exs b/test/test_helper.exs index 15894c5..158a305 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -14,7 +14,9 @@ unless epmd_running?.() do end unless Node.alive?() do - {:ok, _} = Node.start(:"test_#{System.unique_integer([:positive])}@127.0.0.1", :longnames) + # unique_integer is VM-local; include the OS pid to allow concurrent test runs. + name = :"test_#{System.pid()}_#{System.unique_integer([:positive])}@127.0.0.1" + {:ok, _} = Node.start(name, :longnames) Node.set_cookie(:group_test) end From f7f549c54a0bf5ece859fa4147c583d0e863234a Mon Sep 17 00:00:00 2001 From: Jason Stiebs Date: Wed, 9 Sep 2026 15:14:18 -0500 Subject: [PATCH 2/2] Separate CI lanes and improve test concurrency and diagnostics --- .formatter.exs | 7 +- .github/workflows/campaigns.yml | 91 ++++++++++++ .github/workflows/ci.yml | 63 +++++++-- .github/workflows/performance.yml | 83 +++++++++++ README.md | 12 +- priv/bench/comparison.exs | 56 ++++++++ priv/ci/mutations.exs | 76 +++++++++++ test/README.md | 77 +++++++++-- test/conflict_test.exs | 23 ++++ test/diagnostics_test.exs | 61 +++++++++ test/distributed_test.exs | 21 +++ test/group_test.exs | 220 +++++++++--------------------- test/history_test.exs | 87 ++++++++++++ test/support/local_case.ex | 163 ++++++++++++++++++++++ test/support/test_cluster.ex | 49 ++++++- test/support/test_diagnostics.ex | 98 +++++++++++++ test/test_helper.exs | 28 +--- 17 files changed, 1014 insertions(+), 201 deletions(-) create mode 100644 .github/workflows/campaigns.yml create mode 100644 .github/workflows/performance.yml create mode 100644 priv/bench/comparison.exs create mode 100644 priv/ci/mutations.exs create mode 100644 test/conflict_test.exs create mode 100644 test/diagnostics_test.exs create mode 100644 test/history_test.exs create mode 100644 test/support/local_case.ex create mode 100644 test/support/test_diagnostics.ex diff --git a/.formatter.exs b/.formatter.exs index d2cda26..4b4f257 100644 --- a/.formatter.exs +++ b/.formatter.exs @@ -1,4 +1,9 @@ # Used by "mix format" [ - inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"] + inputs: [ + "{mix,.formatter}.exs", + "{config,lib,test}/**/*.{ex,exs}", + "priv/ci/*.exs", + "priv/bench/comparison.exs" + ] ] diff --git a/.github/workflows/campaigns.yml b/.github/workflows/campaigns.yml new file mode 100644 index 0000000..455200a --- /dev/null +++ b/.github/workflows/campaigns.yml @@ -0,0 +1,91 @@ +name: Fault campaigns + +on: + schedule: + - cron: "23 3 * * *" + workflow_dispatch: + release: + types: [published] + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: false + +defaults: + run: + shell: bash + +env: + MIX_ENV: test + ERL_FLAGS: "+S 4:4" + GROUP_HISTORY_STEPS: "2000" + GROUP_TEST_DIAGNOSTICS: ${{ github.workspace }}/_build/test-diagnostics + ERL_CRASH_DUMP: ${{ github.workspace }}/_build/test-diagnostics/erl_crash.dump + +jobs: + histories-and-faults: + name: Seed ${{ matrix.seed }} / ${{ matrix.schedulers }} peer schedulers + runs-on: ubuntu-24.04 + timeout-minutes: 30 + strategy: + fail-fast: false + matrix: + seed: [104729, 130363, 155921] + schedulers: ["1", "2", "4"] + env: + GROUP_PEER_SCHEDULERS: ${{ matrix.schedulers }} + CAMPAIGN_SEED: ${{ matrix.seed }} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1 + with: + elixir-version: "> 0" + otp-version: "> 0" + - run: mix deps.get + - name: Repeat histories, partitions, churn, restarts and race regressions + run: | + mkdir -p "$GROUP_TEST_DIAGNOSTICS" + mix compile --warnings-as-errors + for iteration in 0 1 2 3 4; do + seed=$((CAMPAIGN_SEED + iteration)) + mix test --warnings-as-errors --seed "$seed" 2>&1 | + tee "$GROUP_TEST_DIAGNOSTICS/run-$seed.log" + done + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + if: always() + with: + name: campaign-${{ matrix.seed }}-${{ matrix.schedulers }} + path: _build/test-diagnostics + retention-days: 14 + + mutations: + name: Contract mutations + runs-on: ubuntu-24.04 + timeout-minutes: 20 + env: + GROUP_HISTORY_STEPS: "100" + GROUP_PEER_SCHEDULERS: "2" + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1 + with: + elixir-version: "> 0" + otp-version: "> 0" + - run: mix deps.get + - name: Require behavioral regressions to be detected + run: elixir priv/ci/mutations.exs + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + if: always() + with: + name: mutation-reports + path: | + _build/mutations/**/*.log + _build/mutations/**/diagnostics + retention-days: 14 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fd25d61..9f47520 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -2,7 +2,11 @@ name: CI on: push: + branches: [main] pull_request: + release: + types: [published] + workflow_dispatch: permissions: contents: read @@ -11,9 +15,47 @@ concurrency: group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} cancel-in-progress: true +defaults: + run: + shell: bash + +env: + MIX_ENV: test + ERL_FLAGS: "+S 4:4" + GROUP_PEER_SCHEDULERS: "2" + GROUP_TEST_DIAGNOSTICS: ${{ github.workspace }}/_build/test-diagnostics + ERL_CRASH_DUMP: ${{ github.workspace }}/_build/test-diagnostics/erl_crash.dump + jobs: + local: + name: Local checks + runs-on: ubuntu-24.04 + timeout-minutes: 5 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1 + with: + elixir-version: "> 0" + otp-version: "> 0" + - run: mix deps.get + - name: Formatting, compiler checks, local examples and histories + run: | + mkdir -p "$GROUP_TEST_DIAGNOSTICS" + mix format --check-formatted 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/format.log" + mix compile --warnings-as-errors 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/compile.log" + mix test --only local --warnings-as-errors 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/local.log" + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + if: failure() + with: + name: local-failure + path: _build/test-diagnostics + retention-days: 14 + test: name: ${{ matrix.latest && 'Latest stable Elixir / OTP' || format('Elixir {0} / OTP {1}', matrix.elixir, matrix.otp) }} + needs: local runs-on: ubuntu-24.04 timeout-minutes: 15 strategy: @@ -34,10 +76,6 @@ jobs: - elixir: "> 0" otp: "> 0" latest: true - env: - MIX_ENV: test - # Inherited by :peer nodes; bound scheduler usage for distributed tests. - ERL_FLAGS: "+S 4:4" steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: @@ -51,12 +89,17 @@ jobs: - name: Install dependencies run: mix deps.get - - name: Check formatting - if: matrix.latest - run: mix format --check-formatted - - name: Compile with warnings as errors - run: mix compile --warnings-as-errors + run: | + mkdir -p "$GROUP_TEST_DIAGNOSTICS" + mix compile --warnings-as-errors 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/compile.log" - name: Run local and distributed tests - run: mix test --warnings-as-errors + run: mix test --warnings-as-errors 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/test.log" + + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + if: failure() + with: + name: runtime-failure-${{ strategy.job-index }} + path: _build/test-diagnostics + retention-days: 14 diff --git a/.github/workflows/performance.yml b/.github/workflows/performance.yml new file mode 100644 index 0000000..5b98d1e --- /dev/null +++ b/.github/workflows/performance.yml @@ -0,0 +1,83 @@ +name: Performance comparison + +on: + workflow_dispatch: + inputs: + baseline: + description: "Baseline ref (use a previous release tag for release review)" + type: string + default: main + required: true + release: + types: [published] + +permissions: + contents: read + +defaults: + run: + shell: bash + +jobs: + compare: + runs-on: ubuntu-24.04 + timeout-minutes: 60 + env: + ERL_FLAGS: "+S 4:4" + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + path: candidate + fetch-depth: 0 + persist-credentials: false + - name: Select baseline + id: baseline + env: + BASELINE: ${{ inputs.baseline }} + run: | + if [ -z "$BASELINE" ]; then + BASELINE=$(git -C candidate describe --tags --abbrev=0 "${GITHUB_SHA}^") + fi + printf 'ref=%s\n' "$BASELINE" >> "$GITHUB_OUTPUT" + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + path: baseline + ref: ${{ steps.baseline.outputs.ref }} + persist-credentials: false + - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1 + with: + elixir-version: "> 0" + otp-version: "> 0" + - name: Record environment and revisions + run: | + mkdir -p reports + { + elixir --version + uname -a + lscpu + git -C baseline rev-parse HEAD + git -C candidate rev-parse HEAD + } | tee reports/environment.log + - name: Benchee baseline and candidate on the same runner + run: | + GROUP_BENCH_ROOT="$PWD/baseline" \ + GROUP_BENCH_OUTPUT="$PWD/reports/baseline.benchee" \ + GROUP_BENCH_TAG=baseline \ + elixir candidate/priv/bench/comparison.exs 2>&1 | tee reports/benchee-baseline.log + GROUP_BENCH_ROOT="$PWD/candidate" \ + GROUP_BENCH_OUTPUT="$PWD/reports/candidate.benchee" \ + GROUP_BENCH_BASELINE="$PWD/reports/baseline.benchee" \ + elixir candidate/priv/bench/comparison.exs 2>&1 | tee reports/benchee-candidate.log + - name: Distributed load and recovery, baseline then candidate + run: | + epmd -daemon + for revision in baseline candidate; do + timeout 20m bash "$revision/priv/bench/run_distributed.sh" 2>&1 | + tee "reports/distributed-$revision.log" + done + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + if: always() + with: + name: performance-comparison + path: reports + retention-days: 30 diff --git a/README.md b/README.md index e605916..476f881 100644 --- a/README.md +++ b/README.md @@ -400,21 +400,25 @@ mix test See [`test/README.md`](test/README.md) for details on the distributed test infrastructure. -GitHub Actions runs the full local and distributed suite on pushes and pull -requests using Elixir 1.19 / OTP 26–28, Elixir 1.20 / OTP 27–28, and the latest +GitHub Actions runs fast local checks first, then the full suite on pull requests, +pushes to `main`, and releases using Elixir 1.19 / OTP 26–28, Elixir 1.20 / OTP 27–28, and the latest stable Elixir / OTP pair. Version ranges pick up new patch releases automatically; the latest-stable job also picks up new minor and major releases, excluding -prereleases. Every job treats compilation and test warnings as errors, and the -latest-stable job checks formatting. +prereleases. Every job treats compilation and test warnings as errors. Nightly +campaigns vary seeds and peer scheduler counts, run larger generated histories, +and check targeted mutations. Performance comparisons run separately. +See [`test/README.md`](test/README.md#ci-lanes) for lane budgets and release review. To run the same checks locally: ```bash export MIX_ENV=test export ERL_FLAGS="+S 4:4" +export GROUP_PEER_SCHEDULERS=2 mix deps.get mix format --check-formatted mix compile --warnings-as-errors +mix test --only local --warnings-as-errors # no distribution startup mix test --warnings-as-errors ``` diff --git a/priv/bench/comparison.exs b/priv/bench/comparison.exs new file mode 100644 index 0000000..cfc22f8 --- /dev/null +++ b/priv/bench/comparison.exs @@ -0,0 +1,56 @@ +# Use the same harness and Benchee version for both revisions. Only the Group +# path changes; this also works when the baseline predates this harness. +root = System.get_env("GROUP_BENCH_ROOT") || Path.expand("../..", __DIR__) +Mix.install([{:group, path: root}, {:benchee, "== 1.5.1"}]) + +{:ok, supervisor} = Group.start_link(name: :comparison, shards: 8, log: false) +parent = self() + +members = + for _ <- 1..128 do + spawn_link(fn -> + :ok = Group.join(:comparison, "members", %{}) + send(parent, {:ready, self()}) + + receive do + :stop -> :ok + end + end) + end + +for pid <- members do + receive do + {:ready, ^pid} -> :ok + after + 5_000 -> raise "benchmark member did not become ready" + end +end + +:ok = Group.register(:comparison, "lookup", %{value: 1}) +output = System.get_env("GROUP_BENCH_OUTPUT", "comparison.benchee") +baseline = System.get_env("GROUP_BENCH_BASELINE") + +try do + Benchee.run( + %{ + "lookup" => fn -> Group.lookup(:comparison, "lookup") end, + "members/128" => fn -> Group.members(:comparison, "members") end, + "register/unregister" => fn -> + :ok = Group.register(:comparison, "registry-cycle", %{}) + :ok = Group.unregister(:comparison, "registry-cycle") + end, + "join/leave" => fn -> + :ok = Group.join(:comparison, "pg-cycle", %{}) + :ok = Group.leave(:comparison, "pg-cycle") + end + }, + time: 3, + warmup: 1, + memory_time: 1, + save: [path: output, tag: System.get_env("GROUP_BENCH_TAG", "candidate")], + load: if(baseline, do: [baseline], else: []) + ) +after + for pid <- members, do: send(pid, :stop) + Supervisor.stop(supervisor) +end diff --git a/priv/ci/mutations.exs b/priv/ci/mutations.exs new file mode 100644 index 0000000..019ee2c --- /dev/null +++ b/priv/ci/mutations.exs @@ -0,0 +1,76 @@ +# Focused contract mutations, not an exhaustive mutation framework. Each runs in +# an isolated copy so interruption cannot leave the checkout's source mutated. +mutations = [ + {"reserved-prefix", "lib/group.ex", ~s|String.ends_with?(key, "/")|, + ~s|String.ends_with?(key, "//")|}, + {"zero-limit", "lib/group.ex", "is_integer(limit) and limit >= 0", + "is_integer(limit) and limit > 0"}, + {"conflict-tiebreak", "lib/group/replica.ex", + "time2 > time1 or (time2 == time1 and pid2 > pid1)", + "time2 >= time1 or (time2 == time1 and pid2 > pid1)"} +] + +root = File.cwd!() +run_dir = Path.join(root, "_build/mutations/#{System.pid()}") +File.mkdir_p!(run_dir) + +prepare = fn name -> + directory = Path.join(run_dir, name) + File.mkdir_p!(directory) + + for path <- ["lib", "test", "mix.exs", "mix.lock", ".formatter.exs"] do + File.cp_r!(Path.join(root, path), Path.join(directory, path)) + end + + directory +end + +run = fn directory, args -> + {output, status} = + System.cmd("mix", args, + cd: directory, + stderr_to_stdout: true, + env: [ + {"MIX_ENV", "test"}, + {"MIX_BUILD_PATH", Path.join(directory, "_build/test")}, + {"GROUP_TEST_DIAGNOSTICS", Path.join(directory, "diagnostics")} + ] + ) + + File.write!(Path.join(directory, "#{hd(args)}.log"), output) + IO.puts(output) + status +end + +# Use the local contracts and deterministic conflict regression, not unrelated +# network faults that could falsely "kill" a mutant by flaking. +tests = ["test/group_test.exs", "test/conflict_test.exs", "test/history_test.exs"] +test_args = ["test", "--warnings-as-errors", "--seed", "424242"] ++ tests + +# A broken baseline is not evidence that a mutation was detected. +if run.(prepare.("baseline"), test_args) != 0 do + raise "mutation baseline failed" +end + +for {name, file, original, replacement} <- mutations do + directory = prepare.(name) + path = Path.join(directory, file) + source = File.read!(path) + + unless length(String.split(source, original)) == 2 do + raise "mutation #{name} must match exactly once; update it for the current source" + end + + File.write!(path, String.replace(source, original, replacement)) + IO.puts("=== Mutation: #{name} ===") + + if run.(directory, ["compile", "--warnings-as-errors"]) != 0 do + raise "mutation #{name} did not compile; this is not a detected behavioral regression" + end + + if run.(directory, test_args ++ ["--no-compile"]) == 0 do + raise "mutation #{name} survived" + end +end + +IO.puts("All #{length(mutations)} contract mutations were detected.") diff --git a/test/README.md b/test/README.md index b0aeba4..494b69a 100644 --- a/test/README.md +++ b/test/README.md @@ -4,30 +4,91 @@ ```bash mix test # all tests -mix test test/group_test.exs # local only +mix test --only local # all local examples, histories and regressions mix test test/distributed_test.exs # distributed only ``` +## CI lanes + +| Lane | Scope and budget | +|------|------------------| +| PR / `main` | Latest-stable formatting, compiler checks and local tests first (5-minute cap), then the six supported-runtime jobs with the full suite and fresh peers (15-minute cap each). | +| Nightly, 03:23 UTC / on demand | Five seeds per job across three starting seeds and 1/2/4 peer schedulers; 2,000-operation histories on 1/4/8 shards, plus existing partitions, churn, restart and race scenarios (30-minute cap per job). | +| Nightly / on demand mutations | Three targeted mutations: reserved prefix keys, zero limits, and equal-timestamp conflict ordering. A clean baseline and compilable mutants are required; each mutant must fail the local contract suite (20-minute cap). | +| Dedicated performance | Manually selected baseline versus candidate on one runner: pinned Benchee harness and existing distributed load/recovery benchmarks. Reports retained for 30 days; no noisy hosted-runner threshold gates (60-minute cap). | +| Release | Publishing reruns the runtime matrix, fault campaigns and performance comparison against the previous tag. **Before publishing**, manually run these workflows on the candidate ref and review the performance report. Publication is not blocked automatically. | + +Nightly and release workflows become available after merging onto the default +branch. PR checks do not run benchmarks or mutation campaigns. Branch pushes do +not duplicate the PR's runtime matrix. + +The histories check a sequential reference model for registry/PG metadata and +named-cluster disconnects. They complement, rather than replace, the concurrent +and real-peer tests. The mutation checks are focused contract probes, not an +exhaustive mutation score. The nightly repetitions are bounded fault/churn +campaigns, not a claim of long-duration soak coverage. Mixed Group-version wire +compatibility is not established by this runtime matrix; add a dedicated +mixed-version scenario before promising rolling upgrades across library versions. + +### Replay and diagnostics + +```bash +GROUP_PEER_SCHEDULERS=1 GROUP_HISTORY_STEPS=2000 \ + GROUP_TEST_DIAGNOSTICS=_build/test-diagnostics \ + mix test --seed 104729 --warnings-as-errors + +# Targeted behavioral mutations run in isolated copies under _build/mutations. +elixir priv/ci/mutations.exs +``` + +CI failure artifacts retain output, seeds, generated histories, runtime/config +details, peer/test identities, and snapshots taken **before peer teardown**: +topology, ETS sizes, bounded shard state (including pending buffer counts) and +process queue sizes. Unreachable peers and timed-out state calls are recorded as +unavailable, rather than delaying cleanup indefinitely. PR artifacts are kept +for 14 days on failure; campaign reports are kept on success too. + +### Local concurrency and synchronization + +`group_test.exs` contains separate subsystem modules using `Group.LocalCase`. +Keeping the existing file entry point preserves focused commands and avoids a +large file-movement diff; ExUnit schedules the modules independently. The +clusters, counts and event-batching modules remain serial because they change +VM-global trace patterns. Diagnostics tests also remain serial because they +change an environment variable. Other local tests use uniquely named Group +instances and run asynchronously. + +Fairness regressions wait until requests appear in the suspended shard's mailbox +instead of sleeping for a presumed delivery delay. Keep deliberate timeout and +negative-event assertions distinct from synchronization waits. Destructive +distributed scenarios always get fresh peers; there is no shared peer pool. + ## Test files | File | What it tests | |------|---------------| -| `group_test.exs` | Single-node: register/unregister, join/leave, members, monitor/demonitor, named clusters, concurrent operations | +| `group_test.exs` | Independent subsystem modules: registry, groups, members, monitoring, clusters, requests, fairness, consistency, counts, queries, concurrency, events and buffering | +| `history_test.exs` | Seeded registry/PG/cluster histories against a reference model | +| `conflict_test.exs` | Equal-timestamp resolution in both delivery orders | +| `diagnostics_test.exs` | Failure output, shard snapshots and unreachable-peer diagnostics | | `distributed_test.exs` | Multi-node: replication, peer discovery, node disconnect cleanup, partition healing, conflict resolution, event ordering, rolling restarts | ## How distribution works -The test node starts as a named Erlang node in `test_helper.exs`: - -```elixir -Node.start(:"test_12345@127.0.0.1", :longnames) -Node.set_cookie(:group_test) -``` +`test_helper.exs` starts ExUnit only. `Group.TestCluster.start_peers/2` starts +EPMD/distribution lazily, so `mix test --only local` does not start distribution. +The test node and peers include the OS pid in their names to avoid cross-VM +collisions. Peers use the test node's cookie. Peer nodes are real BEAM VMs started via OTP's `:peer` module (not `Node.spawn`). Each peer has its own schedulers, memory, and GC — they communicate over Erlang distribution just like production nodes. +Correctness peers default to two schedulers, independently of the parent VM's +`ERL_FLAGS`. Override with `GROUP_PEER_SCHEDULERS=1` (1–64) or +`TestCluster.start_peers(3, schedulers: 1)`. Performance runs have separate +scheduler settings. + `:prevent_overlapping_partitions` is set to `false` on all nodes (test node and peers). Without this, disconnecting two peers from each other would also disconnect them from the test node, making partition tests impossible. diff --git a/test/conflict_test.exs b/test/conflict_test.exs new file mode 100644 index 0000000..1142fb9 --- /dev/null +++ b/test/conflict_test.exs @@ -0,0 +1,23 @@ +defmodule Group.ConflictTest do + use Group.LocalCase, async: true + + test "equal timestamps choose the larger pid regardless of delivery order" do + for order <- [:ascending, :descending] do + name = start_single_shard_group() + shard = Group.Replica.shard_name(name, 0) + [loser, winner] = Enum.sort([spawn_forever(), spawn_forever()]) + on_exit(fn -> Enum.each([loser, winner], &kill_if_alive/1) end) + pids = if order == :ascending, do: [loser, winner], else: [winner, loser] + + for pid <- pids do + send(shard, replicated_register(nil, "equal-time", pid, %{owner: pid}, :register, 123)) + end + + flush_replicated_registry_barrier(shard) + assert_receive {:replicated_registry_buffer_flushed, ^shard}, 1_000 + assert Group.lookup(name, "equal-time") == {winner, %{owner: winner}} + assert Process.alive?(winner) + refute Process.alive?(loser) + end + end +end diff --git a/test/diagnostics_test.exs b/test/diagnostics_test.exs new file mode 100644 index 0000000..e3b4354 --- /dev/null +++ b/test/diagnostics_test.exs @@ -0,0 +1,61 @@ +defmodule Group.DiagnosticsTest do + # These checks temporarily change a process-wide environment variable. + use ExUnit.Case, async: false + + @moduletag :local + @moduletag :tmp_dir + @moduletag :capture_log + + setup %{tmp_dir: directory} do + previous = System.get_env("GROUP_TEST_DIAGNOSTICS") + System.put_env("GROUP_TEST_DIAGNOSTICS", directory) + + on_exit(fn -> + if previous do + System.put_env("GROUP_TEST_DIAGNOSTICS", previous) + else + System.delete_env("GROUP_TEST_DIAGNOSTICS") + end + end) + + :ok + end + + test "retains suite seeds and failure details", %{tmp_dir: directory} do + Group.TestDiagnostics.handle_cast({:suite_started, [seed: 12345]}, []) + + failed = %ExUnit.Test{ + name: :synthetic_failure, + module: __MODULE__, + state: {:failed, [{:error, %RuntimeError{message: "diagnostic fixture"}, []}]} + } + + Group.TestDiagnostics.handle_cast({:test_finished, failed}, []) + [suite] = Path.wildcard(Path.join(directory, "*-suite.txt")) + [failure] = Path.wildcard(Path.join(directory, "*-failure.txt")) + assert File.read!(suite) =~ "12345" + assert File.read!(failure) =~ "diagnostic fixture" + end + + test "snapshot includes topology, configuration, shard state and queue sizes" do + name = :"diagnostics_#{System.unique_integer([:positive])}" + start_supervised!({Group, name: name, shards: 1, log: false}) + snapshot = Group.TestDiagnostics.snapshot() + assert snapshot.node == node() + assert snapshot.connected_nodes == Node.list() + group = Enum.find(snapshot.groups, &(&1.name == name)) + assert group.config.num_shards == 1 + [shard] = group.shards + assert shard.process[:message_queue_len] >= 0 + assert shard.state =~ "pending_replicated_registry_len" + assert shard.state =~ "remote_shards" + assert snapshot.tables != [] + end + + test "unreachable peers produce diagnostics instead of hanging", %{tmp_dir: directory} do + peer = :"unreachable_diagnostic_#{System.unique_integer([:positive])}@127.0.0.1" + Group.TestDiagnostics.capture_peers([{nil, peer}]) + [snapshot] = Path.wildcard(Path.join(directory, "*-peer_snapshot.txt")) + assert File.read!(snapshot) =~ "unavailable" + end +end diff --git a/test/distributed_test.exs b/test/distributed_test.exs index dcb9e9c..350565d 100644 --- a/test/distributed_test.exs +++ b/test/distributed_test.exs @@ -1,11 +1,17 @@ defmodule Group.DistributedTest do use ExUnit.Case + @moduletag :distributed @moduletag :capture_log @moduletag timeout: 30_000 alias Group.TestCluster + setup context do + Process.put(:group_test_context, Map.take(context, [:module, :test, :file, :line])) + :ok + end + defp start_group_on_peers(peers, opts) do for {_pid, node} <- peers do TestCluster.start_group(node, opts) @@ -3174,9 +3180,24 @@ defmodule Group.DistributedTest do # Flap 3 times for _i <- 1..3 do + TestCluster.assert_group_nodes(node_a, name, [node_b]) + TestCluster.assert_group_nodes(node_b, name, [node_a]) + TestCluster.flush_shards(node_a, name) + TestCluster.flush_shards(node_b, name) + TestCluster.disconnect_nodes(node_a, node_b) assert_receive {:nodedown_on_remote, ^node_b}, 5000 + # A monitor notification can overtake shard cleanup. Do not let old + # replicated rows satisfy the re-sync assertion in the next cycle. + TestCluster.assert_group_nodes(node_a, name, []) + TestCluster.assert_group_nodes(node_b, name, []) + + TestCluster.assert_eventually(fn -> + TestCluster.rpc!(node_b, Group, :lookup, [name, "stable/a"]) == nil and + TestCluster.rpc!(node_a, Group, :members, [name, "room/nil"]) == [] + end) + TestCluster.reconnect_nodes(node_a, node_b) # Wait for data to re-sync diff --git a/test/group_test.exs b/test/group_test.exs index d0c1c22..454ff08 100644 --- a/test/group_test.exs +++ b/test/group_test.exs @@ -18,16 +18,8 @@ defmodule GroupTest.ResolveRegistryConflict do end end -defmodule GroupTest do - use ExUnit.Case, async: true - - @moduletag :capture_log - - setup do - name = :"test_group_#{System.unique_integer([:positive])}" - start_supervised!({Group, name: name, shards: 4, log: false}) - {:ok, name: name} - end +defmodule GroupTest.Startup do + use Group.LocalCase, async: true describe "startup options" do test "runtime config omits unused callback state", %{name: name} do @@ -48,6 +40,10 @@ defmodule GroupTest do end end end +end + +defmodule GroupTest.ProcessGroups do + use Group.LocalCase, async: true describe "join/3 and leave/2" do test "joined process appears in members/2", %{name: name} do @@ -182,6 +178,10 @@ defmodule GroupTest do [{_pid, %{v: 2}}] = Group.members(name, key) end end +end + +defmodule GroupTest.Registry do + use Group.LocalCase, async: true describe "register/unregister" do test "register makes process discoverable via lookup", %{name: name} do @@ -295,6 +295,10 @@ defmodule GroupTest do assert Group.lookup(name, key) == nil end end +end + +defmodule GroupTest.Membership do + use Group.LocalCase, async: true describe "members/2" do test "returns only joined processes", %{name: name} do @@ -547,6 +551,10 @@ defmodule GroupTest do end end end +end + +defmodule GroupTest.Monitoring do + use Group.LocalCase, async: true describe "self-events" do test "joining process receives its own :joined event if subscribed", %{name: name} do @@ -650,6 +658,11 @@ defmodule GroupTest do refute_receive {:group, _, _}, 200 end end +end + +defmodule GroupTest.Clusters do + # Dispatch assertions change VM-global trace patterns. + use Group.LocalCase, async: false describe "named clusters" do test "connect and disconnect reject non-binary cluster names", %{name: name} do @@ -1206,6 +1219,10 @@ defmodule GroupTest do refute_receive {:group, _, _}, 200 end end +end + +defmodule GroupTest.Requests do + use Group.LocalCase, async: true describe "call timeout option" do test "register honors timeout option" do @@ -1345,6 +1362,10 @@ defmodule GroupTest do assert Group.members(name, key, cluster: cluster) == [] end end +end + +defmodule GroupTest.Fairness do + use Group.LocalCase, async: true describe "local request fairness" do test "local register gets a turn ahead of replicated registry backlog" do @@ -1374,7 +1395,7 @@ defmodule GroupTest do ) on_exit(fn -> kill_if_alive(caller) end) - Process.sleep(20) + wait_for_local_requests(shard) :ok = :sys.resume(shard) assert_receive {:local_register_result, ^caller, :ok}, 1_000 @@ -1410,7 +1431,7 @@ defmodule GroupTest do ) on_exit(fn -> kill_if_alive(caller) end) - Process.sleep(20) + wait_for_local_requests(shard) :ok = :sys.resume(shard) assert_receive {:local_join_result, ^caller, :ok}, 1_000 @@ -1446,7 +1467,7 @@ defmodule GroupTest do ) on_exit(fn -> kill_if_alive(caller) end) - Process.sleep(20) + wait_for_local_requests(shard) :ok = :sys.resume(shard) assert_receive {:local_join_result, ^caller, :ok}, 1_000 @@ -1482,7 +1503,7 @@ defmodule GroupTest do ) on_exit(fn -> kill_if_alive(connect_caller) end) - Process.sleep(20) + wait_for_local_requests(shard) :ok = :sys.resume(shard) assert_receive {:local_connect_result, ^connect_caller, :ok}, 1_000 @@ -1505,7 +1526,7 @@ defmodule GroupTest do ) on_exit(fn -> kill_if_alive(disconnect_caller) end) - Process.sleep(20) + wait_for_local_requests(shard) :ok = :sys.resume(shard) assert_receive {:local_disconnect_result, ^disconnect_caller, :ok}, 1_000 @@ -1541,7 +1562,7 @@ defmodule GroupTest do :fifo_result ) - Process.sleep(20) + wait_for_local_requests(shard) caller2 = spawn_requester( @@ -1556,7 +1577,7 @@ defmodule GroupTest do kill_if_alive(caller2) end) - Process.sleep(20) + wait_for_local_requests(shard, 2) :ok = :sys.resume(shard) assert_receive {:fifo_result, ^caller1, :ok}, 1_000 @@ -1700,6 +1721,10 @@ defmodule GroupTest do assert Group.members(name, join_key2) == [{caller2, %{order: 2}}] end end +end + +defmodule GroupTest.Consistency do + use Group.LocalCase, async: true describe "ETS table consistency" do test "tables are consistent after register + unregister", %{name: name} do @@ -1840,6 +1865,11 @@ defmodule GroupTest do assert Group.TestCluster.assert_ets_consistent(name) == :ok end end +end + +defmodule GroupTest.Counts do + # ETS call-count assertions change VM-global trace patterns. + use Group.LocalCase, async: false describe "local_registry_count/1" do test "local activity checks use bounded ETS selects", %{name: name} do @@ -1992,6 +2022,10 @@ defmodule GroupTest do assert Group.local_member_count(name, other) == 1 end end +end + +defmodule GroupTest.Queries do + use Group.LocalCase, async: true describe "local_members/3" do test "returns local exact-key members and honors the limit", %{name: name} do @@ -2153,6 +2187,10 @@ defmodule GroupTest do ]) end end +end + +defmodule GroupTest.Concurrency do + use Group.LocalCase, async: true describe "concurrent operations" do test "concurrent join/leave on same key doesn't produce duplicates", %{name: name} do @@ -2219,6 +2257,11 @@ defmodule GroupTest do assert length(error_results) == 4 end end +end + +defmodule GroupTest.Events do + # Replication assertions change VM-global trace patterns. + use Group.LocalCase, async: false describe "event batching" do test "process death batches :unregistered and :left into one message", %{name: name} do @@ -2348,6 +2391,10 @@ defmodule GroupTest do assert [%Group.Event{type: :unregistered}] = events end end +end + +defmodule GroupTest.PGBuffering do + use Group.LocalCase, async: true describe "replicated PG receiver buffering" do test "flushes buffered replicated joins when buffer size is reached" do @@ -2506,6 +2553,10 @@ defmodule GroupTest do assert Group.members(name, key) == [{pid, %{v: 1}}] end end +end + +defmodule GroupTest.RegistryBuffering do + use Group.LocalCase, async: true describe "replicated registry receiver buffering" do test "barrier messages flush buffered replicated register and unregister ops in order" do @@ -2853,139 +2904,4 @@ defmodule GroupTest do assert Group.Replica.Data.registry_lookup_by_pid(name, 0, remote_pid) == [] end end - - defp start_single_shard_group(opts \\ []) do - name = :"test_timeout_group_#{System.unique_integer([:positive])}" - opts = Keyword.merge([name: name, shards: 1, log: false], opts) - start_supervised!({Group, opts}) - name - end - - defp suspend_only_shard(name) do - shard = Group.Replica.shard_name(name, 0) - :ok = :sys.suspend(shard) - shard - end - - defp resume_shard_if_alive(shard) do - if Process.whereis(shard) do - :ok = :sys.resume(shard) - end - - :ok - end - - defp assert_genserver_call_timeout(fun) do - assert {:timeout, {GenServer, :call, _}} = catch_exit(fun.()) - end - - defp replicated_pg_join(cluster, key, pid, meta, reason) do - {:replicate_pg_batch, - [{:join, cluster, key, pid, meta, System.system_time(), reason, node(pid)}]} - end - - defp replicated_pg_leave(cluster, key, pid, meta, reason) do - {:replicate_pg_batch, [{:leave, cluster, key, pid, meta, reason}]} - end - - defp replicated_register(cluster, key, pid, meta, _reason, time \\ System.system_time()) do - {:replicate_registry_batch, [{:register, cluster, key, pid, meta, time, node(pid)}]} - end - - defp replicated_unregister(cluster, key, pid, meta, reason) do - {:replicate_registry_batch, [{:unregister, cluster, key, pid, meta, reason}]} - end - - defp enqueue_replicated_pg_backlog(shard, key_prefix, pid, count) do - for i <- 1..count do - send(shard, replicated_pg_join(nil, "#{key_prefix}/#{i}", pid, %{}, :join)) - end - - :ok - end - - defp enqueue_replicated_registry_backlog(shard, key_prefix, pid, count) do - for i <- 1..count do - send(shard, replicated_register(nil, "#{key_prefix}/#{i}", pid, %{seq: i}, :register)) - end - - :ok - end - - defp spawn_requester(fun, tag) do - parent = self() - - spawn(fn -> - result = fun.() - send(parent, {tag, self(), result}) - Process.sleep(:infinity) - end) - end - - defp shard_message_queue_len(shard) do - case Process.info(Process.whereis(shard), :message_queue_len) do - {:message_queue_len, len} -> len - nil -> 0 - end - end - - defp flush_replicated_pg_barrier(shard) do - send(shard, {:group_dispatch, [self()], {:replicated_pg_buffer_flushed, shard}}) - end - - defp flush_replicated_registry_barrier(shard) do - send(shard, {:group_dispatch, [self()], {:replicated_registry_buffer_flushed, shard}}) - end - - defp force_cluster_lease_sweep(name) do - lease_manager = Group.ClusterLease.lease_name(name) - send(lease_manager, :force_sweep) - :sys.get_state(lease_manager) - :ok - end - - defp expire_cluster_lease(name, cluster) do - {ttl_ms, _expires_at} = Group.Replica.Data.cluster_lease(name, cluster) - - Group.Replica.Data.put_cluster_lease( - name, - cluster, - ttl_ms, - System.monotonic_time(:millisecond) - 1 - ) - - ttl_ms - end - - defp spawn_forever do - spawn(fn -> Process.sleep(:infinity) end) - end - - defp kill_if_alive(pid) do - if Process.alive?(pid) do - Process.exit(pid, :kill) - end - - :ok - end - - defp wait_until(fun, timeout \\ 1_000) - - defp wait_until(fun, timeout) do - deadline = System.monotonic_time(:millisecond) + timeout - do_wait_until(fun, deadline) - end - - defp do_wait_until(fun, deadline) do - if fun.() do - :ok - else - if System.monotonic_time(:millisecond) >= deadline do - flunk("condition did not become true") - end - - Process.sleep(10) - do_wait_until(fun, deadline) - end - end end diff --git a/test/history_test.exs b/test/history_test.exs new file mode 100644 index 0000000..5dbc8f6 --- /dev/null +++ b/test/history_test.exs @@ -0,0 +1,87 @@ +defmodule Group.HistoryTest do + use ExUnit.Case, async: true + + @moduletag :local + @moduletag :capture_log + @moduletag :history + + # ExUnit seeds :rand per test. The complete generated input is retained in CI, + # so failures can be replayed with --seed, even when the campaign is larger. + for shards <- [1, 4, 8] do + @tag shards: shards + test "registry and PG histories match a reference model with #{shards} shards", context do + name = :"history_#{System.unique_integer([:positive])}" + start_supervised!({Group, name: name, shards: context.shards, log: false}) + steps = String.to_integer(System.get_env("GROUP_HISTORY_STEPS", "100")) + assert steps in 1..100_000 + + history = + for step <- 1..steps do + {Enum.random([:register, :unregister, :join, :leave, :disconnect]), + Enum.random([nil, "red", "blue"]), "key/#{Enum.random(1..4)}", %{step: step}} + end + + Group.TestDiagnostics.record(:history, %{ + test: context.test, + shards: context.shards, + seed: ExUnit.configuration()[:seed], + operations: history + }) + + Enum.reduce(history, {%{}, %{}}, fn operation, model -> + model = apply_operation(name, operation, model) + assert_model(name, model, operation) + model + end) + end + end + + defp apply_operation(_name, {:disconnect, nil, _key, _meta}, model), do: model + + defp apply_operation(name, {:disconnect, cluster, _key, _meta}, {registry, groups}) do + :ok = Group.disconnect(name, cluster) + keep? = fn {{entry_cluster, _key}, _meta} -> entry_cluster != cluster end + {Map.filter(registry, keep?), Map.filter(groups, keep?)} + end + + defp apply_operation(name, {operation, cluster, key, meta}, {registry, groups}) do + if cluster, do: Group.connect(name, cluster) + opts = [cluster: cluster] + entry = {cluster, key} + + case operation do + :register -> + :ok = Group.register(name, key, meta, opts) + {Map.put(registry, entry, meta), groups} + + :unregister -> + expected = if Map.has_key?(registry, entry), do: :ok, else: {:error, :undefined} + assert Group.unregister(name, key, opts) == expected + {Map.delete(registry, entry), groups} + + :join -> + :ok = Group.join(name, key, meta, opts) + {registry, Map.put(groups, entry, meta)} + + :leave -> + expected = if Map.has_key?(groups, entry), do: :ok, else: {:error, :not_in_group} + assert Group.leave(name, key, opts) == expected + {registry, Map.delete(groups, entry)} + end + end + + defp assert_model(name, {registry, groups}, operation) do + for cluster <- [nil, "red", "blue"], key <- 1..4 do + key = "key/#{key}" + entry = {cluster, key} + expected_registration = if meta = registry[entry], do: {self(), meta} + expected_members = if meta = groups[entry], do: [{self(), meta}], else: [] + + assert Group.lookup(name, key, cluster: cluster) == expected_registration, + "registry mismatch after #{inspect(operation)}" + + assert Group.members(name, key, cluster: cluster) == expected_members, + "PG mismatch after #{inspect(operation)}" + end + end +end diff --git a/test/support/local_case.ex b/test/support/local_case.ex new file mode 100644 index 0000000..5abb864 --- /dev/null +++ b/test/support/local_case.ex @@ -0,0 +1,163 @@ +defmodule Group.LocalCase do + @moduledoc false + use ExUnit.CaseTemplate + + using do + quote do + import Group.LocalCase + @moduletag :local + @moduletag :capture_log + end + end + + setup do + name = :"test_group_#{System.unique_integer([:positive])}" + start_supervised!({Group, name: name, shards: 4, log: false}) + {:ok, name: name} + end + + def start_single_shard_group(opts \\ []) do + name = :"test_timeout_group_#{System.unique_integer([:positive])}" + opts = Keyword.merge([name: name, shards: 1, log: false], opts) + start_supervised!({Group, opts}) + name + end + + def suspend_only_shard(name) do + shard = Group.Replica.shard_name(name, 0) + :ok = :sys.suspend(shard) + shard + end + + def resume_shard_if_alive(shard) do + if Process.whereis(shard) do + :ok = :sys.resume(shard) + end + + :ok + end + + def assert_genserver_call_timeout(fun) do + assert {:timeout, {GenServer, :call, _}} = catch_exit(fun.()) + end + + def replicated_pg_join(cluster, key, pid, meta, reason) do + {:replicate_pg_batch, + [{:join, cluster, key, pid, meta, System.system_time(), reason, node(pid)}]} + end + + def replicated_pg_leave(cluster, key, pid, meta, reason) do + {:replicate_pg_batch, [{:leave, cluster, key, pid, meta, reason}]} + end + + def replicated_register(cluster, key, pid, meta, _reason, time \\ System.system_time()) do + {:replicate_registry_batch, [{:register, cluster, key, pid, meta, time, node(pid)}]} + end + + def replicated_unregister(cluster, key, pid, meta, reason) do + {:replicate_registry_batch, [{:unregister, cluster, key, pid, meta, reason}]} + end + + def enqueue_replicated_pg_backlog(shard, key_prefix, pid, count) do + for i <- 1..count do + send(shard, replicated_pg_join(nil, "#{key_prefix}/#{i}", pid, %{}, :join)) + end + + :ok + end + + def enqueue_replicated_registry_backlog(shard, key_prefix, pid, count) do + for i <- 1..count do + send(shard, replicated_register(nil, "#{key_prefix}/#{i}", pid, %{seq: i}, :register)) + end + + :ok + end + + def spawn_requester(fun, tag) do + parent = self() + + spawn(fn -> + result = fun.() + send(parent, {tag, self(), result}) + Process.sleep(:infinity) + end) + end + + def shard_message_queue_len(shard) do + case Process.info(Process.whereis(shard), :message_queue_len) do + {:message_queue_len, len} -> len + nil -> 0 + end + end + + def wait_for_local_requests(shard, count \\ 1) do + wait_until(fn -> + {:messages, messages} = Process.info(Process.whereis(shard), :messages) + + Enum.count(messages, fn + {:group_local_request, _alias, _request} -> true + {:group_local_request, _caller, _ref, _request} -> true + _ -> false + end) >= count + end) + end + + def flush_replicated_pg_barrier(shard) do + send(shard, {:group_dispatch, [self()], {:replicated_pg_buffer_flushed, shard}}) + end + + def flush_replicated_registry_barrier(shard) do + send(shard, {:group_dispatch, [self()], {:replicated_registry_buffer_flushed, shard}}) + end + + def force_cluster_lease_sweep(name) do + lease_manager = Group.ClusterLease.lease_name(name) + send(lease_manager, :force_sweep) + :sys.get_state(lease_manager) + :ok + end + + def expire_cluster_lease(name, cluster) do + {ttl_ms, _expires_at} = Group.Replica.Data.cluster_lease(name, cluster) + + Group.Replica.Data.put_cluster_lease( + name, + cluster, + ttl_ms, + System.monotonic_time(:millisecond) - 1 + ) + + ttl_ms + end + + def spawn_forever do + spawn(fn -> Process.sleep(:infinity) end) + end + + def kill_if_alive(pid) do + if Process.alive?(pid) do + Process.exit(pid, :kill) + end + + :ok + end + + def wait_until(fun, timeout \\ 1_000) do + deadline = System.monotonic_time(:millisecond) + timeout + do_wait_until(fun, deadline) + end + + defp do_wait_until(fun, deadline) do + if fun.() do + :ok + else + if System.monotonic_time(:millisecond) >= deadline do + flunk("condition did not become true") + end + + Process.sleep(10) + do_wait_until(fun, deadline) + end + end +end diff --git a/test/support/test_cluster.ex b/test/support/test_cluster.ex index c1014a5..153c6a0 100644 --- a/test/support/test_cluster.ex +++ b/test/support/test_cluster.ex @@ -1,13 +1,45 @@ defmodule Group.TestCluster do @moduledoc false + @doc "Start distribution only when a test actually needs real peers." + def ensure_distribution do + unless Node.alive?() do + epmd = System.find_executable("epmd") || raise "epmd executable not found" + + case System.cmd(epmd, ["-daemon"], stderr_to_stdout: true) do + {_, 0} -> :ok + {output, status} -> raise "failed to start epmd (status #{status}): #{output}" + end + + name = :"test_#{System.pid()}_#{System.unique_integer([:positive])}@127.0.0.1" + {:ok, _} = Node.start(name, :longnames) + Node.set_cookie(:group_test) + end + + # Partition tests must keep their control connection to the test node. + :application.set_env(:kernel, :prevent_overlapping_partitions, false) + :ok + end + @doc "Start N peer nodes with Group app loaded and ready" def start_peers(count, opts \\ []) do + ensure_distribution() cookie = Keyword.get(opts, :cookie, Node.get_cookie()) code_paths = :code.get_path() + schedulers = Keyword.get(opts, :schedulers, System.get_env("GROUP_PEER_SCHEDULERS", "2")) + schedulers = String.to_integer(to_string(schedulers)) + unless schedulers in 1..64, do: raise(ArgumentError, "peer schedulers must be in 1..64") args = - [~c"-setcookie", ~c"#{cookie}", ~c"-kernel", ~c"prevent_overlapping_partitions", ~c"false"] ++ + [ + ~c"+S", + ~c"#{schedulers}:#{schedulers}", + ~c"-setcookie", + ~c"#{cookie}", + ~c"-kernel", + ~c"prevent_overlapping_partitions", + ~c"false" + ] ++ Enum.flat_map(code_paths, fn p -> [~c"-pa", p] end) for _i <- 1..count do @@ -18,11 +50,26 @@ defmodule Group.TestCluster do {:ok, _} = :rpc.call(node, :application, :ensure_all_started, [:elixir]) {:ok, _} = :rpc.call(node, :application, :ensure_all_started, [:group]) + + Group.TestDiagnostics.record(:peer_started, %{ + test: Process.get(:group_test_context), + node: node, + schedulers: schedulers + }) + {pid, node} end end def stop_peers(peers) do + try do + Group.TestDiagnostics.capture_peers(peers) + after + do_stop_peers(peers) + end + end + + defp do_stop_peers(peers) do Enum.each(peers, fn {pid, _node} -> if pid do try do diff --git a/test/support/test_diagnostics.ex b/test/support/test_diagnostics.ex new file mode 100644 index 0000000..8ffe902 --- /dev/null +++ b/test/support/test_diagnostics.ex @@ -0,0 +1,98 @@ +defmodule Group.TestDiagnostics do + @moduledoc false + use GenServer + + def init(opts), do: {:ok, opts} + + def handle_cast({:suite_started, opts}, state) do + record(:suite, %{ + options: opts, + elixir: System.version(), + otp: System.otp_release(), + schedulers: :erlang.system_info(:schedulers_online), + peer_schedulers: System.get_env("GROUP_PEER_SCHEDULERS", "2"), + history_steps: System.get_env("GROUP_HISTORY_STEPS", "100") + }) + + {:noreply, state} + end + + def handle_cast({:test_finished, %ExUnit.Test{state: {:failed, _}} = test}, state) do + record(:failure, test) + {:noreply, state} + end + + def handle_cast(_event, state), do: {:noreply, state} + + def record(kind, data) do + if directory = System.get_env("GROUP_TEST_DIAGNOSTICS") do + File.mkdir_p!(directory) + id = "#{System.pid()}-#{System.unique_integer([:positive, :monotonic])}" + path = Path.join(directory, "#{id}-#{kind}.txt") + File.write!(path, inspect(data, pretty: true, limit: :infinity, printable_limit: :infinity)) + end + + :ok + end + + def capture_peers(peers) do + if System.get_env("GROUP_TEST_DIAGNOSTICS") do + for {_pid, peer} <- peers do + snapshot = + try do + :erpc.call(peer, __MODULE__, :snapshot, [], 2_000) + catch + kind, reason -> %{unavailable: {kind, reason}} + end + + record(:peer_snapshot, %{node: peer, snapshot: snapshot}) + end + end + end + + # Called before stopping fresh peers, including during failure/timeout cleanup. + # Bounded sys calls preserve evidence without turning a stuck shard into a hang. + def snapshot do + groups = + for {{Group, name}, config} <- :persistent_term.get(), + is_map(config), + is_integer(config[:num_shards]) do + shards = + for index <- 0..(config.num_shards - 1) do + shard = Group.Replica.shard_name(name, index) + pid = Process.whereis(shard) + + state = + try do + :sys.get_state(shard, 100) + |> inspect(pretty: true, limit: 100, printable_limit: 8_000) + catch + kind, reason -> {kind, reason} + end + + %{ + shard: shard, + state: state, + process: pid && Process.info(pid, [:status, :current_function, :message_queue_len]) + } + end + + %{name: name, config: config, shards: shards} + end + + %{ + node: node(), + connected_nodes: Node.list(), + schedulers: :erlang.system_info(:schedulers_online), + groups: groups, + tables: + Enum.map(:ets.all(), fn table -> + :ets.info(table) + |> case do + :undefined -> :deleted + info -> Keyword.take(info, [:name, :size, :memory, :owner]) + end + end) + } + end +end diff --git a/test/test_helper.exs b/test/test_helper.exs index 158a305..ddab61d 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -1,27 +1,5 @@ -# Start epmd and distribution if they are not already running (needed for distributed tests) -epmd = System.find_executable("epmd") || raise "epmd executable not found" -epmd_running? = fn -> match?({_, 0}, System.cmd(epmd, ["-names"], stderr_to_stdout: true)) end - -unless epmd_running?.() do - case System.cmd(epmd, ["-daemon"], stderr_to_stdout: true) do - {_, 0} -> :ok - {output, status} -> raise "failed to start epmd (status #{status}): #{output}" - end - - unless epmd_running?.() do - raise "epmd did not become available after starting it" - end -end +ExUnit.start() -unless Node.alive?() do - # unique_integer is VM-local; include the OS pid to allow concurrent test runs. - name = :"test_#{System.pid()}_#{System.unique_integer([:positive])}@127.0.0.1" - {:ok, _} = Node.start(name, :longnames) - Node.set_cookie(:group_test) +if System.get_env("GROUP_TEST_DIAGNOSTICS") do + ExUnit.configure(formatters: [ExUnit.CLIFormatter, Group.TestDiagnostics]) end - -# Disable :global's partition prevention to allow peer-to-peer disconnects -# in distributed tests without the test node also being disconnected. -:application.set_env(:kernel, :prevent_overlapping_partitions, false) - -ExUnit.start()