diff --git a/.github/workflows/verify.yml b/.github/workflows/verify.yml index 7cd45b3..70665e1 100644 --- a/.github/workflows/verify.yml +++ b/.github/workflows/verify.yml @@ -47,11 +47,12 @@ jobs: # Live integration tests against a real provider. Several real bugs in v1 # prep (streaming tool calls dropped, multi-send losing assistant history) # shipped past unit tests because the mocks didn't match real provider - # behavior. This costs API tokens, so PRs run unit verification only; main, - # release branch, and tag pushes must have the Anthropic secret configured. + # behavior. Runs on PRs as well as main/release/tag pushes: every PR here + # is a same-repo branch (secrets available, no fork risk), the suite costs + # about a minute of Haiku, and a gate belongs before the merge, not after. + # The Anthropic secret must be configured; the job hard-fails without it. live: runs-on: ubuntu-latest - if: github.event_name == 'push' steps: - name: Checkout diff --git a/lib/cantrip.ex b/lib/cantrip.ex index a7224a5..584a3eb 100644 --- a/lib/cantrip.ex +++ b/lib/cantrip.ex @@ -1513,7 +1513,13 @@ defmodule Cantrip do defp validate_mnesia_storage_opts(opts) when is_map(opts) or is_list(opts) do opts = opts |> normalize_input_map() |> prefer_atom_keys() - case NimbleOptions.validate(Map.to_list(opts), table: [type: :atom], mnesia: [type: :atom]) do + schema = [ + table: [type: :atom], + mnesia: [type: :atom], + on_corrupt: [type: {:in, [:quarantine]}] + ] + + case NimbleOptions.validate(Map.to_list(opts), schema) do {:ok, validated} -> {:ok, Map.new(validated)} {:error, %NimbleOptions.ValidationError{message: msg}} -> {:error, msg} end diff --git a/lib/cantrip/loom.ex b/lib/cantrip/loom.ex index 7fb04eb..a817dbe 100644 --- a/lib/cantrip/loom.ex +++ b/lib/cantrip/loom.ex @@ -113,6 +113,10 @@ defmodule Cantrip.Loom do * The storage backend's prerequisites aren't met (e.g. disk path is unwritable, Mnesia schema not created on this node) + * A prior BEAM died while Mnesia was writing and left a + corrupt transaction log such as LATEST.LOG. Preserve the + old store for forensics and start fresh by passing + `on_corrupt: :quarantine` in the Mnesia loom options. If you want to allow falling back to in-memory loom, do not pass `:loom_storage` (or pass `nil`) when constructing the diff --git a/lib/cantrip/loom/storage/mnesia.ex b/lib/cantrip/loom/storage/mnesia.ex index 38c0ac1..dae787d 100644 --- a/lib/cantrip/loom/storage/mnesia.ex +++ b/lib/cantrip/loom/storage/mnesia.ex @@ -16,7 +16,8 @@ defmodule Cantrip.Loom.Storage.Mnesia do mnesia = Map.get(opts, :mnesia, :mnesia) case with_schema_lock(fn -> - with :ok <- ensure_mnesia_started(mnesia), + with :ok <- preflight_store(opts, mnesia), + :ok <- ensure_mnesia_started(mnesia), :ok <- ensure_table(table, mnesia) do {:ok, %{table: table, mnesia: mnesia}} end @@ -33,7 +34,7 @@ defmodule Cantrip.Loom.Storage.Mnesia do key = System.unique_integer([:positive, :monotonic]) event = storage_event(%{type: :turn, turn: turn}) - case call(mnesia, :transaction, [fn -> call(mnesia, :write, [{table, key, event}]) end]) do + case durable_transaction(mnesia, fn -> call(mnesia, :write, [{table, key, event}]) end) do {:atomic, :ok} -> {:ok, state} {:aborted, reason} -> {:error, reason} other -> {:error, other} @@ -46,7 +47,7 @@ defmodule Cantrip.Loom.Storage.Mnesia do key = System.unique_integer([:positive, :monotonic]) event = storage_event(%{type: :reward, index: index, reward: reward}) - case call(mnesia, :transaction, [fn -> call(mnesia, :write, [{table, key, event}]) end]) do + case durable_transaction(mnesia, fn -> call(mnesia, :write, [{table, key, event}]) end) do {:atomic, :ok} -> {:ok, state} {:aborted, reason} -> {:error, reason} other -> {:error, other} @@ -59,7 +60,7 @@ defmodule Cantrip.Loom.Storage.Mnesia do key = System.unique_integer([:positive, :monotonic]) event = storage_event(event) - case call(mnesia, :transaction, [fn -> call(mnesia, :write, [{table, key, event}]) end]) do + case durable_transaction(mnesia, fn -> call(mnesia, :write, [{table, key, event}]) end) do {:atomic, :ok} -> {:ok, state} {:aborted, reason} -> {:error, reason} other -> {:error, other} @@ -128,6 +129,14 @@ defmodule Cantrip.Loom.Storage.Mnesia do end end + defp durable_transaction(mnesia, fun) when is_function(fun, 0) do + if function_exported?(mnesia, :sync_transaction, 1) do + call(mnesia, :sync_transaction, [fun]) + else + call(mnesia, :transaction, [fun]) + end + end + defp ensure_mnesia_started(mnesia) do case call(mnesia, :system_info, [:is_running]) do :yes -> @@ -145,6 +154,127 @@ defmodule Cantrip.Loom.Storage.Mnesia do end end + defp preflight_store(opts, mnesia) do + case call(mnesia, :system_info, [:is_running]) do + :yes -> + :ok + + _ -> + dir = mnesia_dir() + + case corrupt_log_files(dir) do + [] -> + :ok + + corrupt -> + maybe_quarantine_corrupt_store(dir, corrupt, opts) + end + end + rescue + _ -> :ok + catch + :exit, _ -> :ok + end + + defp mnesia_dir do + case Application.get_env(:mnesia, :dir) do + nil -> ~c"Mnesia.#{node()}" + dir -> dir + end + |> to_string() + end + + defp corrupt_log_files(dir) do + ["LATEST.LOG", "PREVIOUS.LOG"] + |> Enum.map(&Path.join(dir, &1)) + |> Enum.filter(&File.regular?/1) + |> Enum.flat_map(fn path -> + case probe_log(path) do + :ok -> [] + {:error, reason} -> [{path, reason}] + end + end) + end + + defp probe_log(path) do + name = :"#{__MODULE__}.probe.#{System.unique_integer([:positive])}" + + case :disk_log.open(name: name, file: String.to_charlist(path), mode: :read_only) do + {:ok, ^name} -> + :disk_log.close(name) + :ok + + {:repaired, ^name, _recovered, _badbytes} -> + :disk_log.close(name) + :ok + + {:error, {:need_repair, ^name}} -> + :disk_log.close(name) + :ok + + {:error, {:not_a_log_file, _} = reason} -> + {:error, reason} + + {:error, reason} -> + {:error, reason} + end + end + + defp maybe_quarantine_corrupt_store(dir, corrupt, opts) do + case Map.get(opts, :on_corrupt) || Map.get(opts, "on_corrupt") do + :quarantine -> + quarantine_corrupt_store(dir, corrupt) + + "quarantine" -> + quarantine_corrupt_store(dir, corrupt) + + _ -> + {:error, + {:corrupt_mnesia_store, + %{ + dir: dir, + corrupt_logs: corrupt, + repair: "pass on_corrupt: :quarantine to move this store aside and start a fresh loom" + }}} + end + end + + defp quarantine_corrupt_store(dir, corrupt) do + quarantine_dir = quarantine_dir(dir) + do_quarantine_corrupt_store(dir, quarantine_dir, corrupt) + end + + defp do_quarantine_corrupt_store(dir, quarantine_dir, corrupt) do + with :ok <- File.mkdir_p!(Path.dirname(quarantine_dir)), + :ok <- File.rename(dir, quarantine_dir), + :ok <- File.mkdir_p(dir) do + :ok + else + {:error, reason} -> + {:error, + {:corrupt_mnesia_store_quarantine_failed, + %{dir: dir, quarantine_dir: quarantine_dir, corrupt_logs: corrupt, reason: reason}}} + end + rescue + e -> + {:error, + {:corrupt_mnesia_store_quarantine_failed, + %{ + dir: dir, + quarantine_dir: quarantine_dir, + corrupt_logs: corrupt, + reason: Exception.message(e) + }}} + end + + defp quarantine_dir(dir) do + stamp = + DateTime.utc_now() + |> Calendar.strftime("%Y%m%dT%H%M%SZ") + + "#{dir}-corrupt-#{stamp}-#{System.unique_integer([:positive])}" + end + defp ensure_schema(mnesia) do case call(mnesia, :create_schema, [[node()]]) do :ok -> :ok diff --git a/lib/mix/tasks/cantrip.familiar.ex b/lib/mix/tasks/cantrip.familiar.ex index 4691081..924b548 100644 --- a/lib/mix/tasks/cantrip.familiar.ex +++ b/lib/mix/tasks/cantrip.familiar.ex @@ -165,7 +165,7 @@ defmodule Mix.Tasks.Cantrip.Familiar do case :net_kernel.start([name, :longnames]) do {:ok, _} -> :erlang.set_cookie(node(), cookie) - configure_mnesia_dir!(workspace_root) + configure_mnesia_workspace!(workspace_root) {:error, {:already_started, _}} -> :ok @@ -190,15 +190,17 @@ defmodule Mix.Tasks.Cantrip.Familiar do _named -> # Already named (someone launched with --sname/--name). Trust # their setup; just relocate Mnesia under .cantrip/. - configure_mnesia_dir!(workspace_root) + configure_mnesia_workspace!(workspace_root) end end # Point Mnesia at `.cantrip/mnesia/` for this workspace. Mnesia is # in `included_applications` (not `extra_applications`), so it's - # loaded but not yet started. Setting `:dir` before the adapter's - # lazy `:mnesia.start/0` is enough — no stop/restart cycle, no - # orphaned `Mnesia./` dir at cwd from a premature auto-start. + # loaded but not yet started. Setting `:dir` and related durability + # knobs before the adapter's lazy `:mnesia.start/0` is enough — no + # stop/restart cycle, no orphaned `Mnesia./` dir at cwd from a + # premature auto-start, and no repo-root `MnesiaCore.*` files if + # Mnesia fatals while recovering a transaction log. # # Verified empirically: after `mix run`, `Application.started_applications/0` # does not include `:mnesia`, and `:mnesia.system_info(:tables)` @@ -207,13 +209,91 @@ defmodule Mix.Tasks.Cantrip.Familiar do # started with the parent" concern doesn't apply here because # `Cantrip.Application.start/2` never calls `Application.ensure_*` # on Mnesia. - defp configure_mnesia_dir!(workspace_root) do - desired = Path.join([workspace_root, ".cantrip", "mnesia"]) |> String.to_charlist() - File.mkdir_p!(to_string(desired)) - Application.put_env(:mnesia, :dir, desired) + @doc false + @spec configure_mnesia_workspace!(String.t()) :: :ok + def configure_mnesia_workspace!(workspace_root) when is_binary(workspace_root) do + env = mnesia_env_for_workspace(workspace_root) + + env + |> Keyword.fetch!(:dir) + |> to_string() + |> File.mkdir_p!() + + env + |> Keyword.fetch!(:core_dir) + |> to_string() + |> File.mkdir_p!() + + ensure_mnesia_configurable!(env) + + for {key, value} <- env do + Application.put_env(:mnesia, key, value) + end + :ok end + @doc false + @spec mnesia_env_for_workspace(String.t()) :: keyword() + def mnesia_env_for_workspace(workspace_root) when is_binary(workspace_root) do + mnesia_dir = Path.join([workspace_root, ".cantrip", "mnesia"]) + + [ + dir: String.to_charlist(mnesia_dir), + core_dir: String.to_charlist(Path.join(mnesia_dir, "cores")), + auto_repair: true, + dc_dump_limit: 16, + dump_log_write_threshold: 100, + dump_log_time_threshold: 30_000 + ] + end + + defp ensure_mnesia_configurable!(env) do + if mnesia_running?() do + mismatches = + env + |> Enum.flat_map(fn {key, expected} -> + info_key = if key == :dir, do: :directory, else: key + actual = :mnesia.system_info(info_key) + + if actual == expected do + [] + else + [{key, expected, actual}] + end + end) + + if mismatches != [] do + raise """ + Mnesia is already running with workspace settings that do not match this Familiar. + + #{format_mnesia_mismatches(mismatches)} + + Restart the BEAM before launching the workspace Familiar, or opt out of + Mnesia with: + + mix cantrip.familiar --loom-path .cantrip/familiar.jsonl + """ + end + end + end + + defp mnesia_running? do + Code.ensure_loaded?(:mnesia) and :mnesia.system_info(:is_running) == :yes + rescue + _ -> false + catch + :exit, _ -> false + end + + defp format_mnesia_mismatches(mismatches) do + mismatches + |> Enum.map(fn {key, expected, actual} -> + " * #{key}: expected #{inspect(expected)}, got #{inspect(actual)}" + end) + |> Enum.join("\n") + end + # `System.cmd("epmd", ["-daemon"], ...)` raises `ErlangError` when # epmd is not on PATH. Catching here keeps the actionable # `--loom-path` error message in `ensure_named_node!` reachable diff --git a/test/loom_backend_symmetry_test.exs b/test/loom_backend_symmetry_test.exs index 7c7e1f9..e2133f6 100644 --- a/test/loom_backend_symmetry_test.exs +++ b/test/loom_backend_symmetry_test.exs @@ -84,4 +84,97 @@ defmodule Cantrip.LoomBackendSymmetryTest do "#{inspect(module)} does not implement load/1" end end + + test "Mnesia backend names corrupt transaction logs instead of surfacing generic init causes" do + with_mnesia_dir("loom_mnesia_corrupt", fn dir -> + File.write!(Path.join(dir, "LATEST.LOG"), "not an erlang disk log\n") + + error = + assert_raise RuntimeError, fn -> + Loom.new(%{identity: "test"}, storage: {:mnesia, %{table: :loom_mnesia_corrupt}}) + end + + assert error.message =~ "corrupt_mnesia_store" + assert error.message =~ "LATEST.LOG" + assert error.message =~ "on_corrupt: :quarantine" + assert File.exists?(Path.join(dir, "LATEST.LOG")) + end) + end + + test "Mnesia backend can quarantine a corrupt store and start a fresh loom" do + table = :"loom_mnesia_quarantine_#{System.unique_integer([:positive])}" + + try do + with_mnesia_dir("loom_mnesia_quarantine", fn dir -> + File.write!(Path.join(dir, "LATEST.LOG"), "not an erlang disk log\n") + File.write!(Path.join(dir, "forensics.marker"), "preserve me\n") + + loom = + Loom.new(%{identity: "test"}, + storage: {:mnesia, %{table: table, on_corrupt: :quarantine}} + ) + + assert loom.storage_module == Cantrip.Loom.Storage.Mnesia + assert File.dir?(dir) + refute File.exists?(Path.join(dir, "forensics.marker")) + + [quarantine_dir] = Path.wildcard(dir <> "-corrupt-*") + assert File.read!(Path.join(quarantine_dir, "LATEST.LOG")) == "not an erlang disk log\n" + assert File.read!(Path.join(quarantine_dir, "forensics.marker")) == "preserve me\n" + end) + after + try do + :mnesia.delete_table(table) + rescue + _ -> :ok + end + end + end + + defp with_mnesia_dir(prefix, fun) do + dir = + Path.join( + System.tmp_dir!(), + "#{prefix}_#{System.unique_integer([:positive])}" + ) + + keys = [:dir, :auto_repair] + old_env = Map.new(keys, &{&1, Application.fetch_env(:mnesia, &1)}) + + stop_mnesia() + File.rm_rf!(dir) + File.mkdir_p!(dir) + + try do + Application.put_env(:mnesia, :dir, String.to_charlist(dir)) + Application.put_env(:mnesia, :auto_repair, false) + fun.(dir) + after + stop_mnesia() + File.rm_rf!(dir) + + dir + |> Path.dirname() + |> Path.join(Path.basename(dir) <> "-corrupt-*") + |> Path.wildcard() + |> Enum.each(&File.rm_rf!/1) + + for {key, value} <- old_env do + case value do + {:ok, existing} -> Application.put_env(:mnesia, key, existing) + :error -> Application.delete_env(:mnesia, key) + end + end + end + end + + defp stop_mnesia do + if Code.ensure_loaded?(:mnesia) and :mnesia.system_info(:is_running) == :yes do + :mnesia.stop() + end + rescue + _ -> :ok + catch + :exit, _ -> :ok + end end diff --git a/test/loom_storage_test.exs b/test/loom_storage_test.exs index 904f9f1..a2a977e 100644 --- a/test/loom_storage_test.exs +++ b/test/loom_storage_test.exs @@ -17,6 +17,16 @@ defmodule Cantrip.LoomStorageTest do def wait_for_tables(_tables, _timeout), do: :ok end + defmodule MnesiaSyncWrites do + def sync_transaction(fun) do + Process.put(:mnesia_sync_transaction_called, true) + {:atomic, fun.()} + end + + def transaction(_fun), do: raise("append writes must use sync_transaction/1") + def write({_table, _key, _event}), do: :ok + end + defmodule FailingStorage do @behaviour Cantrip.Loom.Storage @@ -46,6 +56,14 @@ defmodule Cantrip.LoomStorageTest do Cantrip.Loom.Storage.Mnesia.init(table: :schema_exists, mnesia: MnesiaAlreadyExists) end + test "mnesia append writes use synchronous transactions when available" do + Process.delete(:mnesia_sync_transaction_called) + + state = %{table: :sync_writes, mnesia: MnesiaSyncWrites} + assert {:ok, ^state} = Cantrip.Loom.Storage.Mnesia.append_event(state, %{type: :event}) + assert Process.get(:mnesia_sync_transaction_called) == true + end + test "explicit malformed loom storage does not fall back to memory" do assert_raise ArgumentError, ~r/invalid loom storage/, fn -> Cantrip.Loom.new(%{system_prompt: nil}, storage: :jsonl) diff --git a/test/mix_cantrip_familiar_test.exs b/test/mix_cantrip_familiar_test.exs index 6bb1eee..4965c00 100644 --- a/test/mix_cantrip_familiar_test.exs +++ b/test/mix_cantrip_familiar_test.exs @@ -188,6 +188,65 @@ defmodule Mix.Tasks.Cantrip.FamiliarTest do end end + describe "Mnesia workspace environment" do + test "places Mnesia data and core dumps under .cantrip with bounded log thresholds" do + root = "/tmp/cantrip-mnesia-env" + env = Task.mnesia_env_for_workspace(root) + + assert env[:dir] == String.to_charlist(Path.join([root, ".cantrip", "mnesia"])) + + assert env[:core_dir] == + String.to_charlist(Path.join([root, ".cantrip", "mnesia", "cores"])) + + assert env[:auto_repair] == true + assert env[:dc_dump_limit] == 16 + assert env[:dump_log_write_threshold] == 100 + assert env[:dump_log_time_threshold] == 30_000 + end + + @tag :mnesia + test "configure_mnesia_workspace! applies workspace env before Mnesia starts" do + preserve_mnesia_env(fn -> + root = tmp_root("fam_mnesia_env_apply") + + try do + assert :ok = Task.configure_mnesia_workspace!(root) + env = Task.mnesia_env_for_workspace(root) + + for {key, value} <- env do + assert Application.get_env(:mnesia, key) == value + end + + assert File.dir?(to_string(env[:dir])) + assert File.dir?(to_string(env[:core_dir])) + after + File.rm_rf!(root) + end + end) + end + + @tag :mnesia + test "configure_mnesia_workspace! fails if Mnesia is already running elsewhere" do + preserve_mnesia_env(fn -> + first = tmp_root("fam_mnesia_env_first") + second = tmp_root("fam_mnesia_env_second") + + try do + assert :ok = Task.configure_mnesia_workspace!(first) + :ok = ensure_mnesia_started() + + assert_raise RuntimeError, ~r/Mnesia is already running/, fn -> + Task.configure_mnesia_workspace!(second) + end + after + stop_mnesia() + File.rm_rf!(first) + File.rm_rf!(second) + end + end) + end + end + # ===================================================================== # Workspace-stable identity for the BEAM node # ===================================================================== @@ -304,9 +363,7 @@ defmodule Mix.Tasks.Cantrip.FamiliarTest do root = #{inspect(root)} mode = #{inspect(mode_text)} sentinel = "cantrip_mnesia_restart_sentinel" - mnesia_dir = Path.join([root, ".cantrip", "mnesia"]) - File.mkdir_p!(mnesia_dir) - Application.put_env(:mnesia, :dir, String.to_charlist(mnesia_dir)) + Mix.Tasks.Cantrip.Familiar.configure_mnesia_workspace!(root) cookie = Cantrip.Familiar.Cookie.for_workspace!(root) :erlang.set_cookie(node(), cookie) @@ -362,6 +419,60 @@ defmodule Mix.Tasks.Cantrip.FamiliarTest do """ end + defp preserve_mnesia_env(fun) do + keys = [ + :dir, + :core_dir, + :auto_repair, + :dc_dump_limit, + :dump_log_write_threshold, + :dump_log_time_threshold + ] + + old_env = Map.new(keys, &{&1, Application.fetch_env(:mnesia, &1)}) + + try do + stop_mnesia() + fun.() + after + stop_mnesia() + + for key <- keys do + case Map.fetch!(old_env, key) do + {:ok, value} -> Application.put_env(:mnesia, key, value) + :error -> Application.delete_env(:mnesia, key) + end + end + end + end + + defp ensure_mnesia_started do + case :mnesia.create_schema([node()]) do + :ok -> :ok + {:error, {_kind, {:already_exists, _node}}} -> :ok + {:error, {:already_exists, _node}} -> :ok + end + + case :mnesia.start() do + :ok -> :ok + {:error, {:already_started, :mnesia}} -> :ok + end + end + + defp stop_mnesia do + if Code.ensure_loaded?(:mnesia) and :mnesia.system_info(:is_running) == :yes do + :stopped = :mnesia.stop() + end + rescue + _ -> :ok + catch + :exit, _ -> :ok + end + + defp tmp_root(prefix) do + Path.join(System.tmp_dir!(), "#{prefix}_#{System.unique_integer([:positive])}") + end + defp cleanup_epmd_name(nil), do: :ok defp cleanup_epmd_name(name) do