diff --git a/metadata-relay/README.md b/metadata-relay/README.md
index be2aa1719..5b1acb98e 100644
--- a/metadata-relay/README.md
+++ b/metadata-relay/README.md
@@ -143,6 +143,7 @@ The service is configured entirely via environment variables for maximum flexibi
 | `SMTP_PORT`       | No       | `587`              | SMTP relay port                                                                                                         |
 | `SMTP_USERNAME`   | No       | -                  | SMTP username. SMTP auth is enabled when both username and password are present                                          |
 | `SMTP_PASSWORD`   | No       | -                  | SMTP password. SMTP auth is enabled when both username and password are present                                          |
+| `P2P_ACCESS_BEARER_TOKENS` | No | -                | Comma-separated list of bearer tokens the self-hosted iroh relay may present to `POST /p2p/access`. Multiple values are accepted so a token can be rotated by deploying the new one alongside the old and removing the old afterwards. Unset means the endpoint denies every relay client |
 
 ### Cache Configuration
 
diff --git a/metadata-relay/config/config.exs b/metadata-relay/config/config.exs
index 1dbad3471..8d0d7bce4 100644
--- a/metadata-relay/config/config.exs
+++ b/metadata-relay/config/config.exs
@@ -6,7 +6,21 @@ config :metadata_relay,
   port: 4000,
   # Ecto repository
   ecto_repos: [MetadataRelay.Repo],
-  dashboard_auth: [username: "admin", password: "admin"]
+  dashboard_auth: [username: "admin", password: "admin"],
+  # Bearer tokens the iroh relay may present to POST /p2p/access.
+  # Populated from P2P_ACCESS_BEARER_TOKENS at runtime. A list, so a token can
+  # be rotated by deploying both values before removing the old one.
+  p2p_access_bearer_tokens: [],
+  # Hard cap on distinct endpoint IDs held in ETS, sized against the pod's
+  # 512Mi memory limit. Above this we keep allowing traffic but stop recording
+  # new identities.
+  p2p_max_sightings: 200_000,
+  # How often accumulated ETS sightings are written to the database.
+  p2p_flush_interval_ms: 30_000,
+  # How often stale sightings are pruned.
+  p2p_prune_interval_ms: 86_400_000,
+  # Sightings not seen within this window are pruned (30 days).
+  p2p_retention_seconds: 2_592_000
 
 config :metadata_relay, MetadataRelay.Feedback.Notifier,
   recipient: nil,
diff --git a/metadata-relay/config/runtime.exs b/metadata-relay/config/runtime.exs
index 800426df8..db634835a 100644
--- a/metadata-relay/config/runtime.exs
+++ b/metadata-relay/config/runtime.exs
@@ -35,6 +35,23 @@ if config_env() != :test do
   config :metadata_relay,
     dashboard_auth: [username: dashboard_username, password: dashboard_password]
 
+  # Bearer tokens accepted on POST /p2p/access, presented by the iroh relay.
+  # Comma-separated so a token can be rotated by deploying both values before
+  # removing the old one. Unset means an empty list, which denies everything.
+  p2p_access_bearer_tokens =
+    case normalize_env.("P2P_ACCESS_BEARER_TOKENS") do
+      nil ->
+        []
+
+      value ->
+        value
+        |> String.split(",")
+        |> Enum.map(&String.trim/1)
+        |> Enum.reject(&(&1 == ""))
+    end
+
+  config :metadata_relay, p2p_access_bearer_tokens: p2p_access_bearer_tokens
+
   # Database configuration (all environments except test)
   db_path = System.get_env("SQLITE_DB_PATH") || "./metadata_relay.db"
 
diff --git a/metadata-relay/config/test.exs b/metadata-relay/config/test.exs
index 048b9c28d..8bfee1a9b 100644
--- a/metadata-relay/config/test.exs
+++ b/metadata-relay/config/test.exs
@@ -16,7 +16,15 @@ config :metadata_relay, MetadataRelay.Repo,
   pool: Ecto.Adapters.SQL.Sandbox
 
 config :metadata_relay,
-  rendezvous_master_pepper: "test-pepper-not-for-production"
+  rendezvous_master_pepper: "test-pepper-not-for-production",
+  p2p_access_bearer_tokens: ["test-relay-bearer"],
+  # The P2pAccess.Store's timers would otherwise fire mid-suite and write to
+  # the database from a process that does not own the sandbox connection.
+  # Pin them well beyond any plausible suite runtime; tests drive the work
+  # explicitly through flush_now/0, prune_now/0 and a direct :reload_blocks.
+  p2p_flush_interval_ms: 3_600_000,
+  p2p_prune_interval_ms: 3_600_000,
+  p2p_reload_retry_interval_ms: 3_600_000
 
 config :metadata_relay, MetadataRelay.Feedback.Notifier,
   recipient: "maintainer@example.com",
diff --git a/metadata-relay/lib/metadata_relay/application.ex b/metadata-relay/lib/metadata_relay/application.ex
index 0abbb75e7..35f3a167f 100644
--- a/metadata-relay/lib/metadata_relay/application.ex
+++ b/metadata-relay/lib/metadata_relay/application.ex
@@ -25,6 +25,8 @@ defmodule MetadataRelay.Application do
         {cache_adapter, cache_opts},
         # Long-lived ETS owner for pairing fallback storage
         MetadataRelay.PairingStore,
+        # Long-lived ETS owner for p2p relay access control
+        MetadataRelay.P2pAccess.Store,
         # Rate limiter for crash reports and pairing
         MetadataRelay.RateLimiter,
         # Metrics collector
