diff --git a/README.md b/README.md index 789c27b..3fba9ca 100644 --- a/README.md +++ b/README.md @@ -315,6 +315,46 @@ Run the tests: mix test ``` +### Generated correctness models (StreamData) + +```bash +mix test --only property --seed 12345 +DURABLE_PROPERTY_RUNS=1000 mix test --only property --seed 12345 --timeout 300000 +``` + +The `property` tests also run in the default suite. Each property runs 100 +generated sequences by default; `DURABLE_PROPERTY_RUNS` increases that budget. +Keep the ExUnit seed, dependency lockfile, and minimized inputs from a failure to +reproduce it. StreamData shrinks command lists and values, not BEAM scheduling. + +Two independent reference models exercise real, single-node EKV storage: + +- **Object generations:** claims, reads, conditional writes/deletes, and + delete/recreate sequences across three keys. Historical ETags are opaque + handles; the model decides whether they are valid using its own logical + generations. Every command checks all modeled keys. +- **Acknowledged durability:** unsynced updates, all three strict sync callback + forms, process kills, explicit restarts, and graceful termination through + `DurableServer.Supervisor.terminate_child/2`. The model tracks in-memory and + durable values separately. Automatic and periodic sync are disabled, so an + unsynced update can be deterministically lost on a kill. A call timeout fails + this fault-free write scenario; it is never interpreted as proof of no commit. + +Every generated input and shrink attempt starts a fresh supervision tree and +storage directory. Teardown stops DurableServer before EKV, then removes that +sample's data, even after assertion failures. A focused test checks isolation +after a deliberately failed sample. Fixed process names are reused only within +these synchronous properties to avoid allocating atoms per generated input. + +The properties themselves need neither cloud credentials nor LocalStack. +Until the fixture-isolation change is merged, the global test helper still +requires LocalStack even when selecting only properties. + +This is sequential, per-key verification—not cross-key transactions, distributed +linearizability, external-side-effect fencing, or deterministic fault simulation. +Cordon/uncordon, lease boundaries, lost responses after commit, eventual discovery, +resource bounds, and mirror migration phases need separate models/fault controls. + ### Integration Tests (with Tigris) Set the required environment variables: diff --git a/config/config.exs b/config/config.exs index bf336dd..bf03980 100644 --- a/config/config.exs +++ b/config/config.exs @@ -2,4 +2,8 @@ import Config if config_env() == :test do config :logger, level: :info + + property_runs = String.to_integer(System.get_env("DURABLE_PROPERTY_RUNS", "100")) + if property_runs < 1, do: raise("DURABLE_PROPERTY_RUNS must be positive") + config :stream_data, max_runs: property_runs end diff --git a/mix.exs b/mix.exs index b92ea1e..25787a2 100644 --- a/mix.exs +++ b/mix.exs @@ -46,6 +46,7 @@ defmodule DurableServer.MixProject do {:finch, "~> 0.18"}, {:sweet_xml, "~> 0.7"}, {:ex_doc, "~> 0.30", only: :dev, runtime: false}, + {:stream_data, "~> 1.2", only: :test}, {:ekv, "~> 0.4.0", optional: true} ] end diff --git a/mix.lock b/mix.lock index 6e8c178..d6bc2a4 100644 --- a/mix.lock +++ b/mix.lock @@ -18,6 +18,7 @@ "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"}, "req": {:hex, :req, "0.5.17", "0096ddd5b0ed6f576a03dde4b158a0c727215b15d2795e59e0916c6971066ede", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0b8bc6ffdfebbc07968e59d3ff96d52f2202d0536f10fef4dc11dc02a2a43e39"}, "req_s3": {:hex, :req_s3, "0.2.3", "ede5f4c792cf39995379307733ff4593032a876f38da29d9d7ea03881b498b51", [:mix], [{:req, "~> 0.5.6", [hex: :req, repo: "hexpm", optional: false]}], "hexpm", "31b5d52490495c8aeea7e3c5cbcec82f49035e11bdaf41f0e58ab716fefe44ca"}, + "stream_data": {:hex, :stream_data, "1.4.0", "026f929db613aabea6208012ae9b8970d3fd5f88b3bdf26831bc536f98c42036", [:mix], [], "hexpm", "2b0ee3a340dcce1c8cf6302a763ee757d1e01c54d6e16d9069062509d68b1dc9"}, "sweet_xml": {:hex, :sweet_xml, "0.7.5", "803a563113981aaac202a1dbd39771562d0ad31004ddbfc9b5090bdcd5605277", [:mix], [], "hexpm", "193b28a9b12891cae351d81a0cead165ffe67df1b73fe5866d10629f4faefb12"}, "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, } diff --git a/test/durability_model_property_test.exs b/test/durability_model_property_test.exs new file mode 100644 index 0000000..eb5240a --- /dev/null +++ b/test/durability_model_property_test.exs @@ -0,0 +1,137 @@ +defmodule DurableServer.DurabilityModelPropertyTest do + use ExUnit.Case, async: false + use ExUnitProperties + + alias DurableServer.{PropertyFixture, StorageBackend, StoredState} + + @moduletag :property + @moduletag capture_log: [level: :warning] + + defmodule Counter do + use DurableServer, vsn: 1 + + def dump_state(state), do: state + def load_state(_vsn, state), do: state + # Disable periodic persistence so losing an unsynced update is deterministic. + # This property tests explicit restarts, not permanent-object discovery. + def init(state), do: {:ok, state, auto_sync: false, sync_every_ms: nil} + + def handle_call(:read, _from, state), do: {:reply, state.count, state} + + def handle_call({:add, delta, :unsynced}, _from, state), + do: {:reply, state.count + delta, %{state | count: state.count + delta}} + + def handle_call({:add, delta, :sync}, _from, state), + do: {:reply, state.count + delta, %{state | count: state.count + delta}, :sync} + + def handle_call({:add, delta, :sync_metadata}, _from, state), + do: + {:reply, state.count + delta, %{state | count: state.count + delta}, + {:sync, %{status: :running}}} + + def handle_call({:add, delta, :sync_option}, _from, state), + do: {:reply, state.count + delta, %{state | count: state.count + delta}, sync: true} + end + + setup do + root = Path.expand("tmp/durability_property/#{DurableServer.UUID.uuid4()}") + on_exit(fn -> File.rm_rf!(root) end) + {:ok, root: root} + end + + property "strict acknowledgements survive crashes; unsynced updates need not", %{root: root} do + check all( + initial <- integer(-100..100), + delta <- member_of([-5, -1, 1, 5]), + mode <- member_of([:sync, :sync_metadata, :sync_option]), + commands <- list_of(command(), max_length: 40) + ) do + PropertyFixture.with_sample(root, :durable, fn fixture -> + pid = start_counter(fixture.supervisor, initial) + model = %{pid: pid, memory: initial, durable: initial} + + # Mandatory witness prevents a generated sequence from passing without a + # strict acknowledgement followed by a crash, including during shrinking. + commands = + [{:add, delta, mode}, {:add, delta, :unsynced}, :crash_restart] ++ + commands ++ [:crash_restart] + + Enum.reduce(commands, model, fn command, model -> + model = step(command, model, fixture) + assert GenServer.call(model.pid, :read) == model.memory + + assert {:ok, %{body: %StoredState{state: %{count: persisted}}}} = + StorageBackend.get_object(fixture.backend, "property/counter", + consistent: true + ) + + assert persisted == model.durable + model + end) + end) + end + end + + defp command do + frequency([ + {6, + tuple( + {constant(:add), integer(-100..100), + member_of([:unsynced, :sync, :sync_metadata, :sync_option])} + )}, + {2, constant(:crash_restart)}, + {1, constant(:graceful_restart)} + ]) + end + + defp step({:add, delta, mode} = command, model, _fixture) do + expected = model.memory + delta + # Do not catch call exits as failed writes: an unacknowledged call is unknown. + # This fault-free command must return an acknowledgement or fail the property. + assert GenServer.call(model.pid, command) == expected + durable = if mode == :unsynced, do: model.durable, else: expected + %{model | memory: expected, durable: durable} + end + + defp step(restart, model, fixture) when restart in [:crash_restart, :graceful_restart] do + ref = Process.monitor(model.pid) + + case restart do + :crash_restart -> Process.exit(model.pid, :kill) + :graceful_restart -> DurableServer.Supervisor.terminate_child(fixture.supervisor, model.pid) + end + + expected_reason = if restart == :crash_restart, do: :killed, else: :normal + assert_receive {:DOWN, ^ref, :process, _, ^expected_reason}, 1_000 + await_unregistered(fixture.supervisor) + + durable = if restart == :crash_restart, do: model.durable, else: model.memory + # A deliberately different initial value detects accidentally starting fresh. + pid = start_counter(fixture.supervisor, durable + 1) + refute pid == model.pid + %{pid: pid, memory: durable, durable: durable} + end + + defp start_counter(supervisor, initial) do + assert {:ok, {pid, _}} = + DurableServer.Supervisor.start_child( + supervisor, + {Counter, key: "counter", initial_state: %{count: initial}} + ) + + pid + end + + defp await_unregistered(supervisor) do + deadline = System.monotonic_time(:millisecond) + 1_000 + await_unregistered(supervisor, deadline) + end + + defp await_unregistered(supervisor, deadline) do + if DurableServer.Supervisor.lookup(supervisor, "counter") do + assert System.monotonic_time(:millisecond) < deadline, "dead owner remained registered" + Process.sleep(1) + await_unregistered(supervisor, deadline) + end + end +end diff --git a/test/storage_model_property_test.exs b/test/storage_model_property_test.exs new file mode 100644 index 0000000..88593c1 --- /dev/null +++ b/test/storage_model_property_test.exs @@ -0,0 +1,144 @@ +defmodule DurableServer.StorageModelPropertyTest do + use ExUnit.Case, async: false + use ExUnitProperties + + alias DurableServer.{PropertyFixture, StorageBackend} + + @moduletag :property + @moduletag capture_log: [level: :warning] + + setup do + root = Path.expand("tmp/storage_property/#{DurableServer.UUID.uuid4()}") + on_exit(fn -> File.rm_rf!(root) end) + {:ok, root: root} + end + + property "claims and conditional mutations follow object generations, including delete/recreate", + %{root: root} do + check all(commands <- list_of(command(), max_length: 60)) do + PropertyFixture.with_sample(root, :storage, fn %{backend: backend} -> + # Give every key a real historical token; no invalid-token stand-ins. + model = + Enum.reduce(["a", "b", "c"], %{}, fn key, model -> + model + |> step({:claim, key, 0}, backend) + |> step({:delete, key, 0}, backend) + end) + + # Guarantee that even an empty/shrunk sequence exercises an old delete + # against a recreated object, as well as an old write. + commands = + [{:claim, "a", 1}, {:delete, "a", 1}, {:write, "a", 2, 1}] ++ commands + + Enum.reduce(commands, model, &step(&2, &1, backend)) + end) + end + end + + test "a failed sample is stopped and removed before the next sample", %{root: root} do + assert_raise RuntimeError, "deliberate sample failure", fn -> + PropertyFixture.with_sample(root, :storage, fn %{backend: backend} -> + assert {:ok, _} = StorageBackend.put_object(backend, "leftover", 1) + raise "deliberate sample failure" + end) + end + + assert File.ls!(root) == [] + + PropertyFixture.with_sample(root, :storage, fn %{backend: backend} -> + assert {:error, :not_found} = + StorageBackend.get_object(backend, "leftover", consistent: true) + end) + + assert File.ls!(root) == [] + end + + defp command do + key = member_of(["a", "b", "c"]) + value = integer(-100..100) + token = integer(0..8) + + frequency([ + {3, tuple({constant(:claim), key, value})}, + {4, tuple({constant(:write), key, value, token})}, + {3, tuple({constant(:delete), key, token})}, + {1, tuple({constant(:read), key})} + ]) + end + + # The reference state uses logical generations, not EKV versions or ETag + # equality. ETags are opaque handles used only to invoke the real backend. + defp step(model, command, backend) do + key = elem(command, 1) + object = Map.get(model, key, %{current: nil, history: [], generation: 0}) + object = execute(command, object, backend) + model = Map.put(model, key, object) + + # Check every key, not only the one just touched (cross-key corruption). + for {key, expected} <- model do + actual = StorageBackend.get_object(backend, key, consistent: true) + + case expected.current do + nil -> + assert actual == {:error, :not_found} + + %{value: value, etag: etag} -> + assert actual == {:ok, %{body: value, etag: etag}} + end + end + + model + end + + defp execute({:claim, key, value}, object, backend) do + result = StorageBackend.try_claim(backend, key, value) + + if object.current do + assert result == {:error, :already_claimed} + object + else + assert {:ok, {:claimed, etag}} = result + remember(object, value, etag) + end + end + + defp execute({:write, key, value, index}, object, backend) do + token = Enum.at(object.history, rem(index, length(object.history))) + result = StorageBackend.put_object(backend, key, value, etag: token.etag, max_retries: 0) + + if object.current && object.current.generation == token.generation do + assert {:ok, %{body: ^value, etag: etag}} = result + remember(object, value, etag) + else + assert result == {:error, :conflict} + object + end + end + + defp execute({:delete, key, index}, object, backend) do + token = Enum.at(object.history, rem(index, length(object.history))) + result = StorageBackend.delete_object(backend, key, etag: token.etag) + + cond do + object.current == nil -> + assert result == {:error, :not_found} + object + + object.current.generation == token.generation -> + assert result == :ok + %{object | current: nil} + + true -> + assert result == {:error, :conflict} + object + end + end + + defp execute({:read, _key}, object, _backend), do: object + + defp remember(object, value, etag) do + refute Enum.any?(object.history, &(&1.etag == etag)), "an old ownership token was reused" + next = %{value: value, etag: etag, generation: object.generation + 1} + %{object | current: next, generation: next.generation, history: [next | object.history]} + end +end diff --git a/test/support/property_fixture.ex b/test/support/property_fixture.ex new file mode 100644 index 0000000..8123cff --- /dev/null +++ b/test/support/property_fixture.ex @@ -0,0 +1,54 @@ +defmodule DurableServer.PropertyFixture do + @moduledoc false + import ExUnit.Assertions + import ExUnit.Callbacks + + alias DurableServer.Backends.EKVStore + alias DurableServer.StorageBackend + + # These names are deliberately reused: properties are synchronous, and allocating + # atoms for every generated input/shrink would leak atoms in long campaigns. + @store :durable_property_ekv + @supervisor :durable_property_supervisor + + def with_sample(root, kind, fun) do + dir = Path.join(root, Integer.to_string(System.unique_integer([:positive, :monotonic]))) + + children = + [{EKV, name: @store, data_dir: dir, cluster_size: 1, node_id: 1, log: false}] ++ + durable_children(kind) + + try do + start_supervised!(%{ + id: __MODULE__, + start: {Supervisor, :start_link, [children, [strategy: :one_for_one]]}, + type: :supervisor, + shutdown: :infinity + }) + + {:ok, backend} = StorageBackend.init_backend(EKVStore, name: @store) + fun.(%{backend: backend, supervisor: @supervisor}) + after + # Reverse child shutdown order keeps EKV alive for final durable writes. + # This runs for every input AND shrink, including assertion failures. + stopped = stop_supervised(__MODULE__) + assert stopped in [:ok, {:error, :not_found}] + File.rm_rf!(dir) + end + end + + defp durable_children(:storage), do: [] + + defp durable_children(:durable) do + [ + {DurableServer.Supervisor, + name: @supervisor, + prefix: "property/", + backend: {EKVStore, name: @store, start: false}, + initial_discovery_delay_ms: 60_000, + discovery_interval_ms: 60_000, + discovery_burst_count: 0, + graceful_shutdown_timeout_ms: 500} + ] + end +end