Model storage fencing and crash durability with StreamData

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:

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

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

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"},
 }

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

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