Tablet 46: the watch — rate limits at every gate, per IP at the pe'ah, per key inside

Tamid §1.1 wrote the pe'ah's limit (10/min, 100/h per IP); nothing
enforced it, on any tier, and every admitted request spends manna.

VavPipeline.BigdeiKodesh.Mishmar: fixed minute and hour windows per
subject in ETS, the subject stored only as sha256(boot salt, tier,
subject) — no IP and no key is ever in memory as itself, the sweep
drops closed windows, nothing survives a restart (the Tamid's "not
logged beyond the current request", kept in memory too). A refused
knock is not counted. 429 RATE_LIMITED with Retry-After and the manna
explanation (Shemot 16:18). Pe'ah per peer address; courtyard and
chatzer per key, after the gate (a wrong key is 401 and never counted).
Numbers: pe'ah 10/100 (Tamid), courtyard 20/200 and chatzer 5/60
(from Construction §2.3; config :rate_limits).

Service shorashim-external: externalTrafficPolicy Local. With Cluster
the pod saw every caller as the node, 91.92.153.113 (observed via
/proc/net/tcp) — a per-IP limit would have been one bucket for the
world.

Spec: tamid §1.1 (watch, 429 rows, the Service note), KOHEN.md.
Doorpost 42b38626… → e4e3d56d…. tablet46_mishmar_test; Elixir 533/0
(L3 on and off).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

vav_pipeline/lib/vav_pipeline/application.ex

@@ -47,7 +47,8 @@ defmodule VavPipeline.Application do
         {VavPipeline.VoiceMonitor, []},
         {VavPipeline.MachtzitHaShekel, []},
         # Tablet 30 — the index that lets a question name its own verse
-        {VavPipeline.L1Index, []}
+        {VavPipeline.L1Index, []},
+        {VavPipeline.BigdeiKodesh.Mishmar, []}
       ]
       |> maybe_add_http_server()
       |> maybe_add_repo()

vav_pipeline/lib/vav_pipeline/bigdei_kodesh/gate.ex

@@ -85,7 +85,9 @@ defmodule VavPipeline.BigdeiKodesh.Gate do
 
   # ------------------------------------------------------------
 
-  defp bearer(conn) do
+  @doc "The bearer key on the request, if the header has one."
+  @spec bearer(Plug.Conn.t()) :: {:ok, String.t()} | {:error, :no_header}
+  def bearer(conn) do
     case Plug.Conn.get_req_header(conn, "authorization") do
       ["Bearer " <> key] when key != "" -> {:ok, String.trim(key)}
       _ -> {:error, :no_header}

vav_pipeline/lib/vav_pipeline/bigdei_kodesh/mishmar.ex