diff --git a/metadata-relay/lib/metadata_relay/p2p_access.ex b/metadata-relay/lib/metadata_relay/p2p_access.ex
new file mode 100644
index 000000000..eed545a07
--- /dev/null
+++ b/metadata-relay/lib/metadata_relay/p2p_access.ex
@@ -0,0 +1,129 @@
+defmodule MetadataRelay.P2pAccess do
+  @moduledoc """
+  Access control policy for the self-hosted iroh relay.
+
+  The relay POSTs to `/p2p/access` before accepting an endpoint. The endpoint
+  ID it sends is proven by the relay handshake, so it is a trustworthy
+  identifier and needs no further authentication from the client.
+
+  This does not verify that a caller is a genuine Mydia instance. Mydia is open
+  source and self-hosted with no user accounts, so there is no per-user secret
+  to check. What this provides is attribution, a size-capped record of who used
+  the relay, and the ability to revoke a specific endpoint.
+
+  Phase 1 policy: allow everyone except explicitly blocked endpoints.
+  """
+
+  alias MetadataRelay.P2pAccess.Store
+
+  @endpoint_id_length 64
+
+  @doc """
+  The authorization decision for an endpoint. ETS only.
+
+  Case-normalizes the endpoint ID before recording the sighting and checking
+  the blocklist, since `block/2` and `unblock/1` store and match on the
+  downcased form and a block must not be bypassable by changing case. This
+  does **not** validate the ID: malformed input is still recorded and
+  checked (and will simply never match a block). Callers handling untrusted
+  input who need to distinguish a malformed ID from a denied one should call
+  `normalize_endpoint_id/1` first.
+  """
+  def authorize(endpoint_id) when is_binary(endpoint_id) do
+    endpoint_id = String.downcase(endpoint_id)
+
+    Store.record_sighting(endpoint_id)
+
+    if Store.blocked?(endpoint_id) do
+      MetadataRelay.Metrics.inc("metadata_relay_p2p_access_total", result: "deny")
+      :deny
+    else
+      MetadataRelay.Metrics.inc("metadata_relay_p2p_access_total", result: "allow")
+      :allow
+    end
+  end
+
+  @doc """
+  Validates and downcases an endpoint ID.
+
+  iroh endpoint IDs are 32-byte ed25519 public keys, hex-encoded to 64
+  characters.
+  """
+  def normalize_endpoint_id(endpoint_id) when is_binary(endpoint_id) do
+    normalized = String.downcase(endpoint_id)
+
+    if String.length(normalized) == @endpoint_id_length and
+         String.match?(normalized, ~r/\A[0-9a-f]+\z/) do
+      {:ok, normalized}
+    else
+      :error
+    end
+  end
+
+  def normalize_endpoint_id(_), do: :error
+
+  @doc """
+  Whether a bearer token presented by the relay is one we accept.
+
+  The configured value is a list so a token can be rotated by deploying both
+  the old and new value before removing the old one. An empty list rejects
+  everything, which fails closed if the deployment forgets the secret.
+  """
+  def valid_bearer?(token) when is_binary(token) do
+    Enum.any?(bearer_tokens(), fn configured ->
+      Plug.Crypto.secure_compare(configured, token)
+    end)
+  end
+
+  def valid_bearer?(_), do: false
+
+  @doc """
+  Denies an endpoint relay access. Callable over rpc.
+
+      MetadataRelay.P2pAccess.block("abcd...", "bandwidth abuse")
+  """
+  def block(endpoint_id, reason) when is_binary(reason) do
+    case normalize_endpoint_id(endpoint_id) do
+      {:ok, normalized} -> Store.put_block(normalized, reason)
+      :error -> {:error, :invalid_endpoint_id}
+    end
+  end
+
+  @doc """
+  Restores relay access for an endpoint. Callable over rpc.
+  """
+  def unblock(endpoint_id) do
+    case normalize_endpoint_id(endpoint_id) do
+      {:ok, normalized} -> Store.delete_block(normalized)
+      :error -> {:error, :invalid_endpoint_id}
+    end
+  end
+
+  @doc """
+  The most recently active endpoints, newest first. Callable over rpc.
+
+  ETS only, deliberately: this is what an operator reaches for mid-incident,
+  and it must not depend on the database being responsive. It still survives a
+  restart, because `Store.seed_sightings/0` repopulates ETS from the durable
+  table at boot.
+  """
+  def list_recent(limit \\ 50) when is_integer(limit) and limit > 0 do
+    :p2p_sightings
+    |> :ets.tab2list()
+    |> Enum.sort_by(fn {_id, _first, last_seen, _count} -> last_seen end, :desc)
+    |> Enum.take(limit)
+    |> Enum.map(fn {endpoint_id, first_seen, last_seen, conn_count} ->
+      %{
+        endpoint_id: endpoint_id,
+        first_seen: DateTime.from_unix!(first_seen),
+        last_seen: DateTime.from_unix!(last_seen),
+        conn_count: conn_count,
+        blocked: Store.blocked?(endpoint_id)
+      }
+    end)
+  end
+
+  defp bearer_tokens do
+    Application.get_env(:metadata_relay, :p2p_access_bearer_tokens, [])
+  end
+end
diff --git a/metadata-relay/lib/metadata_relay/p2p_access/block.ex b/metadata-relay/lib/metadata_relay/p2p_access/block.ex
new file mode 100644
index 000000000..16bb4b893
--- /dev/null
+++ b/metadata-relay/lib/metadata_relay/p2p_access/block.ex
@@ -0,0 +1,16 @@
+defmodule MetadataRelay.P2pAccess.Block do
+  @moduledoc """
+  An endpoint that is denied relay access.
+
+  Written synchronously on admin action, and loaded into ETS at boot.
+  """
+
+  use Ecto.Schema
+
+  @primary_key {:endpoint_id, :string, autogenerate: false}
+
+  schema "p2p_blocked_endpoints" do
+    field(:reason, :string)
+    field(:blocked_at, :utc_datetime)
+  end
+end
diff --git a/metadata-relay/lib/metadata_relay/p2p_access/sighting.ex b/metadata-relay/lib/metadata_relay/p2p_access/sighting.ex
new file mode 100644
index 000000000..731ec4682
--- /dev/null
+++ b/metadata-relay/lib/metadata_relay/p2p_access/sighting.ex
@@ -0,0 +1,18 @@
+defmodule MetadataRelay.P2pAccess.Sighting do
+  @moduledoc """
+  A p2p endpoint the relay has asked us about.
+
+  Written only by the periodic flush in `MetadataRelay.P2pAccess.Store`.
+  Never read or written on the authorization hot path.
+  """
+
+  use Ecto.Schema
+
+  @primary_key {:endpoint_id, :string, autogenerate: false}
+
+  schema "p2p_endpoint_sightings" do
+    field(:first_seen, :utc_datetime)
+    field(:last_seen, :utc_datetime)
+    field(:conn_count, :integer, default: 0)
+  end
+end
diff --git a/metadata-relay/lib/metadata_relay/p2p_access/store.ex b/metadata-relay/lib/metadata_relay/p2p_access/store.ex
new file mode 100644
index 000000000..a2087ebba
--- /dev/null
+++ b/metadata-relay/lib/metadata_relay/p2p_access/store.ex
@@ -0,0 +1,466 @@
+defmodule MetadataRelay.P2pAccess.Store do
+  @moduledoc """
+  Long-lived ETS owner for p2p relay access state.
+
+  Owns two tables:
+
+    * `:p2p_sightings` - `{endpoint_id, first_seen_unix, last_seen_unix, conn_count}`
+    * `:p2p_blocked`   - `{endpoint_id, reason, blocked_at_unix}`
+
+  Two functions here sit on the relay authorization hot path and must stay
+  ETS-only: `record_sighting/1` and `blocked?/1`. The access callback is
+  fail-closed, so if either becomes slow or raises, the relay refuses real
+  users.
+
+  The rest do touch the database, but never inline with a request. Seeding runs
+  at boot, flush and prune run on timers inside the GenServer, and blocking runs
+  on an operator's admin action.
+
+  Both tables are seeded from the database at boot, so a restart does not
+  silently unblock revoked endpoints or reset sighting history.
+  """
+
+  use GenServer
+
+  import Ecto.Query, only: [from: 2]
+
+  require Logger
+
+  alias MetadataRelay.P2pAccess.{Block, Sighting}
+  alias MetadataRelay.Repo
+
+  @sightings :p2p_sightings
+  @blocked :p2p_blocked
+
+  @default_max_sightings 200_000
+  @default_flush_interval_ms 30_000
+  @default_prune_interval_ms 86_400_000
+  @default_retention_seconds 2_592_000
+  @default_reload_retry_interval_ms 5_000
+
+  # exqlite builds SQLite with SQLITE_MAX_VARIABLE_NUMBER=32766, and each
+  # sighting row binds 4 parameters, so a single insert_all/3 tops out at
+  # 8_191 rows. The configured sighting cap is far higher than that, so the
+  # write is chunked well below the ceiling rather than left to fail.
+  @insert_chunk_size 2_000
+
+  def start_link(opts) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @doc """
+  Creates both ETS tables if they do not exist. Idempotent.
+  """
+  def init_tables do
+    # Sightings take a write (update_counter + update_element) on every relay
+    # connection, so they need write_concurrency for that many-distinct-keys
+    # counter-bump pattern (mirrors MetadataRelay.Metrics's counters table).
+    ensure_table(@sightings, [
+      :set,
+      :public,
+      :named_table,
+      read_concurrency: true,
+      write_concurrency: true
+    ])
+
+    # Blocked is read on every connection but written only on rare admin
+    # action, so it stays read-optimized instead.
+    ensure_table(@blocked, [:set, :public, :named_table, read_concurrency: true])
+    :ok
+  end
+
+  @doc """
+  Records that the relay asked about this endpoint.
+
+  New endpoints are only recorded while the table is below
+  `:p2p_max_sightings`. Endpoints already in the table are always updated, so
+  a flood of unknown endpoint IDs costs us telemetry rather than memory.
+
+  The cap is a soft cap, not a hard ceiling: the membership check, the size
+  check, and the write are three unsynchronized ETS calls, so concurrent
+  callers racing at the boundary with different new endpoint IDs can each
+  observe the table below the cap and all proceed, overshooting it by
+  roughly the number of callers racing at that instant. This is accepted
+  deliberately, because the alternative is serializing every relay
+  connection through this GenServer, which would make it the very
+  bottleneck this ETS-only hot path exists to avoid. What this guarantees
+  is bounded growth, not an exact maximum.
+  """
+  def record_sighting(endpoint_id) when is_binary(endpoint_id) do
+    now = System.system_time(:second)
+
+    cond do
+      :ets.member(@sightings, endpoint_id) ->
+        bump(endpoint_id, now)
+
+      :ets.info(@sightings, :size) < max_sightings() ->
+        bump(endpoint_id, now)
+
+      true ->
+        MetadataRelay.Metrics.inc("metadata_relay_p2p_sightings_shed_total")
+        :ok
+    end
+  end
+
+  def sighting_count, do: :ets.info(@sightings, :size)
+
+  def lookup_sighting(endpoint_id) when is_binary(endpoint_id) do
+    case :ets.lookup(@sightings, endpoint_id) do
+      [{^endpoint_id, first_seen, last_seen, conn_count}] ->
+        {:ok, {first_seen, last_seen, conn_count}}
+
+      [] ->
+        :error
+    end
+  end
+
+  @doc """
+  Whether this endpoint is denied relay access. ETS only, hot path.
+  """
+  def blocked?(endpoint_id) when is_binary(endpoint_id) do
+    :ets.member(@blocked, endpoint_id)
+  end
+
+  @doc """
+  Blocks an endpoint. Writes the database first so a crash between the two
+  writes fails safe: the block survives and is restored by `reload_blocks/0`.
+  """
+  def put_block(endpoint_id, reason) when is_binary(endpoint_id) and is_binary(reason) do
+    blocked_at = DateTime.utc_now() |> DateTime.truncate(:second)
+
+    result =
+      Repo.insert(
+        %Block{endpoint_id: endpoint_id, reason: reason, blocked_at: blocked_at},
+        on_conflict: {:replace, [:reason, :blocked_at]},
+        conflict_target: :endpoint_id
+      )
+
+    case result do
+      {:ok, _} ->
+        :ets.insert(@blocked, {endpoint_id, reason, DateTime.to_unix(blocked_at)})
+        Logger.warning("Blocked p2p endpoint #{String.slice(endpoint_id, 0, 8)}: #{reason}")
+        :ok
+
+      {:error, changeset} ->
+        {:error, changeset}
+    end
+  end
+
+  @doc """
+  Unblocks an endpoint. Removes it from ETS first so access is restored even
+  if the database delete fails.
+  """
+  def delete_block(endpoint_id) when is_binary(endpoint_id) do
+    :ets.delete(@blocked, endpoint_id)
+    Repo.delete_all(from(b in Block, where: b.endpoint_id == ^endpoint_id))
+    Logger.info("Unblocked p2p endpoint #{String.slice(endpoint_id, 0, 8)}")
+    :ok
+  end
+
+  @doc """
+  Repopulates the ETS blocklist from the database. Called at boot.
+
+  Only the database read is fault-tolerant. If the read succeeds, any error
+  in the transform or the ETS write is a programming error and crashes.
+  """
+  def reload_blocks do
+    case fetch_blocks() do
+      {:ok, blocks} ->
+        rows =
+          Enum.map(blocks, fn block ->
+            {block.endpoint_id, block.reason, DateTime.to_unix(block.blocked_at)}
+          end)
+
+        :ets.delete_all_objects(@blocked)
+        :ets.insert(@blocked, rows)
+
+        Logger.info("Loaded #{length(rows)} blocked p2p endpoints")
+        :ok
+
+      {:error, error} ->
+        Logger.error("Could not load p2p blocklist from the database: #{inspect(error)}")
+        {:error, error}
+    end
+  end
+
+  @doc """
+  Repopulates the ETS sightings table from the database. Called at boot.
+
+  Without this, `list_recent/1` would only ever show endpoints seen since the
+  last restart, and the first flush after a restart would overwrite each
+  persisted `conn_count` with a counter that had just restarted at zero.
+
+  At most `:p2p_max_sightings` rows are seeded, most recently seen first, so
+  a boot cannot start out over the cap.
+
+  Sightings are telemetry, not policy: a failed seed is logged and the table
+  is left empty rather than blocking boot or being retried.
+  """
+  def seed_sightings do
+    case fetch_recent_sightings(max_sightings()) do
+      {:ok, sightings} ->
+        rows =
+          Enum.map(sightings, fn sighting ->
+            {sighting.endpoint_id, DateTime.to_unix(sighting.first_seen),
+             DateTime.to_unix(sighting.last_seen), sighting.conn_count}
+          end)
+
+        :ets.insert(@sightings, rows)
+
+        Logger.info("Seeded #{length(rows)} p2p endpoint sightings from the database")
+        :ok
+
+      {:error, error} ->
+        Logger.warning("Could not seed p2p sightings from the database: #{inspect(error)}")
+        {:error, error}
+    end
+  end
+
+  @doc """
+  Writes accumulated ETS sightings to the database.
+
+  Returns the number of rows written to the database. If the write fails, the
+  failure is logged and this returns `{:ok, 0}` — zero is the truth here,
+  since nothing was actually persisted.
+  """
+  def flush_now, do: GenServer.call(__MODULE__, :flush)
+
+  @doc """
+  Drops sightings older than the retention window from ETS and the database.
+
+  Returns the number of ETS rows evicted. That eviction has already happened
+  by the time the database delete is attempted, so if the database delete
+  fails, the failure is logged but the count is unchanged — the stale rows
+  are left in the database to be cleaned up by a later prune, rather than
+  being subtracted from what ETS actually evicted.
+  """
+  def prune_now, do: GenServer.call(__MODULE__, :prune)
+
+  # The database is not always reachable when the Store starts: under the
+  # ExUnit sandbox no connection is checked out yet, and in development the
+  # database can lag the application. An unreachable database must not stop
+  # the service from booting.
+  defp fetch_blocks do
+    {:ok, Repo.all(Block)}
+  rescue
+    error -> {:error, error}
+  end
+
+  defp fetch_recent_sightings(limit) do
+    query = from(s in Sighting, order_by: [desc: s.last_seen], limit: ^limit)
+    {:ok, Repo.all(query)}
+  rescue
+    error -> {:error, error}
+  end
+
+  @impl true
+  def init(_opts) do
+    init_tables()
+    reload_blocks_or_retry()
+    seed_sightings()
+
+    schedule(:flush, flush_interval_ms())
+    schedule(:prune, prune_interval_ms())
+
+    # Zero, not the current time, so the first flush after boot persists the
+    # sightings just seeded from the database as well as anything recorded
+    # since. The chunking in do_flush/1 is what keeps that first, largest
+    # flush safe.
+    {:ok, %{last_flush_at: 0}}
+  end
+
+  # A blocklist that fails to load must not leave the service running with an
+  # empty one: that silently restores access for every revoked endpoint, which
+  # is the exact failure this feature exists to prevent. Retry until it loads.
+  defp reload_blocks_or_retry do
+    case reload_blocks() do
+      :ok ->
+        :ok
+
+      {:error, _} = error ->
+        Logger.warning(
+          "The p2p blocklist is empty and revoked endpoints are being allowed. " <>
+            "Retrying the load in #{reload_retry_interval_ms()}ms."
+        )
+
+        schedule(:reload_blocks, reload_retry_interval_ms())
+        error
+    end
+  end
+
+  # `update_counter` with a default tuple creates the row atomically when it is
+  # missing, so `update_element` that follows always finds a row to touch.
+  defp bump(endpoint_id, now) do
+    :ets.update_counter(@sightings, endpoint_id, {4, 1}, {endpoint_id, now, now, 0})
+    :ets.update_element(@sightings, endpoint_id, {3, now})
+    :ok
+  end
+
+  defp max_sightings do
+    Application.get_env(:metadata_relay, :p2p_max_sightings, @default_max_sightings)
+  end
+
+  defp ensure_table(name, opts) do
+    if :ets.whereis(name) == :undefined do
+      :ets.new(name, opts)
+      Logger.info("Created ETS table #{inspect(name)} for p2p access control")
+    end
+
+    :ok
+  end
+
+  @impl true
+  def handle_call(:flush, _from, state) do
+    {result, state} = do_flush(state)
+    {:reply, result, state}
+  end
+
+  @impl true
+  def handle_call(:prune, _from, state), do: {:reply, do_prune(), state}
+
+  @impl true
+  def handle_info(:flush, state) do
+    {_result, state} = do_flush(state)
+    schedule(:flush, flush_interval_ms())
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info(:prune, state) do
+    do_prune()
+    schedule(:prune, prune_interval_ms())
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info(:reload_blocks, state) do
+    reload_blocks_or_retry()
+    {:noreply, state}
+  end
+
+  defp do_flush(state) do
+    # Read the clock before the select, never after: a sighting recorded
+    # between the two would otherwise fall outside both this window and the
+    # next one and never be persisted.
+    now = System.system_time(:second)
+
+    rows =
+      @sightings
+      |> select_since(state.last_flush_at)
+      |> Enum.map(fn {endpoint_id, first_seen, last_seen, conn_count} ->
+        %{
+          endpoint_id: endpoint_id,
+          first_seen: DateTime.from_unix!(first_seen),
+          last_seen: DateTime.from_unix!(last_seen),
+          conn_count: conn_count
+        }
+      end)
+
+    case rows do
+      [] ->
+        {{:ok, 0}, %{state | last_flush_at: now}}
+
+      rows ->
+        # Only the database write below is fault-tolerant. It can fail for
+        # environmental reasons (connection drop, lock timeout) and must not
+        # take down the Store, which owns the ETS tables the request path
+        # depends on: crashing it would discard every sighting and the
+        # in-memory blocklist. The transform above, including
+        # DateTime.from_unix!/1, is deliberately left outside this rescue —
+        # a bad value there is a programming error and must crash loudly
+        # rather than be reported as a silent "flush failed".
+        try do
+          count = insert_sightings(rows)
+          {{:ok, count}, %{state | last_flush_at: now}}
+        rescue
+          error ->
+            Logger.error("p2p sighting flush failed to write to the database: #{inspect(error)}")
+
+            # The window is deliberately not advanced on failure, so the next
+            # flush retries these rows instead of dropping them.
+            {{:ok, 0}, state}
+        end
+    end
+  end
+
+  # Only rows touched since the previous flush need writing, so each cycle
+  # costs the delta rather than a full copy of the table. The bound is `>=`,
+  # not `>`: `last_seen` has one-second resolution, so a sighting recorded in
+  # the same second as the previous flush would be skipped forever by a
+  # strict comparison. Re-writing a row an extra time is harmless; losing one
+  # is not. Position 3 of the tuple is `last_seen`.
+  defp select_since(table, since) do
+    :ets.select(table, [{{:_, :_, :"$1", :_}, [{:>=, :"$1", since}], [:"$_"]}])
+  end
+
+  # One insert_all/3 per chunk, summed, so a large flush cannot blow past
+  # SQLite's bind-parameter ceiling. Errors propagate to the caller's rescue.
+  defp insert_sightings(rows) do
+    rows
+    |> Enum.chunk_every(@insert_chunk_size)
+    |> Enum.reduce(0, fn chunk, total ->
+      {count, _} =
+        Repo.insert_all(Sighting, chunk,
+          on_conflict: {:replace, [:last_seen, :conn_count]},
+          conflict_target: :endpoint_id
+        )
+
+      total + count
+    end)
+  end
+
+  defp do_prune do
+    cutoff = System.system_time(:second) - retention_seconds()
+
+    stale =
+      :ets.select(@sightings, [
+        {{:"$1", :_, :"$2", :_}, [{:<, :"$2", cutoff}], [:"$1"]}
+      ])
+
+    Enum.each(stale, &:ets.delete(@sightings, &1))
+
+    cutoff_dt = DateTime.from_unix!(cutoff)
+
+    # As with do_flush/0, only the database delete below is fault-tolerant:
+    # it can fail for environmental reasons and must not take down the
+    # Store. The ETS work above has already removed the stale rows the
+    # request path relies on, regardless of whether the mirrored database
+    # cleanup completes.
+    try do
+      Repo.delete_all(from(s in Sighting, where: s.last_seen < ^cutoff_dt))
+    rescue
+      error ->
+        Logger.error("p2p sighting prune failed to delete database rows: #{inspect(error)}")
+    end
+
+    if stale != [] do
+      Logger.info("Pruned #{length(stale)} stale p2p endpoint sightings")
+    end
+
+    {:ok, length(stale)}
+  end
+
+  defp schedule(message, interval_ms) do
+    Process.send_after(self(), message, interval_ms)
+  end
+
+  defp flush_interval_ms do
+    Application.get_env(:metadata_relay, :p2p_flush_interval_ms, @default_flush_interval_ms)
+  end
+
+  defp prune_interval_ms do
+    Application.get_env(:metadata_relay, :p2p_prune_interval_ms, @default_prune_interval_ms)
+  end
+
+  defp reload_retry_interval_ms do
+    Application.get_env(
+      :metadata_relay,
+      :p2p_reload_retry_interval_ms,
+      @default_reload_retry_interval_ms
+    )
+  end
+
+  defp retention_seconds do
+    Application.get_env(:metadata_relay, :p2p_retention_seconds, @default_retention_seconds)
+  end
+end
diff --git a/metadata-relay/lib/metadata_relay/router.ex b/metadata-relay/lib/metadata_relay/router.ex
index b485c0c25..fc390dae3 100644
--- a/metadata-relay/lib/metadata_relay/router.ex
+++ b/metadata-relay/lib/metadata_relay/router.ex
@@ -13,6 +13,17 @@ defmodule MetadataRelay.Router do
   alias MetadataRelay.OpenSubtitles.Handler, as: SubtitlesHandler
   alias MetadataRelay.Pairing.Handler, as: PairingHandler
   alias MetadataRelay.Trakt.Handler, as: TraktHandler
