diff --git a/vav_pipeline/lib/vav_pipeline/manna_ledger.ex b/vav_pipeline/lib/vav_pipeline/manna_ledger.ex index 4cdb3fd8..2eadd910 100644 --- a/vav_pipeline/lib/vav_pipeline/manna_ledger.ex +++ b/vav_pipeline/lib/vav_pipeline/manna_ledger.ex @@ -20,6 +20,27 @@ defmodule VavPipeline.MannaLedger do Architecture §4, Step 26: Manna anti-hoarding enforcement and cycle-boundary surplus release. + + ## What is eaten stays eaten + + The day's consumption is written to `ledger.json` beside the provision + file, after every cycle that eats. A ledger that lives only in memory + forgets what was eaten whenever the process restarts: the same day's + provision is gathered again, whole. While manna was an emptyDir that + needed the kohen's hand; once the provision was put on a volume + (2026-10-04) a crash loop alone would have refilled the jar, as often + as it crashed. *Ish l'fi ochlo* — each according to what he eats + (Shemot 16:18): the omer is measured once. + + - On the first load of a day — and at once when the process starts — + if `ledger.json` carries TODAY's date, consumption resumes from it. Another day's file is not this day's + record and is ignored. + - At the day boundary the file is cleared with the surplus. A release + within the same day clears the allocation but not the record. + - The file is a convenience of the vessel, never a source of manna: + it can only reduce what remains. If it cannot be read or written the + ledger keeps counting in memory and says so; it does not stop the + cycle. """ use GenServer @@ -129,13 +150,23 @@ defmodule VavPipeline.MannaLedger do end @doc """ - Reset the ledger to initial state. For testing only. + Reset the ledger to initial state, and forget the persisted day. + For testing only. """ @spec reset() :: :ok def reset do GenServer.call(__MODULE__, :reset) end + @doc """ + Where the day's consumption is kept: `ledger.json`, beside the + provision file (on the manna volume in production). + """ + @spec ledger_file() :: String.t() + def ledger_file do + Path.join(Path.dirname(Manna.provision_file()), "ledger.json") + end + # ============================================================ # GenServer callbacks # ============================================================ @@ -152,11 +183,30 @@ defmodule VavPipeline.MannaLedger do } Logger.info("MannaLedger: ledger ready — d'var yom b'yomo") - {:ok, state} + {:ok, state, {:continue, :resume_day}} + end + + # A process that starts in the middle of a day whose record is on disk + # takes the day up at once, rather than at its first cycle — so that + # /kodesh/status never shows an untouched jar that is in fact half + # eaten. With no record for today there is nothing to resume, and the + # first cycle gathers as it always did. + @impl true + def handle_continue(:resume_day, state) do + today = Date.utc_today() |> Date.to_iso8601() + + if persisted_date() == today do + {_reply, new_state} = load_for(today, state) + {:noreply, new_state} + else + {:noreply, state} + end end @impl true def handle_call(:reset, _from, _state) do + forget_persisted() + {:reply, :ok, %{ cycle_date: nil, @@ -175,40 +225,8 @@ defmodule VavPipeline.MannaLedger do if state.cycle_date == today do {:reply, {:ok, :already_loaded}, state} else - case Manna.check() do - {:ok, :no_provision} -> - new_state = %{ - state - | cycle_date: today, - allocated: @zero_provision, - consumed: @zero_provision, - cycle_count: state.cycle_count + 1 - } - - Logger.warning("MannaLedger: no provision for #{today} — system will wait") - {:reply, {:ok, :no_provision}, new_state} - - {:ok, manna} -> - new_state = %{ - state - | cycle_date: today, - allocated: manna.provision, - consumed: @zero_provision, - cycle_count: state.cycle_count + 1 - } - - Logger.info( - "MannaLedger: provision loaded for #{today} (#{manna.type}) — " <> - "compute: #{manna.provision.compute_units}, " <> - "api_calls: #{manna.provision.api_calls_external}, " <> - "panels: #{manna.provision.sanhedrin_panels_max}" - ) - - {:reply, {:ok, :loaded}, new_state} - - {:error, reason} -> - {:reply, {:error, reason}, state} - end + {reply, new_state} = load_for(today, state) + {:reply, reply, new_state} end end @@ -243,6 +261,7 @@ defmodule VavPipeline.MannaLedger do new_consumed = Map.put(state.consumed, resource, consumed + amount) new_remaining = remaining - amount new_state = %{state | consumed: new_consumed} + persist(new_state) # Signal when provision runs low (< 10% remaining) if new_remaining < allocated * 0.1 and new_remaining > 0 do @@ -303,6 +322,155 @@ defmodule VavPipeline.MannaLedger do hoarding_violations: state.hoarding_violations } + # The day's record goes with the day's surplus — when the day is + # over. A release called while it is still the ledger's day (the + # late-provision re-gather in Manna.gather_for_cycle/0) lets go of + # the allocation, not of the memory of what was eaten: the reload + # that follows resumes today's count. + if state.cycle_date != (Date.utc_today() |> Date.to_iso8601()) do + forget_persisted() + end + {:reply, {:ok, %{released: released}}, new_state} end + + # ============================================================ + # Loading + # ============================================================ + + # Gather `today`'s provision into the ledger, resuming today's + # consumption if this process is not the first to gather it. + defp load_for(today, state) do + case Manna.check() do + {:ok, :no_provision} -> + new_state = %{ + state + | cycle_date: today, + allocated: @zero_provision, + consumed: @zero_provision, + cycle_count: state.cycle_count + 1 + } + + Logger.warning("MannaLedger: no provision for #{today} — system will wait") + {{:ok, :no_provision}, new_state} + + {:ok, manna} -> + # What was already eaten today, if this process is not the + # first to gather it. + consumed = restore_consumed(today) + + new_state = %{ + state + | cycle_date: today, + allocated: manna.provision, + consumed: consumed, + cycle_count: state.cycle_count + 1 + } + + Logger.info( + "MannaLedger: provision loaded for #{today} (#{manna.type}) — " <> + "compute: #{manna.provision.compute_units}, " <> + "api_calls: #{manna.provision.api_calls_external}, " <> + "panels: #{manna.provision.sanhedrin_panels_max}" <> + if(consumed == @zero_provision, + do: "", + else: " — resuming the day: already consumed #{inspect(consumed)}" + ) + ) + + {{:ok, :loaded}, new_state} + + {:error, reason} -> + {{:error, reason}, state} + end + end + + # ============================================================ + # Persistence — ledger.json beside the provision + # ============================================================ + + # Written whole and renamed into place, so a reader (or a crash) never + # meets half a file. + defp persist(state) do + path = ledger_file() + tmp = path <> ".tmp" + + body = + Jason.encode!(%{ + "cycle_date" => state.cycle_date, + "consumed" => Map.new(state.consumed, fn {k, v} -> {Atom.to_string(k), v} end), + "written_at" => DateTime.utc_now() |> DateTime.to_iso8601() + }) + + with :ok <- File.write(tmp, body), + :ok <- File.rename(tmp, path) do + :ok + else + {:error, reason} -> + Logger.warning( + "MannaLedger: could not persist consumption to #{path} (#{inspect(reason)}) — " <> + "counting in memory only; a restart today would forget what was eaten" + ) + + :ok + end + end + + # Today's consumption from an earlier life of this process, or zeros. + defp restore_consumed(today) do + path = ledger_file() + + with {:ok, contents} <- File.read(path), + {:ok, %{"cycle_date" => ^today, "consumed" => consumed}} when is_map(consumed) <- + Jason.decode(contents), + {:ok, restored} <- parse_consumed(consumed) do + restored + else + {:error, :enoent} -> + @zero_provision + + {:ok, %{"cycle_date" => other}} -> + Logger.info("MannaLedger: #{path} is for #{inspect(other)}, not #{today} — not this day's record") + @zero_provision + + other -> + Logger.warning( + "MannaLedger: #{path} is unreadable (#{inspect(other)}) — " <> + "starting today's count from zero" + ) + + @zero_provision + end + end + + # Only the five known resources, only non-negative integers. Anything + # else is not a ledger. + defp parse_consumed(consumed) do + Enum.reduce_while(@zero_provision, {:ok, %{}}, fn {key, _}, {:ok, acc} -> + case Map.get(consumed, Atom.to_string(key), 0) do + n when is_integer(n) and n >= 0 -> {:cont, {:ok, Map.put(acc, key, n)}} + bad -> {:halt, {:error, {:bad_consumed, key, bad}}} + end + end) + end + + # The date the record on disk is for, or nil. + defp persisted_date do + with {:ok, contents} <- File.read(ledger_file()), + {:ok, %{"cycle_date" => date}} when is_binary(date) <- Jason.decode(contents) do + date + else + _ -> nil + end + end + + defp forget_persisted do + case File.rm(ledger_file()) do + :ok -> :ok + {:error, :enoent} -> :ok + {:error, reason} -> + Logger.warning("MannaLedger: could not clear #{ledger_file()} (#{inspect(reason)})") + :ok + end + end end diff --git a/vav_pipeline/test/manna_ledger_persistence_test.exs b/vav_pipeline/test/manna_ledger_persistence_test.exs new file mode 100644 index 00000000..86d0f2a7 --- /dev/null +++ b/vav_pipeline/test/manna_ledger_persistence_test.exs @@ -0,0 +1,233 @@ +defmodule VavPipeline.MannaLedgerPersistenceTest do + @moduledoc """ + What is eaten stays eaten. + + Until 2026-10-04 the ledger lived only in memory. A restart forgot the + day's consumption and the next cycle gathered the whole provision again. + That was hidden while manna was an emptyDir (a restart emptied the jar + too, and the kohen had to refill it by hand); once the provision was put + on a volume, a crash loop alone would have refilled it. + + These tests restart the ledger process under its real supervisor — a new + process, an empty state, the same disk — and ask what it remembers. + """ + + use ExUnit.Case, async: false + + alias VavPipeline.{HaltMonitor, Manna, MannaLedger} + + setup do + if HaltMonitor.halted?(), do: HaltMonitor.resolve("manna persistence test setup") + MannaLedger.reset() + write_provision(10) + + on_exit(fn -> + MannaLedger.reset() + File.rm(Manna.provision_file()) + end) + + :ok + end + + describe "the record" do + test "is written beside the provision, with today's date, when a cycle eats" do + refute File.exists?(MannaLedger.ledger_file()) + + assert {:ok, :loaded} = MannaLedger.load_provision() + assert {:ok, 7} = MannaLedger.consume(:compute_units, 3) + + assert Path.dirname(MannaLedger.ledger_file()) == Path.dirname(Manna.provision_file()) + record = MannaLedger.ledger_file() |> File.read!() |> Jason.decode!() + + assert record["cycle_date"] == today() + assert record["consumed"]["compute_units"] == 3 + assert record["consumed"]["api_calls_external"] == 0 + end + + test "is not written by a refused consumption" do + assert {:ok, :loaded} = MannaLedger.load_provision() + assert {:error, :provision_exhausted} = MannaLedger.consume(:compute_units, 11) + refute File.exists?(MannaLedger.ledger_file()) + end + end + + describe "a restart in the middle of the day" do + test "resumes the count — the provision is not gathered whole again" do + assert {:ok, :loaded} = MannaLedger.load_provision() + assert {:ok, 6} = MannaLedger.consume(:compute_units, 4) + + restart_ledger() + + # The new process takes the day up as it starts — before any cycle + # asks — so the status never shows an untouched jar. + ledger = MannaLedger.ledger() + assert ledger.cycle_date == today() + assert ledger.allocated.compute_units == 10 + assert ledger.consumed.compute_units == 4 + + assert {:ok, :already_loaded} = MannaLedger.load_provision() + assert MannaLedger.remaining(:compute_units) == 6 + + # And it keeps counting from there. + assert {:ok, 5} = MannaLedger.consume(:compute_units, 1) + end + + test "an exhausted day stays exhausted — a crash loop does not refill the jar" do + assert {:ok, :loaded} = MannaLedger.load_provision() + assert {:ok, 0} = MannaLedger.consume(:compute_units, 10) + + for _ <- 1..3 do + restart_ledger() + assert {:waiting, :provision_exhausted} = Manna.gather_for_cycle() + end + end + + test "the whole cycle path: gather, restart, gather" do + assert :ok = Manna.gather_for_cycle() + assert :ok = Manna.gather_for_cycle() + restart_ledger() + assert :ok = Manna.gather_for_cycle() + assert MannaLedger.ledger().consumed.compute_units == 3 + end + end + + describe "a restart with nothing to resume" do + test "no record: the new process holds no day until the first cycle" do + restart_ledger() + assert MannaLedger.ledger().cycle_date == nil + assert {:ok, :loaded} = MannaLedger.load_provision() + assert MannaLedger.remaining(:compute_units) == 10 + end + + test "a record but no provision: nothing is resumed into an empty day" do + assert {:ok, :loaded} = MannaLedger.load_provision() + assert {:ok, 9} = MannaLedger.consume(:compute_units, 1) + File.rm!(Manna.provision_file()) + + restart_ledger() + assert MannaLedger.remaining(:compute_units) == 0 + + # The vessel waits. (Which of the two waiting reasons it gives is + # the ledger's older behaviour for a day pinned with no provision; + # the router answers MANNA_DEPLETED for both.) + assert {:waiting, reason} = Manna.gather_for_cycle() + assert reason in [:no_provision, :provision_exhausted] + end + end + + describe "what the record may not do" do + test "another day's record is not this day's — it is ignored" do + write_record(%{"cycle_date" => "2026-01-01", "consumed" => %{"compute_units" => 9}}) + + assert {:ok, :loaded} = MannaLedger.load_provision() + assert MannaLedger.remaining(:compute_units) == 10 + end + + test "a record that is not a ledger is ignored, and the cycle goes on" do + for garbage <- [ + "not json", + Jason.encode!(%{"cycle_date" => today(), "consumed" => "all of it"}), + Jason.encode!(%{"cycle_date" => today(), "consumed" => %{"compute_units" => -5}}), + Jason.encode!(%{"cycle_date" => today(), "consumed" => %{"compute_units" => "3"}}) + ] do + MannaLedger.reset() + File.write!(MannaLedger.ledger_file(), garbage) + + assert {:ok, :loaded} = MannaLedger.load_provision() + assert MannaLedger.remaining(:compute_units) == 10 + assert {:ok, 9} = MannaLedger.consume(:compute_units, 1) + end + end + + test "it can only take away: a record larger than the provision leaves nothing" do + write_record(%{"cycle_date" => today(), "consumed" => %{"compute_units" => 500}}) + + assert {:ok, :loaded} = MannaLedger.load_provision() + assert MannaLedger.remaining(:compute_units) == 0 + assert {:error, :provision_exhausted} = MannaLedger.consume(:compute_units, 1) + end + end + + describe "release" do + test "at a day boundary the record goes with the surplus" do + # The ledger holds no day (cycle_date nil ≠ today), as it does when + # the day it loaded has passed and been released. + write_record(%{"cycle_date" => "2026-01-01", "consumed" => %{"compute_units" => 2}}) + + assert {:ok, _} = MannaLedger.enforce_cycle_boundary() + refute File.exists?(MannaLedger.ledger_file()) + end + + test "a release within the same day lets go of the allocation, not the memory" do + assert {:ok, :loaded} = MannaLedger.load_provision() + assert {:ok, 8} = MannaLedger.consume(:compute_units, 2) + + # Manna.gather_for_cycle/0 does this when a provision arrives late. + assert {:ok, _} = MannaLedger.enforce_cycle_boundary() + assert File.exists?(MannaLedger.ledger_file()) + + assert {:ok, :loaded} = MannaLedger.load_provision() + assert MannaLedger.remaining(:compute_units) == 8 + end + + test "reset forgets the record (tests must not leak a day into each other)" do + assert {:ok, :loaded} = MannaLedger.load_provision() + assert {:ok, 9} = MannaLedger.consume(:compute_units, 1) + assert File.exists?(MannaLedger.ledger_file()) + + MannaLedger.reset() + refute File.exists?(MannaLedger.ledger_file()) + end + end + + # ------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------ + + defp today, do: Date.utc_today() |> Date.to_iso8601() + + # Bring the ledger down and up again through its own supervisor — a + # new process with a new, empty state, and the file on disk untouched. + # That is a restart, not a reset. + # + # (Done with terminate_child/restart_child rather than Process.exit: + # this file restarts the ledger six times in well under five seconds, + # and a supervisor that sees more than three CRASHES in five seconds + # gives up and takes the application down with it — as the first + # version of this test found out.) + defp restart_ledger do + old = Process.whereis(MannaLedger) + :ok = Supervisor.terminate_child(VavPipeline.Supervisor, MannaLedger) + {:ok, new} = Supervisor.restart_child(VavPipeline.Supervisor, MannaLedger) + assert new != old + assert Process.whereis(MannaLedger) == new + end + + defp write_record(record) do + File.mkdir_p!(Path.dirname(MannaLedger.ledger_file())) + File.write!(MannaLedger.ledger_file(), Jason.encode!(record)) + end + + defp write_provision(compute_units) do + expires = DateTime.utc_now() |> DateTime.add(6 * 3600) |> DateTime.to_iso8601() + + provision = %{ + "date" => today(), + "provision" => %{ + "compute_units" => compute_units, + "memory_mb" => 512, + "api_calls_external" => 5, + "storage_delta_mb" => 2, + "sanhedrin_panels_max" => 2 + }, + "type" => "regular", + "expires_at" => expires, + "written_by" => "manna_ledger_persistence_test", + "written_at" => DateTime.utc_now() |> DateTime.to_iso8601() + } + + path = Manna.provision_file() + File.mkdir_p!(Path.dirname(path)) + File.write!(path, Jason.encode!(provision)) + end +end diff --git a/vav_pipeline/test/test_helper.exs b/vav_pipeline/test/test_helper.exs index 27f3f356..858379e6 100644 --- a/vav_pipeline/test/test_helper.exs +++ b/vav_pipeline/test/test_helper.exs @@ -2,6 +2,10 @@ # Run with: SHORASHIM_L3=1 mix test --include l3 ExUnit.start(exclude: [:l3]) +# The manna ledger resumes a day it finds on disk (ledger.json). A run +# that was interrupted could leave one behind; no test should inherit it. +if Process.whereis(VavPipeline.MannaLedger), do: VavPipeline.MannaLedger.reset() + if System.get_env("SHORASHIM_L3") do {:ok, _} = VavPipeline.Repo.start_link() Ecto.Adapters.SQL.Sandbox.mode(VavPipeline.Repo, :manual)