@@ -0,0 +1,180 @@
+defmodule VavPipeline.BigdeiKodesh.Mishmar do
+  @moduledoc """
+  The watch at each gate — how often one hand may knock.
+
+  Bamidbar 18:4 — *v'shamru et mishmeret ohel mo'ed*: the Levites keep
+  the watch of the tent. The watch does not decide who may enter (that
+  is the Gate); it decides how often. The day's manna is one portion for
+  everyone who comes; without a watch one caller eats it all before noon.
+
+  Tamid §1.1 fixes the pe'ah: 10 requests a minute and 100 an hour per
+  IP, and "no identification stored — IP is used for rate-limiting only,
+  not logged beyond the current request." For the courtyard and the
+  chatzer the Tamid says only that 429 exists and carries the manna
+  explanation; the numbers here come from Construction §2.3 (peak
+  korbanot per hour 200; five Sanhedrin panels at once) and are
+  configuration, not law — `config :vav_pipeline, :rate_limits`.
+
+  **What is kept.** Two fixed windows per subject — the current minute
+  and the current hour — as counters in an ETS table. The subject is
+  not the IP and not the key: it is `sha256(salt <> subject)` with a
+  salt drawn at boot and never written anywhere. The table cannot be
+  read back into a list of who called; it cannot survive a restart;
+  the sweep drops a window a few minutes after it closes. This is the
+  Tamid's "not logged beyond the current request", kept in memory too.
+
+  **What is answered.** `{:ok, remaining_this_minute}` or
+  `{:error, window, retry_after_seconds}`. The router turns the latter
+  into 429 with `Retry-After`.
+  """
+
+  use GenServer
+  require Logger
+
+  @table __MODULE__
+  @sweep_ms :timer.minutes(1)
+
+  @type tier :: :peah | :courtyard | :chatzer
+  @type subject :: binary()
+
+  @default_limits [
+    # Tamid §1.1, verbatim.
+    peah: [minute: 10, hour: 100],
+    # Construction §2.3: 200 korbanot an hour at peak — one key may not
+    # be the whole peak. 20 a minute lets a client work; 200 an hour is
+    # the system's own ceiling.
+    courtyard: [minute: 20, hour: 200],
+    # Construction §2.3: five panels at once, deliberation measured in
+    # minutes. A key convening more than five cases a minute is not a
+    # court, it is a loop.
+    chatzer: [minute: 5, hour: 60]
+  ]
+
+  def start_link(opts \\ []) do
+    GenServer.start_link(__MODULE__, opts, name: Keyword.get(opts, :name, __MODULE__))
+  end
+
+  @doc "The configured limits for a tier, `[minute: n, hour: n]`."
+  @spec limits(tier()) :: keyword()
+  def limits(tier) do
+    configured = Application.get_env(:vav_pipeline, :rate_limits, [])
+    Keyword.get(configured, tier) || Keyword.fetch!(@default_limits, tier)
+  end
+
+  @doc """
+  Count one knock by `subject` at `tier`.
+
+  Returns `{:ok, remaining}` (knocks left this minute) or
+  `{:error, :minute | :hour, retry_after_seconds}`. A refused knock is
+  not counted: a caller who keeps knocking is not punished further, but
+  is not let in either.
+
+  When the watch is not running (tests without the supervisor) every
+  knock is allowed — and logged once as MISHMAR_ABSENT, because that is
+  not a state production may be in.
+  """
+  @spec knock(tier(), subject(), integer()) ::
+          {:ok, non_neg_integer()} | {:error, :minute | :hour, pos_integer()}
+  def knock(tier, subject, now \\ System.os_time(:second)) when is_binary(subject) do
+    case :ets.whereis(@table) do
+      :undefined ->
+        absent()
+        {:ok, 0}
+
+      _ ->
+        tier_limits = limits(tier)
+        per_minute = Keyword.fetch!(tier_limits, :minute)
+        per_hour = Keyword.fetch!(tier_limits, :hour)
+        id = hashed(tier, subject)
+        minute = div(now, 60)
+        hour = div(now, 3600)
+
+        minute_count = :ets.lookup_element(@table, {id, :minute, minute}, 2, 0)
+        hour_count = :ets.lookup_element(@table, {id, :hour, hour}, 2, 0)
+
+        cond do
+          hour_count >= per_hour ->
+            {:error, :hour, (hour + 1) * 3600 - now}
+
+          minute_count >= per_minute ->
+            {:error, :minute, (minute + 1) * 60 - now}
+
+          true ->
+            :ets.update_counter(@table, {id, :minute, minute}, {2, 1}, {{id, :minute, minute}, 0})
+            :ets.update_counter(@table, {id, :hour, hour}, {2, 1}, {{id, :hour, hour}, 0})
+            {:ok, per_minute - minute_count - 1}
+        end
+    end
+  end
+
+  @doc "The subject for a request at the pe'ah: its peer address, as text."
+  @spec peer(Plug.Conn.t()) :: subject()
+  def peer(%Plug.Conn{remote_ip: ip}) do
+    case :inet.ntoa(ip) do
+      {:error, _} -> inspect(ip)
+      chars -> List.to_string(chars)
+    end
+  end
+
+  @doc "Forget every window. For tests."
+  def reset(server \\ __MODULE__), do: GenServer.call(server, :reset)
+
+  @doc "Drop the windows that have closed. Runs on its own every minute; callable for tests."
+  def sweep(server \\ __MODULE__, now \\ System.os_time(:second)), do: GenServer.call(server, {:sweep, now})
+
+  # ------------------------------------------------------------
+
+  @impl true
+  def init(_opts) do
+    :ets.new(@table, [:named_table, :public, :set, write_concurrency: true, read_concurrency: true])
+    # The salt lives only in this process's state and in the (private)
+    # persistent key below. It is never logged, never written, and is
+    # different after every restart.
+    salt = :crypto.strong_rand_bytes(32)
+    :persistent_term.put({__MODULE__, :salt}, salt)
+    Process.send_after(self(), :sweep, @sweep_ms)
+    {:ok, %{}}
+  end
+
+  @impl true
+  def handle_call(:reset, _from, state) do
+    :ets.delete_all_objects(@table)
+    {:reply, :ok, state}
+  end
+
+  def handle_call({:sweep, now}, _from, state) do
+    do_sweep(now)
+    {:reply, :ok, state}
+  end
+
+  @impl true
+  def handle_info(:sweep, state) do
+    do_sweep(System.os_time(:second))
+    Process.send_after(self(), :sweep, @sweep_ms)
+    {:noreply, state}
+  end
+
+  # A minute window is dead once the minute after it has passed; an
+  # hour window once the hour after it has. Nothing older is kept.
+  defp do_sweep(now) do
+    minute = div(now, 60)
+    hour = div(now, 3600)
+
+    :ets.select_delete(@table, [
+      {{{:_, :minute, :"$1"}, :_}, [{:<, :"$1", minute - 1}], [true]},
+      {{{:_, :hour, :"$1"}, :_}, [{:<, :"$1", hour - 1}], [true]}
+    ])
+  end
+
+  defp hashed(tier, subject) do
+    salt = :persistent_term.get({__MODULE__, :salt})
+    :crypto.hash(:sha256, [salt, Atom.to_string(tier), 0, subject])
+  end
+
+  defp absent do
+    unless :persistent_term.get({__MODULE__, :absent_logged}, false) do
+      :persistent_term.put({__MODULE__, :absent_logged}, true)
+      Logger.warning("MISHMAR_ABSENT: the watch is not running — no rate limit is applied")
+    end
+  end
+end