+  alias MetadataRelay.P2pAccess
+
+  # The header the relay actually sends the endpoint ID in is `X-Iroh-NodeId`.
+  # Upstream names the constant `X_IROH_ENDPOINT_ID` and its doc comment
+  # advertises `X-Iroh-Endpoint-Id`, but the value it puts on the wire is
+  # `X-Iroh-NodeId` (iroh-relay v1.0.0, src/main.rs:36 and src/main.rs:319).
+  # The code is authoritative; upstream's own docs contradict it. The
+  # `x-iroh-endpoint-id` fallback is deliberate forward-compatibility for the
+  # day upstream corrects the constant to match its documentation — do not
+  # remove it as dead code. Plug downcases header names, so both are lowercase.
+  @endpoint_id_headers ["x-iroh-nodeid", "x-iroh-endpoint-id"]
 
   @feedback_param_atoms %{
     "type" => :type,
@@ -357,6 +368,27 @@ defmodule MetadataRelay.Router do
     handle_pairing_delete(conn, fn -> PairingHandler.delete_claim(code) end)
   end
 
+  # ============================================================================
+  # P2P Relay Access Control
+  # ============================================================================
+
+  # Called by the self-hosted iroh relay before it accepts an endpoint.
+  # The relay grants access only on a 200 response whose body is "true"; every
+  # other outcome, including an error or timeout, denies. Keep this handler
+  # ETS-only so it can never become the slow path for relay connections.
+  post "/p2p/access" do
+    with :ok <- authorize_relay_caller(conn),
+         {:ok, endpoint_id} <- read_endpoint_id(conn) do
+      case P2pAccess.authorize(endpoint_id) do
+        :allow -> send_access_response(conn, 200, "true")
+        :deny -> send_access_response(conn, 403, "false")
+      end
+    else
+      {:error, :unauthorized} -> send_access_response(conn, 403, "false")
+      {:error, :invalid_endpoint_id} -> send_access_response(conn, 400, "false")
+    end
+  end
+
   # ============================================================================
   # Trakt.tv API Proxy
   # ============================================================================
@@ -630,6 +662,48 @@ defmodule MetadataRelay.Router do
     end
   end
 
+  defp authorize_relay_caller(conn) do
+    token =
+      case get_req_header(conn, "authorization") do
+        ["Bearer " <> token | _] -> token
+        _ -> nil
+      end
+
+    if P2pAccess.valid_bearer?(token) do
+      :ok
+    else
+      MetadataRelay.Metrics.inc("metadata_relay_p2p_access_total", result: "unauthorized")
+      {:error, :unauthorized}
+    end
+  end
+
+  defp read_endpoint_id(conn) do
+    raw =
+      Enum.find_value(@endpoint_id_headers, fn header ->
+        case get_req_header(conn, header) do
+          [value | _] -> value
+          [] -> nil
+        end
+      end)
+
+    case P2pAccess.normalize_endpoint_id(raw) do
+      {:ok, endpoint_id} ->
+        {:ok, endpoint_id}
+
+      :error ->
+        MetadataRelay.Metrics.inc("metadata_relay_p2p_access_total", result: "malformed")
+        {:error, :invalid_endpoint_id}
+    end
+  end
+
+  # The relay parses the body as text and requires exactly "true" to allow.
+  # Do not switch this to JSON.
+  defp send_access_response(conn, status, body) do
+    conn
+    |> put_resp_content_type("text/plain")
+    |> send_resp(status, body)
+  end
+
   defp process_crash_report(conn) do
     with {:ok, body} <- validate_crash_report(conn.body_params),
          {:ok, _occurrence} <- store_crash_report(body) do
diff --git a/metadata-relay/priv/repo/migrations/20260730120000_create_p2p_access_tables.exs b/metadata-relay/priv/repo/migrations/20260730120000_create_p2p_access_tables.exs
new file mode 100644
index 000000000..86380da88
--- /dev/null
+++ b/metadata-relay/priv/repo/migrations/20260730120000_create_p2p_access_tables.exs
@@ -0,0 +1,20 @@
+defmodule MetadataRelay.Repo.Migrations.CreateP2pAccessTables do
+  use Ecto.Migration
+
+  def change do
+    create table(:p2p_endpoint_sightings, primary_key: false) do
+      add(:endpoint_id, :string, primary_key: true)
+      add(:first_seen, :utc_datetime, null: false)
+      add(:last_seen, :utc_datetime, null: false)
+      add(:conn_count, :integer, null: false, default: 0)
+    end
+
+    create(index(:p2p_endpoint_sightings, [:last_seen]))
+
+    create table(:p2p_blocked_endpoints, primary_key: false) do
+      add(:endpoint_id, :string, primary_key: true)
+      add(:reason, :string)
+      add(:blocked_at, :utc_datetime, null: false)
+    end
+  end
+end
diff --git a/metadata-relay/test/metadata_relay/p2p_access/router_test.exs b/metadata-relay/test/metadata_relay/p2p_access/router_test.exs
new file mode 100644
index 000000000..446cb2969
--- /dev/null
+++ b/metadata-relay/test/metadata_relay/p2p_access/router_test.exs
@@ -0,0 +1,159 @@
+defmodule MetadataRelay.P2pAccess.RouterTest do
+  use ExUnit.Case, async: false
+
+  @moduletag :capture_log
+
+  alias MetadataRelay.P2pAccess
+  alias MetadataRelay.P2pAccess.Store
+  alias MetadataRelay.Router
+
+  @bearer "test-relay-bearer"
+
+  setup do
+    :ok = Ecto.Adapters.SQL.Sandbox.checkout(MetadataRelay.Repo)
+    Ecto.Adapters.SQL.Sandbox.mode(MetadataRelay.Repo, {:shared, self()})
+
+    Store.init_tables()
+    :ets.delete_all_objects(:p2p_sightings)
+    :ets.delete_all_objects(:p2p_blocked)
+    MetadataRelay.Repo.delete_all(MetadataRelay.P2pAccess.Block)
+
+    original = Application.get_env(:metadata_relay, :p2p_access_bearer_tokens)
+    Application.put_env(:metadata_relay, :p2p_access_bearer_tokens, [@bearer])
+    on_exit(fn -> Application.put_env(:metadata_relay, :p2p_access_bearer_tokens, original) end)
+
+    :ok
+  end
+
+  defp endpoint_id(n),
+    do: String.pad_leading(Integer.to_string(n, 16), 64, "0") |> String.downcase()
+
+  defp request(opts) do
+    conn = Plug.Test.conn(:post, "/p2p/access", "")
+
+    conn =
+      case Keyword.get(opts, :bearer, @bearer) do
+        nil -> conn
+        token -> Plug.Conn.put_req_header(conn, "authorization", "Bearer " <> token)
+      end
+
+    conn =
+      case Keyword.get(opts, :endpoint_id, endpoint_id(1)) do
+        nil -> conn
+        id -> Plug.Conn.put_req_header(conn, Keyword.get(opts, :header, "x-iroh-nodeid"), id)
+      end
+
+    Router.call(conn, [])
+  end
+
+  test "allows an unknown endpoint with the literal body true" do
+    conn = request([])
+
+    assert conn.status == 200
+    assert conn.resp_body == "true"
+  end
+
+  test "responds as text/plain, not JSON" do
+    conn = request([])
+
+    assert ["text/plain" <> _] = Plug.Conn.get_resp_header(conn, "content-type")
+  end
+
+  test "denies when the bearer is missing" do
+    conn = request(bearer: nil)
+
+    assert conn.status == 403
+    refute conn.resp_body == "true"
+  end
+
+  test "denies when the bearer is wrong" do
+    conn = request(bearer: "wrong-token")
+
+    assert conn.status == 403
+    refute conn.resp_body == "true"
+  end
+
+  test "accepts any bearer from a rotation list" do
+    Application.put_env(:metadata_relay, :p2p_access_bearer_tokens, ["old", "new"])
+
+    assert request(bearer: "old").status == 200
+    assert request(bearer: "new").status == 200
+  end
+
+  test "denies a blocked endpoint" do
+    id = endpoint_id(2)
+    :ok = P2pAccess.block(id, "bandwidth abuse")
+
+    conn = request(endpoint_id: id)
+
+    assert conn.status == 403
+    refute conn.resp_body == "true"
+  end
+
+  # The relay sends the endpoint ID as `X-Iroh-NodeId`, whatever upstream's
+  # doc comment claims. Reading the wrong header denies every client, so both
+  # the real header and the documented-but-unused one are covered here.
+  test "reads the endpoint id from x-iroh-nodeid, the header the relay sends" do
+    id = endpoint_id(5)
+
+    conn = request(header: "x-iroh-nodeid", endpoint_id: id)
+
+    assert conn.status == 200
+    assert conn.resp_body == "true"
+    assert {:ok, _} = Store.lookup_sighting(id)
+  end
+
+  test "still reads the endpoint id from the x-iroh-endpoint-id fallback" do
+    id = endpoint_id(6)
+
+    conn = request(header: "x-iroh-endpoint-id", endpoint_id: id)
+
+    assert conn.status == 200
+    assert conn.resp_body == "true"
+    assert {:ok, _} = Store.lookup_sighting(id)
+  end
+
+  test "rejects a missing endpoint id header" do
+    conn = request(endpoint_id: nil)
+
+    assert conn.status == 400
+    refute conn.resp_body == "true"
+  end
+
+  test "rejects a malformed endpoint id" do
+    conn = request(endpoint_id: "not-a-valid-endpoint-id")
+
+    assert conn.status == 400
+    refute conn.resp_body == "true"
+  end
+
+  test "does not record a sighting when the endpoint id is malformed" do
+    request(endpoint_id: "not-a-valid-endpoint-id")
+
+    assert Store.sighting_count() == 0
+  end
+
+  test "accepts an uppercase endpoint id and records it downcased" do
+    upper = String.duplicate("AB", 32)
+
+    assert request(endpoint_id: upper).status == 200
+    assert {:ok, _} = Store.lookup_sighting(String.duplicate("ab", 32))
+  end
+
+  test "increments the sighting count across repeated calls" do
+    id = endpoint_id(3)
+
+    request(endpoint_id: id)
+    request(endpoint_id: id)
+
+    assert {:ok, {_first, _last, 2}} = Store.lookup_sighting(id)
+  end
+
+  test "does not record a sighting when the bearer is rejected" do
+    id = endpoint_id(4)
+
+    request(bearer: "wrong-token", endpoint_id: id)
+
+    assert Store.lookup_sighting(id) == :error
+  end
+end
diff --git a/metadata-relay/test/metadata_relay/p2p_access/schema_test.exs b/metadata-relay/test/metadata_relay/p2p_access/schema_test.exs
new file mode 100644
index 000000000..63fb9c6a0
--- /dev/null
+++ b/metadata-relay/test/metadata_relay/p2p_access/schema_test.exs
@@ -0,0 +1,42 @@
+defmodule MetadataRelay.P2pAccess.SchemaTest do
+  use ExUnit.Case, async: false
+
+  @moduletag :capture_log
+
+  alias MetadataRelay.P2pAccess.{Block, Sighting}
+  alias MetadataRelay.Repo
+
+  setup do
+    :ok = Ecto.Adapters.SQL.Sandbox.checkout(Repo)
+    :ok
+  end
+
+  @endpoint_id String.duplicate("ab", 32)
+
+  test "persists and reads back a sighting" do
+    now = DateTime.utc_now() |> DateTime.truncate(:second)
+
+    {:ok, _} =
+      Repo.insert(%Sighting{
+        endpoint_id: @endpoint_id,
+        first_seen: now,
+        last_seen: now,
+        conn_count: 3
+      })
+
+    assert %Sighting{conn_count: 3} = Repo.get(Sighting, @endpoint_id)
+  end
+
+  test "persists and reads back a block" do
+    now = DateTime.utc_now() |> DateTime.truncate(:second)
+
+    {:ok, _} =
+      Repo.insert(%Block{
+        endpoint_id: @endpoint_id,
+        reason: "bandwidth abuse",
+        blocked_at: now
+      })
+
+    assert %Block{reason: "bandwidth abuse"} = Repo.get(Block, @endpoint_id)
+  end
+end
diff --git a/metadata-relay/test/metadata_relay/p2p_access/store_test.exs b/metadata-relay/test/metadata_relay/p2p_access/store_test.exs
new file mode 100644
index 000000000..aa145f480
--- /dev/null
+++ b/metadata-relay/test/metadata_relay/p2p_access/store_test.exs
@@ -0,0 +1,387 @@
+defmodule MetadataRelay.P2pAccess.StoreTest do
+  use ExUnit.Case, async: false
+
+  @moduletag :capture_log
+
+  alias MetadataRelay.P2pAccess.Store
+
+  setup do
+    Store.init_tables()
+    :ets.delete_all_objects(:p2p_sightings)
+    :ets.delete_all_objects(:p2p_blocked)
+
+    original_cap = Application.get_env(:metadata_relay, :p2p_max_sightings)
+    on_exit(fn -> Application.put_env(:metadata_relay, :p2p_max_sightings, original_cap) end)
+
+    :ok
+  end
+
+  defp endpoint_id(n),
+    do: String.pad_leading(Integer.to_string(n, 16), 64, "0") |> String.downcase()
+
+  # The database-failure tests deliberately check the sandbox connection back
+  # in mid-test. Restoring it from on_exit rather than from a trailing
+  # statement means a failing assertion cannot leave the sandbox checked in
+  # and poison every test that runs after it.
+  defp restore_sandbox_on_exit do
+    on_exit(fn ->
+      :ok = Ecto.Adapters.SQL.Sandbox.checkout(MetadataRelay.Repo)
+      Ecto.Adapters.SQL.Sandbox.mode(MetadataRelay.Repo, {:shared, self()})
+    end)
+  end
+
+  defp eventually(fun, timeout_ms \\ 2_000) do
+    deadline = System.monotonic_time(:millisecond) + timeout_ms
+
+    Stream.repeatedly(fn ->
+      result = fun.()
+      unless result, do: Process.sleep(10)
+      result
+    end)
+    |> Enum.find(fn result ->
+      result or System.monotonic_time(:millisecond) > deadline
+    end)
+  end
+
+  test "records a first sighting with count 1" do
+    id = endpoint_id(1)
+
+    assert :ok = Store.record_sighting(id)
+    assert {:ok, {first_seen, last_seen, 1}} = Store.lookup_sighting(id)
+    assert first_seen == last_seen
+  end
+
+  test "increments the connection count on repeat sightings" do
+    id = endpoint_id(2)
+
+    :ok = Store.record_sighting(id)
+    :ok = Store.record_sighting(id)
+    :ok = Store.record_sighting(id)
+
+    assert {:ok, {_first, _last, 3}} = Store.lookup_sighting(id)
+  end
+
+  test "preserves first_seen across repeat sightings" do
+    id = endpoint_id(3)
+
+    :ok = Store.record_sighting(id)
+    {:ok, {first_seen, _, _}} = Store.lookup_sighting(id)
+
+    :ok = Store.record_sighting(id)
+
+    assert {:ok, {^first_seen, _, 2}} = Store.lookup_sighting(id)
+  end
+
+  test "stops recording new endpoints once the cap is reached" do
+    Application.put_env(:metadata_relay, :p2p_max_sightings, 2)
+
+    :ok = Store.record_sighting(endpoint_id(10))
+    :ok = Store.record_sighting(endpoint_id(11))
+    :ok = Store.record_sighting(endpoint_id(12))
+
+    assert Store.sighting_count() == 2
+    assert Store.lookup_sighting(endpoint_id(12)) == :error
+  end
+
+  test "keeps updating known endpoints after the cap is reached" do
+    Application.put_env(:metadata_relay, :p2p_max_sightings, 1)
+
+    id = endpoint_id(20)
+    :ok = Store.record_sighting(id)
+    :ok = Store.record_sighting(id)
+
+    assert Store.sighting_count() == 1
+    assert {:ok, {_first, _last, 2}} = Store.lookup_sighting(id)
+  end
+
+  describe "blocklist" do
+    setup do
+      :ok = Ecto.Adapters.SQL.Sandbox.checkout(MetadataRelay.Repo)
+      Ecto.Adapters.SQL.Sandbox.mode(MetadataRelay.Repo, {:shared, self()})
+      MetadataRelay.Repo.delete_all(MetadataRelay.P2pAccess.Block)
+      :ok
+    end
+
+    test "an unknown endpoint is not blocked" do
+      refute Store.blocked?(endpoint_id(30))
+    end
+
+    test "put_block marks the endpoint blocked in ETS" do
+      id = endpoint_id(31)
+
+      assert :ok = Store.put_block(id, "bandwidth abuse")
+      assert Store.blocked?(id)
+    end
+
+    test "put_block persists the block to the database" do
+      id = endpoint_id(32)
+
+      assert :ok = Store.put_block(id, "bandwidth abuse")
+
+      assert %MetadataRelay.P2pAccess.Block{reason: "bandwidth abuse"} =
+               MetadataRelay.Repo.get(MetadataRelay.P2pAccess.Block, id)
+    end
+
+    test "delete_block clears ETS and the database" do
+      id = endpoint_id(33)
+      :ok = Store.put_block(id, "mistake")
+
+      assert :ok = Store.delete_block(id)
+
+      refute Store.blocked?(id)
+      assert MetadataRelay.Repo.get(MetadataRelay.P2pAccess.Block, id) == nil
+    end
+
+    test "reload_blocks repopulates ETS from the database" do
+      id = endpoint_id(34)
+      :ok = Store.put_block(id, "bandwidth abuse")
+
+      # Simulate a fresh boot where ETS is empty but the row survives.
+      :ets.delete_all_objects(:p2p_blocked)
+      refute Store.blocked?(id)
+
+      assert :ok = Store.reload_blocks()
+      assert Store.blocked?(id)
+    end
+
+    test "a failed blocklist load reschedules itself and repopulates ETS when it succeeds" do
+      original = Application.get_env(:metadata_relay, :p2p_reload_retry_interval_ms)
+
+      on_exit(fn ->
+        Application.put_env(:metadata_relay, :p2p_reload_retry_interval_ms, original)
+      end)
+
+      Application.put_env(:metadata_relay, :p2p_reload_retry_interval_ms, 50)
+      restore_sandbox_on_exit()
+
+      # Take the sandbox connection away so the Store's Repo.all/1 raises,
+      # driving the boot-failure branch rather than the happy path. A Store
+      # that gave up here would run with an empty blocklist forever, which
+      # silently restores access for every revoked endpoint.
+      Ecto.Adapters.SQL.Sandbox.checkin(MetadataRelay.Repo)
+      send(Store, :reload_blocks)
+      # The GenServer handles messages in order, so this returns only once
+      # the failed attempt has been processed and the retry scheduled.
+      _ = :sys.get_state(Store)
+
+      # Hand the sandbox back so the scheduled retry finds a usable
+      # connection, and add the block the retry is expected to pick up.
+      :ok = Ecto.Adapters.SQL.Sandbox.checkout(MetadataRelay.Repo)
+      Ecto.Adapters.SQL.Sandbox.mode(MetadataRelay.Repo, {:shared, self()})
+
+      id = endpoint_id(36)
+
+      MetadataRelay.Repo.insert!(%MetadataRelay.P2pAccess.Block{
+        endpoint_id: id,
+        reason: "bandwidth abuse",
+        blocked_at: DateTime.utc_now() |> DateTime.truncate(:second)
+      })
+
+      refute Store.blocked?(id)
+      assert eventually(fn -> Store.blocked?(id) end)
+    end
+
+    test "put_block is idempotent and updates the reason" do
+      id = endpoint_id(35)
+
+      :ok = Store.put_block(id, "first reason")
+      :ok = Store.put_block(id, "second reason")
+
+      assert %MetadataRelay.P2pAccess.Block{reason: "second reason"} =
+               MetadataRelay.Repo.get(MetadataRelay.P2pAccess.Block, id)
+    end
+  end
+
+  describe "flush and prune" do
+    setup do
+      :ok = Ecto.Adapters.SQL.Sandbox.checkout(MetadataRelay.Repo)
+      Ecto.Adapters.SQL.Sandbox.mode(MetadataRelay.Repo, {:shared, self()})
+      MetadataRelay.Repo.delete_all(MetadataRelay.P2pAccess.Sighting)
+
+      # The Store is a singleton that outlives every test, and a flush only
+      # writes sightings touched since the previous one. Reset that window so
+      # a test starts from a clean one instead of inheriting whatever the
+      # previously run test left behind.
+      :sys.replace_state(Store, fn state -> %{state | last_flush_at: 0} end)
+
+      original = Application.get_env(:metadata_relay, :p2p_retention_seconds)
+      on_exit(fn -> Application.put_env(:metadata_relay, :p2p_retention_seconds, original) end)
+
+      :ok
+    end
+
+    test "flush writes ETS sightings to the database" do
+      id = endpoint_id(40)
+      :ok = Store.record_sighting(id)
+      :ok = Store.record_sighting(id)
+
+      assert {:ok, 1} = Store.flush_now()
+
+      assert %MetadataRelay.P2pAccess.Sighting{conn_count: 2} =
+               MetadataRelay.Repo.get(MetadataRelay.P2pAccess.Sighting, id)
+    end
+
+    test "flush updates an existing row rather than failing on conflict" do
+      id = endpoint_id(41)
+      :ok = Store.record_sighting(id)
+      {:ok, 1} = Store.flush_now()
+
+      :ok = Store.record_sighting(id)
+      assert {:ok, 1} = Store.flush_now()
+
+      assert %MetadataRelay.P2pAccess.Sighting{conn_count: 2} =
+               MetadataRelay.Repo.get(MetadataRelay.P2pAccess.Sighting, id)
+    end
+
+    test "flush with no sightings writes nothing" do
+      assert {:ok, 0} = Store.flush_now()
+    end
+
+    test "prune removes sightings older than the retention window" do
+      stale = endpoint_id(42)
+      fresh = endpoint_id(43)
+      now = System.system_time(:second)
+
+      :ets.insert(:p2p_sightings, {stale, now - 100, now - 100, 1})
+      :ets.insert(:p2p_sightings, {fresh, now, now, 1})
+
+      Application.put_env(:metadata_relay, :p2p_retention_seconds, 50)
+
+      assert {:ok, 1} = Store.prune_now()
+      assert Store.lookup_sighting(stale) == :error
+      assert {:ok, _} = Store.lookup_sighting(fresh)
+    end
+
+    test "prune also removes the database rows" do
+      stale = endpoint_id(44)
+      now = System.system_time(:second)
+
+      :ets.insert(:p2p_sightings, {stale, now - 100, now - 100, 1})
+      {:ok, 1} = Store.flush_now()
+
+      Application.put_env(:metadata_relay, :p2p_retention_seconds, 50)
+      {:ok, 1} = Store.prune_now()
+
+      assert MetadataRelay.Repo.get(MetadataRelay.P2pAccess.Sighting, stale) == nil
+    end
+
+    test "flush chunks a batch larger than one chunk and writes every row" do
+      now = System.system_time(:second)
+      rows = for n <- 1..3_000, do: {endpoint_id(100_000 + n), now, now, 1}
+      true = :ets.insert(:p2p_sightings, rows)
+
+      # A single insert_all/3 binds 4 parameters per row against SQLite's
+      # 32_766 limit, so this must be split across statements and the counts
+      # summed rather than reporting only the last chunk.
+      assert {:ok, 3_000} = Store.flush_now()
+
+      assert MetadataRelay.Repo.aggregate(MetadataRelay.P2pAccess.Sighting, :count) == 3_000
+    end
+
+    test "a second flush writes only what changed since the first" do
+      old = endpoint_id(47)
+      now = System.system_time(:second)
+      :ets.insert(:p2p_sightings, {old, now - 10, now - 10, 1})
+
+      assert {:ok, 1} = Store.flush_now()
+
+      # Nothing has been touched since, so the delta is empty.
+      assert {:ok, 0} = Store.flush_now()
+
+      fresh = endpoint_id(48)
+      :ok = Store.record_sighting(fresh)
+
+      assert {:ok, 1} = Store.flush_now()
+
+      assert %MetadataRelay.P2pAccess.Sighting{} =
+               MetadataRelay.Repo.get(MetadataRelay.P2pAccess.Sighting, fresh)
+    end
+
+    test "seeded sightings resume the persisted conn_count instead of resetting" do
+      id = endpoint_id(49)
+      :ok = Store.record_sighting(id)
+      :ok = Store.record_sighting(id)
+      {:ok, 1} = Store.flush_now()
+
+      # Simulate a restart: ETS starts empty, the row survives on disk.
+      :ets.delete_all_objects(:p2p_sightings)
+      assert :ok = Store.seed_sightings()
+      assert {:ok, {_first, _last, 2}} = Store.lookup_sighting(id)
+
+      :ok = Store.record_sighting(id)
+
+      assert {:ok, {_first, _last, 3}} = Store.lookup_sighting(id)
+      assert {:ok, 1} = Store.flush_now()
+      assert %{conn_count: 3} = MetadataRelay.Repo.get(MetadataRelay.P2pAccess.Sighting, id)
+    end
+
+    test "seeding respects the sighting cap" do
+      now = System.system_time(:second)
+
+      for n <- 1..5 do
+        :ets.insert(:p2p_sightings, {endpoint_id(50 + n), now - n, now - n, 1})
+      end
+
+      {:ok, 5} = Store.flush_now()
+      :ets.delete_all_objects(:p2p_sightings)
+
+      Application.put_env(:metadata_relay, :p2p_max_sightings, 2)
+
+      assert :ok = Store.seed_sightings()
+      assert Store.sighting_count() == 2
+      # Most recently seen first, so the two newest rows are the ones kept.
+      assert {:ok, _} = Store.lookup_sighting(endpoint_id(51))
+      assert {:ok, _} = Store.lookup_sighting(endpoint_id(52))
+    end
+
+    test "flush degrades to {:ok, 0} and keeps the Store alive when the database write fails" do
+      restore_sandbox_on_exit()
+
+      id = endpoint_id(45)
+      :ok = Store.record_sighting(id)
+
+      pid = Process.whereis(Store)
+      # Take the sandbox connection away so the Store's Repo.insert_all/3
+      # call has no ownership to use and raises DBConnection.OwnershipError,
+      # exercising the rescue in do_flush/0 instead of the happy path.
+      Ecto.Adapters.SQL.Sandbox.checkin(MetadataRelay.Repo)
+
+      assert {:ok, 0} = Store.flush_now()
+
+      # The point of the rescue: a failed database write must not take the
+      # Store down with it, since it owns the ETS tables the request path
+      # depends on.
+      assert Process.whereis(Store) == pid
+      assert Process.alive?(pid)
+    end
+
+    test "prune degrades to a rescued {:ok, count} and keeps the Store alive when the database delete fails" do
+      restore_sandbox_on_exit()
+
+      stale = endpoint_id(46)
+      now = System.system_time(:second)
+
+      :ets.insert(:p2p_sightings, {stale, now - 100, now - 100, 1})
+      Application.put_env(:metadata_relay, :p2p_retention_seconds, 50)
+
+      pid = Process.whereis(Store)
+      # Take the sandbox connection away so the Store's Repo.delete_all/1
+      # call has no ownership to use and raises DBConnection.OwnershipError,
+      # exercising the rescue in do_prune/0 instead of the happy path.
+      Ecto.Adapters.SQL.Sandbox.checkin(MetadataRelay.Repo)
+
+      assert {:ok, 1} = Store.prune_now()
+
+      # ETS eviction already happened before the database delete was
+      # attempted, so the count reflects it regardless of the database
+      # outcome.
+      assert Store.lookup_sighting(stale) == :error
+
+      # The point of the rescue: a failed database delete must not take the
+      # Store down with it, since it owns the ETS tables the request path
+      # depends on.
+      assert Process.whereis(Store) == pid
+      assert Process.alive?(pid)
+    end
+  end
+end
diff --git a/metadata-relay/test/metadata_relay/p2p_access_test.exs b/metadata-relay/test/metadata_relay/p2p_access_test.exs
new file mode 100644
index 000000000..81ef633c9
--- /dev/null
+++ b/metadata-relay/test/metadata_relay/p2p_access_test.exs
@@ -0,0 +1,192 @@
+defmodule MetadataRelay.P2pAccessTest do
+  use ExUnit.Case, async: false
+
+  @moduletag :capture_log
+
+  alias MetadataRelay.P2pAccess
+  alias MetadataRelay.P2pAccess.Store
+
+  setup do
+    :ok = Ecto.Adapters.SQL.Sandbox.checkout(MetadataRelay.Repo)
+    Ecto.Adapters.SQL.Sandbox.mode(MetadataRelay.Repo, {:shared, self()})
+
+    Store.init_tables()
+    :ets.delete_all_objects(:p2p_sightings)
+    :ets.delete_all_objects(:p2p_blocked)
+    MetadataRelay.Repo.delete_all(MetadataRelay.P2pAccess.Block)
+
+    original = Application.get_env(:metadata_relay, :p2p_access_bearer_tokens)
+    on_exit(fn -> Application.put_env(:metadata_relay, :p2p_access_bearer_tokens, original) end)
+
+    :ok
+  end
+
+  defp endpoint_id(n),
+    do: String.pad_leading(Integer.to_string(n, 16), 64, "0") |> String.downcase()
+
+  describe "normalize_endpoint_id/1" do
+    test "accepts 64 hex characters" do
+      assert {:ok, _} = P2pAccess.normalize_endpoint_id(endpoint_id(1))
+    end
+
+    test "downcases mixed-case input" do
+      upper = String.duplicate("AB", 32)
+      assert {:ok, lower} = P2pAccess.normalize_endpoint_id(upper)
+      assert lower == String.duplicate("ab", 32)
+    end
+
+    test "rejects a short id" do
+      assert :error = P2pAccess.normalize_endpoint_id("abc")
+    end
+
+    test "rejects 63 hex characters, one short of the boundary" do
+      assert :error = P2pAccess.normalize_endpoint_id(String.duplicate("a", 63))
+    end
+
+    test "rejects 65 hex characters, one over the boundary" do
+      assert :error = P2pAccess.normalize_endpoint_id(String.duplicate("a", 65))
+    end
+
+    test "rejects non-hex characters" do
+      assert :error = P2pAccess.normalize_endpoint_id(String.duplicate("z", 64))
+    end
+
+    test "rejects a nil id" do
+      assert :error = P2pAccess.normalize_endpoint_id(nil)
+    end
+  end
+
+  describe "authorize/1" do
+    test "allows an unknown endpoint" do
+      assert :allow = P2pAccess.authorize(endpoint_id(2))
+    end
+
+    test "records a sighting for an allowed endpoint" do
+      id = endpoint_id(3)
+      :allow = P2pAccess.authorize(id)
+
+      assert {:ok, {_first, _last, 1}} = Store.lookup_sighting(id)
+    end
+
+    test "denies a blocked endpoint" do
+      id = endpoint_id(4)
+      :ok = P2pAccess.block(id, "bandwidth abuse")
+
+      assert :deny = P2pAccess.authorize(id)
+    end
+
+    test "denies a blocked endpoint even when the caller sends a different case" do
+      id = endpoint_id(10)
+      :ok = P2pAccess.block(id, "bandwidth abuse")
+
+      assert :deny = P2pAccess.authorize(String.upcase(id))
+    end
+
+    test "still records a sighting for a blocked endpoint" do
+      id = endpoint_id(5)
+      :ok = P2pAccess.block(id, "bandwidth abuse")
+      :deny = P2pAccess.authorize(id)
+
+      assert {:ok, {_first, _last, 1}} = Store.lookup_sighting(id)
+    end
+
+    test "allows again after unblocking" do
+      id = endpoint_id(6)
+      :ok = P2pAccess.block(id, "mistake")
+      :deny = P2pAccess.authorize(id)
+
+      :ok = P2pAccess.unblock(id)
+
+      assert :allow = P2pAccess.authorize(id)
+    end
+  end
+
+  describe "valid_bearer?/1" do
+    test "accepts a configured token" do
+      Application.put_env(:metadata_relay, :p2p_access_bearer_tokens, ["secret-one"])
+      assert P2pAccess.valid_bearer?("secret-one")
+    end
+
+    test "accepts any token in a rotation list" do
+      Application.put_env(:metadata_relay, :p2p_access_bearer_tokens, ["old", "new"])
+      assert P2pAccess.valid_bearer?("old")
+      assert P2pAccess.valid_bearer?("new")
+    end
+
+    test "rejects an unconfigured token" do
+      Application.put_env(:metadata_relay, :p2p_access_bearer_tokens, ["secret-one"])
+      refute P2pAccess.valid_bearer?("wrong")
+    end
+
+    test "rejects nil" do
+      Application.put_env(:metadata_relay, :p2p_access_bearer_tokens, ["secret-one"])
+      refute P2pAccess.valid_bearer?(nil)
+    end
+
+    test "rejects everything when no tokens are configured" do
+      Application.put_env(:metadata_relay, :p2p_access_bearer_tokens, [])
+      refute P2pAccess.valid_bearer?("anything")
+      refute P2pAccess.valid_bearer?("")
+    end
+  end
+
+  describe "block/2 and unblock/1" do
+    test "rejects a malformed endpoint id" do
+      assert {:error, :invalid_endpoint_id} = P2pAccess.block("nope", "reason")
+      assert {:error, :invalid_endpoint_id} = P2pAccess.unblock("nope")
+    end
+  end
+
+  describe "list_recent/1" do
+    test "returns endpoints ordered by most recent activity" do
+      old = endpoint_id(7)
+      recent = endpoint_id(8)
+      now = System.system_time(:second)
+
+      :ets.insert(:p2p_sightings, {old, now - 100, now - 100, 1})
+      :ets.insert(:p2p_sightings, {recent, now, now, 5})
+
+      assert [%{endpoint_id: ^recent}, %{endpoint_id: ^old}] = P2pAccess.list_recent(10)
+    end
+
+    test "marks blocked endpoints" do
+      id = endpoint_id(9)
+      :allow = P2pAccess.authorize(id)
+      :ok = P2pAccess.block(id, "bandwidth abuse")
+
+      assert [%{endpoint_id: ^id, blocked: true}] = P2pAccess.list_recent(10)
+    end
+
+    test "includes endpoints seeded from the database at boot" do
+      id = endpoint_id(11)
+      now = DateTime.utc_now() |> DateTime.truncate(:second)
+
+      MetadataRelay.Repo.delete_all(MetadataRelay.P2pAccess.Sighting)
+
+      MetadataRelay.Repo.insert!(%MetadataRelay.P2pAccess.Sighting{
+        endpoint_id: id,
+        first_seen: now,
+        last_seen: now,
+        conn_count: 7
+      })
+
+      # Simulate a restart: ETS starts empty, the rows survive on disk.
+      # Without the boot seed an operator investigating an incident would
+      # only see endpoints seen since the last deploy.
+      :ets.delete_all_objects(:p2p_sightings)
+      assert :ok = Store.seed_sightings()
+
+      assert [%{endpoint_id: ^id, conn_count: 7}] = P2pAccess.list_recent(10)
+    end
+
+    test "honours the limit" do
+      now = System.system_time(:second)
+
+      for n <- 100..110 do
+        :ets.insert(:p2p_sightings, {endpoint_id(n), now, now, 1})
+      end
+
+      assert length(P2pAccess.list_recent(3)) == 3
+    end
+  end
+end