bikoret/vav_pipeline/lib/vav_pipeline/l1_integrity.ex

defmodule VavPipeline.L1Integrity do
  @moduledoc """
  The witness that the scroll in this vessel is the scroll that was engraved.

  Broken tablet 38 (2026-10-04). `L1Reader.verify_integrity/0` read the
  hash stored in `integrity.sha256` and compared that string with the hash
  compiled into this module's neighbour. It never opened a verse. Two
  copies of one constant were compared with each other, and on that
  comparison rested `l1_hash: pass`, the Cloud's "L1 corrupted — the cloud
  is not visible", and Construction §1.2's "if the hash does not match, the
  system does not start." A verse could have been changed and every one of
  them would have gone on saying *tov*.

  This process reads the scroll. When it starts it hashes the whole of
  `torah/` — every path and every byte, in the order `scripts/build_l1.py`
  hashed them when L1 was engraved — and it does so again at an interval
  (ten minutes unless configured). `L1Reader.verify_integrity/0` asks this
  process what it found; the answer is a comparison of what is on disk
  with what was pinned, not of the pin with itself.

  Hashing 5,859 files takes about a second (three on the production
  node), which is too long to do before every cycle and every health
  probe; so the cycle is answered from the last reading, and the reading
  is never older than the interval. Callers that arrive while the FIRST
  reading is under way wait for it — nobody is answered before the scroll
  has been read once. The periodic re-readings run beside the process,
  not in it: while one is under way the last reading goes on answering,
  and no cycle or probe waits three seconds for it.

  *Lo tosif alav v'lo tigra mimenu* — you shall not add to it and you shall
  not take from it (Devarim 13:1). The check is how the vessel knows.
  """

  use GenServer
  require Logger

  alias VavPipeline.L1Reader

  @default_interval :timer.minutes(10)

  @type reading :: %{
          status: :ok | {:error, String.t()},
          computed: String.t() | nil,
          expected: String.t(),
          read_at: DateTime.t()
        }

  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: Keyword.get(opts, :name, __MODULE__))
  end

  @doc "`:ok` if the scroll on disk hashes to the pinned hash, else `{:error, reason}`."
  @spec status(GenServer.server()) :: :ok | {:error, String.t()}
  def status(server \\ __MODULE__), do: GenServer.call(server, :status, 60_000)

  @doc "The last reading in full — what was computed, what was expected, and when."
  @spec reading(GenServer.server()) :: reading()
  def reading(server \\ __MODULE__), do: GenServer.call(server, :reading, 60_000)

  @doc "Read the scroll again now, and return what was found."
  @spec recheck(GenServer.server()) :: :ok | {:error, String.t()}
  def recheck(server \\ __MODULE__), do: GenServer.call(server, :recheck, 60_000)

  # ------------------------------------------------------------
  # GenServer
  # ------------------------------------------------------------

  @impl true
  def init(opts) do
    state = %{
      root: Keyword.get(opts, :l1_root, L1Reader.l1_root()),
      expected: Keyword.get(opts, :expected_hash, L1Reader.expected_hash()),
      interval:
        Keyword.get(
          opts,
          :interval,
          Application.get_env(:vav_pipeline, :l1_rehash_interval_ms, @default_interval)
        ),
      reading: nil
    }

    {:ok, state, {:continue, :read}}
  end

  @impl true
  def handle_continue(:read, state) do
    state = read(state)
    schedule(state)
    {:noreply, state}
  end

  # The periodic reading is done by a task, so that this process stays
  # free to answer from the last one.
  @impl true
  def handle_info(:read, state) do
    witness = self()
    root = state.root

    Task.start(fn ->
      send(witness, {:read_done, :timer.tc(fn -> L1Reader.compute_tree_hash(root) end)})
    end)

    # The next reading is scheduled now, not when this one reports back:
    # a reading that died without a word must not be the last one.
    schedule(state)
    {:noreply, state}
  end

  def handle_info({:read_done, {micros, result}}, state) do
    {:noreply, record(state, result, micros)}
  end

  @impl true
  def handle_call(:status, _from, state), do: {:reply, state.reading.status, state}
  def handle_call(:reading, _from, state), do: {:reply, state.reading, state}

  def handle_call(:recheck, _from, state) do
    state = read(state)
    {:reply, state.reading.status, state}
  end

  defp schedule(%{interval: :never}), do: :ok
  defp schedule(%{interval: ms}), do: Process.send_after(self(), :read, ms)

  defp read(state) do
    {micros, result} = :timer.tc(fn -> L1Reader.compute_tree_hash(state.root) end)
    record(state, result, micros)
  end

  defp record(state, result, micros) do
    previous = state.reading && state.reading.status

    {status, computed} =
      case result do
        {:ok, hash} when hash == state.expected ->
          {:ok, hash}

        {:ok, hash} ->
          {{:error,
            "L1_INTEGRITY_FAILED: the scroll on disk hashes to #{hash}, " <>
              "expected #{state.expected}"}, hash}

        {:error, reason} ->
          {{:error, "L1_INTEGRITY_FAILED: the scroll could not be read — #{reason}"}, nil}
      end

    announce(status, previous, div(micros, 1000))

    %{
      state
      | reading: %{
          status: status,
          computed: computed,
          expected: state.expected,
          read_at: DateTime.utc_now()
        }
    }
  end

  # Say it when it is first known and whenever it changes; stay quiet on
  # the ten-minute readings that find what the last one found.
  defp announce(status, status, _ms), do: :ok

  defp announce(:ok, _previous, ms) do
    Logger.info("L1Integrity: the scroll was read and matches its pinned hash (#{ms}ms)")
  end

  defp announce({:error, reason}, _previous, _ms) do
    Logger.error("L1Integrity: #{reason}")

    if Process.whereis(VavPipeline.Chatzotzrot) do
      VavPipeline.Chatzotzrot.signal(:assembly_leaders, reason: reason)
    end

    :ok
  end
end