vav_pipeline/lib/vav_pipeline/bigdei_kodesh/router.ex

@@ -28,7 +28,7 @@ defmodule VavPipeline.BigdeiKodesh.Router do
 
   require Logger
 
-  alias VavPipeline.BigdeiKodesh.Gate
+  alias VavPipeline.BigdeiKodesh.{Gate, Mishmar}
 
   # Form 3 architectural enforcement (vehaya §3.2). NoTrackingLogger
   # replaces Plug.Logger; sanitized paths on /peah/* honor the
@@ -63,17 +63,21 @@ defmodule VavPipeline.BigdeiKodesh.Router do
   end
 
   post "/peah/ask" do
-    case conn.body_params do
-      %{"question" => question} when is_binary(question) and question != "" ->
-        process_peah(conn, question)
+    # Tamid §1.1: 10 a minute, 100 an hour, per IP; the IP is used for
+    # this and nothing else (Mishmar keeps a salted hash, not the IP).
+    under_the_watch(conn, :peah, Mishmar.peer(conn), fn conn ->
+      case conn.body_params do
+        %{"question" => question} when is_binary(question) and question != "" ->
+          process_peah(conn, question)
 
-      _ ->
-        send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{
-          issue: "Missing or empty 'question' field",
-          correction: "Submit JSON with a 'question' string",
-          resubmit: true
-        })
-    end
+        _ ->
+          send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{
+            issue: "Missing or empty 'question' field",
+            correction: "Submit JSON with a 'question' string",
+            resubmit: true
+          })
+      end
+    end)
   end
 
   # ============================================================
@@ -129,6 +133,32 @@ defmodule VavPipeline.BigdeiKodesh.Router do
     })
   end
 
+  # The watch (Mishmar): how often one subject may knock at a tier.
+  # 429 carries the manna explanation Tamid §1.1 asks for, and Retry-After.
+  defp under_the_watch(conn, tier, subject, serve) do
+    case Mishmar.knock(tier, subject) do
+      {:ok, _remaining} ->
+        serve.(conn)
+
+      {:error, window, retry_after} ->
+        limits = Mishmar.limits(tier)
+
+        conn
+        |> put_resp_header("retry-after", Integer.to_string(retry_after))
+        |> send_error(429, "RATE_LIMITED", %{
+          issue:
+            "Too many requests this #{window}: the #{tier} allows " <>
+              "#{limits[:minute]} a minute and #{limits[:hour]} an hour per caller",
+          correction: "Wait #{retry_after}s and resubmit",
+          manna:
+            "The day's provision is one portion for everyone who comes (Shemot 16:18 — " <>
+              "the one who gathered much had nothing over). The watch keeps one caller " <>
+              "from gathering the whole omer.",
+          resubmit: true
+        })
+    end
+  end
+
   # === COURTYARD — Authenticated ===
   #
   # The key is checked against the keys file (VavPipeline.BigdeiKodesh.Gate),
@@ -136,7 +166,10 @@ defmodule VavPipeline.BigdeiKodesh.Router do
   defp through_the_gate(conn, tier, serve) do
     case Gate.admit(conn, tier) do
       {:ok, _granted} ->
-        serve.(conn)
+        # Past the gate, the key is the hand that knocks: the watch
+        # counts per key, not per IP. A refused key never reaches it.
+        {:ok, key} = Gate.bearer(conn)
+        under_the_watch(conn, tier, key, serve)
 
       {:error, :tier_too_low} ->
         send_error(conn, 403, "FORBIDDEN", %{
@@ -951,6 +984,7 @@ defmodule VavPipeline.BigdeiKodesh.Router do
   defp error_type("HALT" <> _), do: "cardinal_sin_detected"
   defp error_type("UNAUTHORIZED"), do: "authentication_required"
   defp error_type("FORBIDDEN"), do: "insufficient_tier"
+  defp error_type("RATE_LIMITED"), do: "too_many_requests"
   defp error_type("TEIKU" <> _), do: "unresolvable"
   defp error_type("LO_TOV"), do: "specification_mismatch"
   defp error_type("CHATAT" <> _), do: "correction_needed"

vav_pipeline/lib/vav_pipeline/doorpost.ex

@@ -175,7 +175,7 @@ defmodule VavPipeline.Doorpost do
   # changes; the update is a deliberate operator commit. A bare
   # hash-update commit (without a corresponding spec change in the
   # same commit) is itself a drift signal.
-  @expected_hash "42b38626c6e2bd3e6fe10ed7f4cb34ee8dae118205379b7d822f278fe7eaa110"
+  @expected_hash "e4e3d56de7422280b5562c4a7ebce02fe05225a938bbce26d6e52f26b6b240eb"
 
   @doc "Return the list of files that constitute the category-grade specification."
   @spec files() :: [String.t()]