diff --git a/vav_pipeline/lib/vav_pipeline/l1_integrity.ex b/vav_pipeline/lib/vav_pipeline/l1_integrity.ex index 8e5b618c..311ea0d7 100644 --- a/vav_pipeline/lib/vav_pipeline/l1_integrity.ex +++ b/vav_pipeline/lib/vav_pipeline/l1_integrity.ex @@ -18,10 +18,14 @@ defmodule VavPipeline.L1Integrity do 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 a fraction of a second, 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. + 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. @@ -85,13 +89,27 @@ defmodule VavPipeline.L1Integrity do {: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 - state = read(state) + 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} @@ -106,6 +124,10 @@ defmodule VavPipeline.L1Integrity do 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} = diff --git a/vav_pipeline/test/tablet38_l1_integrity_test.exs b/vav_pipeline/test/tablet38_l1_integrity_test.exs index d5ff2ca2..17966dd2 100644 --- a/vav_pipeline/test/tablet38_l1_integrity_test.exs +++ b/vav_pipeline/test/tablet38_l1_integrity_test.exs @@ -126,6 +126,27 @@ defmodule VavPipeline.Tablet38L1IntegrityTest do File.write!(Path.join([root, "torah", "index.json"]), "INDEX, altered") assert eventually(fn -> match?({:error, _}, L1Integrity.status(:tablet38_periodic)) end) + + # And it keeps re-reading: restored, it is tov again without being asked. + File.write!(Path.join([root, "torah", "index.json"]), "INDEX") + assert eventually(fn -> L1Integrity.status(:tablet38_periodic) == :ok end) + end + + test "answers from the last reading while a periodic re-reading is under way" do + root = small_scroll() + {:ok, pinned} = L1Reader.compute_tree_hash(root) + + {:ok, pid} = + L1Integrity.start_link(name: :tablet38_busy, l1_root: root, expected_hash: pinned, interval: :never) + + on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end) + + # Trigger a periodic reading by hand; the process must not be busy + # with it — a status call is answered at once, from the last reading. + send(pid, :read) + {micros, status} = :timer.tc(fn -> L1Integrity.status(:tablet38_busy) end) + assert status == :ok + assert micros < 50_000 end end