#!/usr/bin/env python3
# Copyright (c) 2026 The Sequentia developers
# Distributed under the MIT software license.
"""pignusd: the loan book, the chain watcher, and the page that shows them.

The daemon holds no keys and no funds. It cannot originate a loan, cannot settle
one, and cannot alter the terms of one -- the terms live inside the vault
address, and a borrower checks that address themselves before signing. So this
is an index and a viewer, and it is written to be obviously that.

    pignusd --config pignusd.json

Config:

    {
      "listen":   "127.0.0.1:8741",
      "book":     "/var/lib/pignus/book.json",
      "oracle":   "http://127.0.0.1:8740",
      "oracles":  ["http://127.0.0.1:8742", "http://127.0.0.1:8743"],
      "registry": "http://127.0.0.1:3005",
      "markets":  ["GOLD/USDX", "SILVR/USDX", "OILX/USDX"],
      "poll":     30,
      "block_seconds":    60,
      "reference_ticker": "USDX",
      "max_price_age":    600,
      "min_depth":        2,
      "rescan_depth":     1500,
      "back_scan_cap":    200,
      "prune_after":      2592000,
      "trusted_proxies":  ["127.0.0.1", "::1"],
      "rpc":      {"url": "http://127.0.0.1:18200", "user": "...",
                   "password": "...", "wallet": ""}
    }

`rpc` is optional: without a node the book still serves offers and whatever it
last knew, and says so in /healthz rather than pretending the states are fresh.
`registry` resolves each market's tickers to asset ids and precisions, so
nobody has to type a 64-hex id into a form; the node's own asset labels are
the fallback.

`oracle` is the PRIMARY oracle: its key is the one this book hands lenders and
its attestations are the prices shown. `oracles` are further independent
endpoints, quoted for threshold loans; they never stand in for the primary,
because a vault verifies against the key baked into it. A price older than
`max_price_age` seconds is stale: health, LTV and liquidatable are withheld
rather than computed from it, and /healthz turns unhealthy.

`min_depth` is how deep a funding must be before a loan reads as LIVE, and
`rescan_depth` how far back a poll walks to find a spend it missed.
`back_scan_cap` is how many blocks one poll may fetch for those backward walks:
the walk is the expensive half of a poll, so it is rationed and a search that
runs out of budget resumes on the next one. A watcher that was down for longer
than `rescan_depth` cannot reach the gap at all; start it once with
`--rescan-from <height>` and the forward scan reads those blocks again.
`prune_after` is how long a finished record is kept: an offer whose coin is
spent, a vault that closed, a cross-chain take that ended. The chain is the
record and this book is an index of it, so nothing pruned is lost -- but every
list is rendered from end to end on every read, and a record nothing can happen
to any more pays that for ever. Anything still open is kept whatever its age,
and a zero keeps everything.
`trusted_proxies` are the peers whose `X-Forwarded-For` this daemon believes,
which is how the write rate limit is keyed per client behind a reverse proxy;
a request arriving from a trusted proxy WITHOUT that header is the box's own
tooling and is not rate-limited.

The book follows the chain on its own. An offer's coin is watched; when it is
taken, the borrower's payout program is read out of the witness, the new vault
is registered as a loan, and the offer moves to its remainder. So a loan taken
by ANY wallet -- through this page or not -- turns up here, provided the take is
within the watcher's reach: forwards from wherever it last scanned, and back at
most `rescan_depth` blocks. After longer downtime than that, start it once with
`--rescan-from <height>`.

Endpoints:
    GET  /                     the page
    GET  /v1/markets           markets: prices, asset ids, tickers, precisions
    GET  /v1/assets            the registry, keyed by asset id
    GET  /v1/fees              the node's fee exchange rates and relay floor
    GET  /v1/offers            open offers  (?market=, ?status=all, ?limit=)
    GET  /v1/offers/{id}       one offer, with what its coin holds right now
    POST /v1/offers            publish a funded offer (checked on chain first)
    DELETE /v1/offers/{id}     drop a listing (the coin is untouched)
    GET  /v1/loans             known loans (?state=, ?market=, ?oracle_x=,
                               ?limit=)
    POST /v1/loans             register a loan for watching (checked first)
    GET  /v1/loans/{id}        one loan, with health at the current price
    GET  /v1/loans/{id}/exit   how a closed loan ended, read back off the chain
    GET  /v1/outpoint/{txid}/{vout}   what an outpoint holds (mempool included)
    GET  /v1/btc/outpoint/{txid}/{vout}  is a Bitcoin outpoint still there,
                               and if not, the witness that spent it
    GET  /v1/spend/{txid}/{vout}      the spend of an outpoint, and the
                                      preimages its witness published
    GET  /v1/scan/{scriptPubKey}      an unspent output at an address
                                      (?asset=, ?amount= in atoms)
    GET  /v1/attestation/{market}     the primary oracle's latest attestation
                               (?oracle=<x> for one other oracle's)
    GET  /v1/attestations/{market}    one verified attestation per oracle
    GET  /v1/stats             counts, live debt, positions near their strike
    GET  /v1/unrenderable      records this book holds and cannot show, and why
    GET  /v1/oracle            the primary oracle key this book quotes against
    GET  /v1/oracles           every oracle key, for an m-of-n threshold loan
    GET  /v1/vectors           the golden vectors the browser pins itself to
    GET  /v1/btc/offers        BTC-collateral offers        (?status=)
    GET  /v1/btc/offer/{id}    one offer, with the lots left on it
    POST /v1/btc/offers        a lender publishes one, signed by their own key
    POST /v1/btc/offers/{id}/withdraw   the same lender takes it down
    POST /v1/btc/offers/{id}/resign     ...or replaces its signature, and
                                        nothing else, over the stored terms
    POST /v1/btc/take          a borrower asks for a loan
    POST /v1/btc/hash          the lender draws this loan's secret and
                               publishes its hash, which both chains commit to
    POST /v1/btc/presig        the borrower signs the move into the vault,
                               which they can only derive once that hash exists
    GET  /v1/btc/takes         the queue   (?status=, ?offer_id=, ?borrower_x=,
                               ?oracle_x=)
    GET  /v1/btc/take/{id}     one take
    POST /v1/btc/adaptor       the lender returns a release, with its own secret
    POST /v1/btc/disbursed     the lender reports the principal paid
    POST /v1/btc/upgraded      the lender reports the loan started
    POST /v1/btc/claimed       the lender publishes the secret that releases
                               the collateral
    POST /v1/btc/refunded      the lender reports an unclaimed principal taken
                               back
    POST /v1/btc/claimed-principal   the borrower reports taking the principal
    POST /v1/btc/repaid        the borrower reports paying the debt
    GET  /healthz

`/v1/offer/{id}` and `/v1/loan/{id}` are accepted as aliases of the plural
forms. The page's own files in web/ are served by their bare name (.js, .json,
.css, .svg) and nothing else is.
"""

import argparse
import json
import os
import re
import subprocess
import sys
import gzip
import threading
import time
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs

sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), ".."))

from pignus import __version__, atoms               # noqa: E402
from pignus import compat                           # noqa: E402
from pignus import oracle as O                      # noqa: E402
from pignus import fees as F                        # noqa: E402
from pignus.book import Book, OfferExists, TakeMoved  # noqa: E402
from pignus.node import Node, RpcError              # noqa: E402
from pignus.ratelimit import RateLimiter            # noqa: E402
from pignus.offers import (                         # noqa: E402
    NotOnChain, check_outpoint, offer_address, offer_vault_address,
)
from pignus.terms import LoanTerms, feed_id         # noqa: E402
from pignus.watcher import (                            # noqa: E402
    State, VaultWatcher, CLOSED, _locktime_open,
)

WEB = os.path.join(os.path.dirname(os.path.realpath(__file__)), "..", "web")
MAX_BODY = 256 * 1024
# A book anyone can write to is a book anyone can flood. These are generous for
# a testnet and small enough that the whole file stays cheap to rewrite. They
# count what is still LIVE: a book that has seen 2000 listings over its life is
# not full, and a lifetime cap would eventually refuse everybody.
MAX_OFFERS = 2000
# How long a take nobody has signed holds a lender's lot.
#
# Short, because asking for a loan is free and anonymous: a request costs its
# sender one signed-by-nobody POST, and every second it holds a lot is a second
# a real borrower cannot take one. The handshake it is waiting for takes
# SECONDS -- a running responder reserves within one poll, five by default --
# so anything longer is not patience, it is an offer somebody else can hold
# shut for nothing. The write rate limiter bounds how fast one client can make
# them; this bounds how long each one costs.
#
# A take that WAS signed is a different thing and keeps its lot far longer
# (`BTC_SIGNED_TAKE_TTL`): the lender has committed to that one.
BTC_TAKE_TTL = 300
# Every value `?status=` accepts on an offer listing. A misspelling is refused
# rather than answered with an empty list, which reads as "there are none".
OFFER_STATUS_NAMES = frozenset(
    ("open", "delisted", "taken", "withdrawn", "gone", "ghost", "expired",
     "all"))

# The upgrade fee's floor, from the library that every other party checks it
# with. Typed here as well, it would drift the day one of them changed and the
# relay would accept an offer no responder would sign.
from pignus.btc_collateral import (                 # noqa: E402
    BIG_LOAN_FIELDS, MIN_UPGRADE_FEE)
# What a Bitcoin output has to hold to be worth anything: below this the network
# will not relay the transaction that spends it.
# As many cross-chain offers as this book will hold at once. They cost their
# publisher nothing -- a signature and no chain state -- and every one is
# re-serialised on every write, so without a cap one script fills the file.
MAX_BTC_OFFERS = 500
BTC_DUST = 330
# And how long a take that WAS signed holds one without the collateral ever
# appearing. A lender has committed to a signed take, so it keeps its place --
# but not for ever against a taker who never funded anything, or every lot of
# every offer could be consumed for the cost of a few requests.
BTC_SIGNED_TAKE_TTL = 6 * 3600
MAX_LOANS = 5000
# How many registrations one client may have waiting to bury at once. A loan is
# registered before its funding is confirmed, so a caller can hand the book a
# mempool transaction and then double-spend it: every one of those becomes a
# GHOST that stays in the book. A real borrower opens one loan at a time; this
# bounds what a flood of never-confirmed ones can cost.
MAX_PENDING_PER_CLIENT = 20
CLOSED_NAMES = {s.value for s in CLOSED}
# The states a registration is still waiting on: not yet buried, or buried and
# then undone. Both are cheap to manufacture, so both count against the cap.
PENDING_NAMES = {State.UNCONFIRMED.value, State.GHOST.value}
STATE_NAMES = {s.value for s in State}


# The amounts a book DERIVES rather than stores: what a seizure would pay at
# today's price, and what a close actually paid against what it should have.
# They are amounts like any other and go out as decimal strings for the same
# reason -- a browser rounds a JSON number past 2^53, and these are the
# figures a lender checks a payout against, where a rounded one reads as a
# shortfall that is not there.
# The most a book may take as a reclaim fee: never more in absolute terms than
# a reclaim could plausibly cost, never more than a fifth of the collateral,
# and never so little that an honest one cannot relay. Both bounds bind, which
# is why this is the smaller of them. `reclaimFeeCap` in web/btcborrow.js is
# the same function; a borrower's page refuses anything above it.
RECLAIM_FEE_CEILING = 50_000
# A reclaim is about this many vbytes: one taproot script-path input, one
# output, and a witness carrying two signatures and a 32-byte preimage.
RECLAIM_VSIZE = 150
# ...and the least a reclaim fee may be, whatever the chain says: below about
# 1 sat/vB nothing relays it at all.
MIN_RECLAIM_FEE = RECLAIM_VSIZE


def reclaim_fee_cap(collateral):
    return min(RECLAIM_FEE_CEILING, max(BTC_DUST * 10, int(collateral) // 5))


def reclaim_fee_floor(feerate_sat_vb=None):
    """The LEAST a reclaim fee may be, on a chain charging `feerate_sat_vb`.

    The cap bounds what a relay may TAKE; this bounds what it may LEAVE, and
    the second half was missing. The lender signs a release over exactly one
    transaction, paying the collateral less this fee, and it can be neither
    replaced nor fee-bumped -- so a fee too small to relay is not a slow
    reclaim, it is a reclaim that never happens and a collateral that sits in
    the vault until the lender's own timeout sweep takes it.
    """
    try:
        r = float(feerate_sat_vb or 0)
    except (TypeError, ValueError):
        r = 0.0
    priced = int(r * RECLAIM_VSIZE + 0.999) if r > 0 else 0
    return max(MIN_RECLAIM_FEE, priced)


DERIVED_AMOUNTS = ("seizure_if_liquidated", "surplus_if_liquidated",
                   "seize_paid", "seize_expected", "surplus_paid",
                   "surplus_expected", "lender_paid", "debt", "collateral")


def as_amounts(out, keys=DERIVED_AMOUNTS):
    """The given keys as decimal strings, leaving an absent one absent."""
    for k in keys:
        v = out.get(k)
        if isinstance(v, int) and not isinstance(v, bool):
            out[k] = str(v)
    return out


# Market symbols that are not registry tickers. The Sequence token is listed
# as tSEQ on the testnet; native Bitcoin has no Sequentia asset id at all, so a
# BTC market is cross-chain: it is not lendable through the covenant offers,
# and is served by the /v1/btc/* endpoints instead.
ALIASES = {"SEQ": "tSEQ", "TSEQ": "tSEQ"}
NATIVE = {"BTC", "TBTC"}
# What a client may say about an offer. Everything else about a record -- its
# id, its status, when it was created, what its coin holds -- is the book's to
# decide or this daemon's to read off the chain, or a publisher could overwrite
# somebody else's listing by naming it. `manage_token` is passed through
# because the book uses it to recognise a REPUBLISH of the same coin by the
# same publisher; a listing that does not exist yet is not authenticated by it.
OFFER_FIELDS = ("terms", "kind", "outpoint", "principal", "collateral",
                "expiry_locktime", "manage_token")


WRITE_LIMIT = RateLimiter()
# A second bucket for everyone together, so a flood spread over many source
# addresses still cannot rewrite the book file thousands of times a second.
GLOBAL_WRITE_LIMIT = RateLimiter(rate=20.0, burst=200)
# Reads that touch the NODE, which is a different cost from reads served out of
# memory: `/v1/spend` walks blocks backwards and `/v1/outpoint` asks the node
# per call. Both are unauthenticated and both must stay so -- a borrower
# recovering their own collateral has no account here -- so they are rationed
# instead. Roomier than the write limit, because a page legitimately makes a
# handful in a row.
SCAN_LIMIT = RateLimiter(rate=2.0, burst=30)
# The whole-book listings -- `/v1/loans`, `/v1/offers`, `/v1/btc/offers`,
# `/v1/stats`. Served from a render cache, so a request is cheap while the
# cache holds and a full render when it does not; the page polls each every
# thirty seconds, so two a second is thirty times what any tab needs.
LIST_LIMIT = RateLimiter(rate=2.0, burst=30)
# Renders that may run at once. A render of a full book is half a second
# of CPU and tens of megabytes; past this many the daemon says "busy" for
# five seconds rather than queue requests without bound, each holding a
# thread and its memory, which is how one flood turned into an OOM.
RENDER_SLOTS = 8
# How long a render may be served from the cache without a change to the
# book or the poll behind it. An attestation goes stale by the clock alone,
# so a render cannot live on the epochs alone.
RENDER_TTL = 10
# Connections the server holds at once. Beyond this an accept is closed
# unread: a client that then retries gets a slot when one frees, and one
# holding hundreds of sockets open cannot make the server hold hundreds of
# threads for it.
MAX_CONNECTIONS = 256
GLOBAL_SCAN_LIMIT = RateLimiter(rate=40.0, burst=400)

# Records that cannot be rendered, and the error each last failed with. Every
# list is fetched every few seconds by every open page and by the liquidator,
# so one bad record is logged when it starts failing rather than once per read.
UNRENDERABLE = {}


def served_loan(rec):
    """A stored record with its loan's big amounts as decimal strings.

    Publishing coerces them at the door, which is right and is where a bad
    value has to be refused -- but it only reaches records written since. A
    record stored before that lives on with plain JSON numbers in it, and a
    browser parses an amount past 2**53 into a different number without saying
    so. The wire format is a property of the ENDPOINT, not of the day a record
    happened to be written, so it is applied on the way out as well.

    Copies, always: the stored dict is the book's, and a view is a view.
    """
    if not isinstance(rec, dict):
        return rec
    loan = rec.get("loan")
    if not isinstance(loan, dict):
        return dict(rec)
    fixed = dict(loan)
    for k in BIG_LOAN_FIELDS:
        v = fixed.get(k)
        if isinstance(v, bool) or v is None or v == "":
            continue
        if isinstance(v, int) or (isinstance(v, str) and v.strip().lstrip("-").isdigit()):
            fixed[k] = str(int(v))
    return dict(rec, loan=fixed)


def _is_hex(text):
    return bool(text) and all(c in "0123456789abcdefABCDEF" for c in text)


def _git_rev():
    """The checkout this daemon is running from, when there is one.

    Two boxes reporting the same version number can still be running different
    code; the revision is what settles it. Read once, at start-up, and left out
    entirely when there is no checkout to read -- a release tarball is not a
    working tree, and inventing a revision for it would be worse than silence.
    """
    root = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
    try:
        out = subprocess.run(["git", "-C", root, "rev-parse", "--short", "HEAD"],
                             capture_output=True, text=True, timeout=5)
    except (OSError, subprocess.SubprocessError):
        return None
    if out.returncode != 0:
        return None
    return out.stdout.strip() or None


GIT_REV = _git_rev()


def _limit(rows, q):
    """`?limit=N` over a list that is already newest-first. The default is the
    whole book, because the page filters states itself and would otherwise
    silently stop showing a reader their own older loans."""
    try:
        n = int(q.get("limit", ["0"])[0])
    except (TypeError, ValueError):
        raise ValueError("limit must be a number")
    if n < 0:
        raise ValueError("limit must not be negative")
    return rows[:n] if n else rows


def _attested(att, oracle_x, max_age):
    """One attestation as it is served: the signed fields untouched, with who
    signed it and how old it is beside them.

    Age is not a signed property and cannot be -- an attestation is a signature
    over a price at a timestamp, and nothing about it changes as it ages. But a
    reader acting on it needs to know, and a page that shows an hour-old price
    without saying so is showing fiction. Readers that only want the signed
    fields drop these on the way in.
    """
    d = att.to_dict()
    d["oracle_x"] = oracle_x
    # The age SIGNED, not clamped at zero: an attestation dated ahead of this
    # machine's clock is the one case a reader most needs to see, because it is
    # what makes a dead price read as a current one.
    d["age"] = O.age_of(att)
    d["stale"] = not O.current(att, max_age)
    return d


def _loan_oracle_keys(rec):
    """Every oracle key a stored loan's document names, lower-cased.

    Read straight out of the terms rather than through LoanTerms: this backs a
    filter, and a filter must not fail on the one record that cannot be built.
    Both `oracle_x` and the threshold set are included, because the question it
    answers -- may this key be retired yet? -- is better answered generously.
    """
    try:
        d = json.loads(rec.get("terms") or "{}")
    except (TypeError, ValueError):
        return ()
    keys = list(d.get("oracles") or []) + [d.get("oracle_x")]
    return tuple(str(k).lower() for k in keys if k)


def _views(render, rows):
    """Render each record, skipping any that cannot be rendered rather than
    500-ing the whole list because one stored record is malformed."""
    out = []
    for r in rows:
        rid = r.get("offer_id") or r.get("loan_id") or "?"
        try:
            out.append(render(r))
        except Exception as e:                      # noqa: BLE001
            msg = f"{type(e).__name__}: {e}"
            if UNRENDERABLE.get(rid) != msg:
                sys.stderr.write(f"skipping unrenderable record {rid}: {msg}\n")
            UNRENDERABLE[rid] = msg
        else:
            UNRENDERABLE.pop(rid, None)
    return out


class Service:
    def __init__(self, cfg):
        self.cfg = cfg
        # The book is rewritten through a temporary file in its own directory,
        # so that directory has to exist before the first write rather than at
        # the first POST, which is hours later and looks like a different fault.
        os.makedirs(os.path.dirname(os.path.abspath(cfg["book"])) or ".",
                    exist_ok=True)
        self.book = Book(cfg["book"])
        self.oracle_url = cfg.get("oracle", "").rstrip("/")
        # One book can quote against several INDEPENDENT oracles, which is what
        # a threshold (m-of-n) loan needs. `oracle` is the default single one;
        # `oracles` is a list of additional endpoints. Each has its own key.
        urls = []
        if self.oracle_url:
            urls.append(self.oracle_url)
        for u in cfg.get("oracles", []):
            u = str(u).rstrip("/")
            if u and u not in urls:
                urls.append(u)
        self.oracle_urls = urls
        # The key each oracle is EXPECTED to serve, parallel to the URL list
        # (the primary first, then `oracles`); an empty entry pins nothing.
        # Unpinned, the book adopts whatever `/v1/pubkey` serves on every
        # poll: a lost-and-recreated key, or a substituted one on a plain-http
        # URL, is accepted silently while every loan baked to the old key
        # goes unpriced.
        keys = [str(k).lower() for k in cfg.get("oracle_keys", [])]
        self.oracle_expected = {u: keys[i] for i, u in enumerate(urls)
                                if i < len(keys) and keys[i]}
        self.compromised_keys = set()       # declared by any quoted oracle
        self.previous_keys = {}             # url -> keys it used to sign with
        self.oracle_keys = []                       # x-only hex, in url order
        self.oracle_key_by_url = {}
        self.att_by_oracle = {}                     # (oracle_x, market) -> att
        # What each oracle says the two assets of a market are denominated in.
        # The price is an integer scaled by those precisions, so an oracle
        # working from one set and a registry publishing another do not disagree
        # about a label -- they disagree about the number, by a factor of ten
        # per decimal, and nothing on chain would notice.
        self.oracle_precisions = {}                 # (oracle_x, market) -> (c,d)
        # The scale the PRIMARY quotes each market at. An attestation carries a
        # perfectly good signature over an integer whose meaning depends on it.
        self.price_scales = {}                      # market -> price_scale
        self.registry_url = cfg.get("registry", "").rstrip("/")
        self.markets = list(cfg.get("markets", []))
        # The last VERIFIED attestation per market, whatever its age. The
        # prices acted on are the young ones, from fresh_prices().
        self.attestations = {}
        self.oracle_x = None
        self.node = None
        self.node_error = None
        self.oracle_error = None
        self.oracle_errors = []
        # Offer events this daemon dropped because it could never apply them,
        # and how many are still queued. Both are served by /healthz: a queue
        # that only grows is a book falling behind its own chain, and nothing
        # else would say so.
        self.event_errors = []
        self.event_backlog = 0
        self._mtp = None
        self._mtp_at = 0
        # Closed-loan accounts, keyed by (loan, spend, height). A mined close
        # is a fact about a block and cannot change, and rebuilding one pulls a
        # whole verbose block.
        self._exit_cache = {}
        # And the spends `/v1/spend` finds, which walk blocks to find. Mined
        # ones only: a mempool spend can be replaced.
        self.spend_cache = {}
        self.last_poll = 0
        self.rescan_needed_from = None
        self._absence_checked = False
        # What the last poll raised, if it raised. The thread
        # survives one, so this is the only place a caller can
        # see that everything the book knows stopped updating.
        self.poll_error = None
        self._btc_feerate = None
        self._btc_feerate_at = 0
        self._btc_height = None
        self._btc_height_at = 0
        # The render cache; see `Handler._listing`. `view_epoch` is bumped
        # after every poll step, so a render made before new prices, a new
        # tip or a re-read registry is never served after them.
        self.view_epoch = 0
        self._render_cache = {}
        self._render_lock = threading.Lock()
        self._render_slots = threading.BoundedSemaphore(RENDER_SLOTS)
        self.height = None
        self.assets = {}            # asset id -> {ticker, name, precision}
        self.by_ticker = {}         # TICKER (upper) -> asset id
        self.fee_table = F.empty_table()
        self._registry_at = 0
        self.poll_interval = max(2, int(cfg.get("poll", 30)))
        # How deep a funding must be before a loan reads as LIVE. The design
        # says depth is the lender's risk appetite, so it is configuration, not
        # a constant -- and the page shows the number it is actually using.
        self.min_depth = int(cfg.get("min_depth", 2))
        self.rescan_depth = int(cfg.get("rescan_depth", 1500))
        # The backward walk is the expensive half of a poll. This is how many
        # blocks one poll may fetch for it, so a book with a long history cannot
        # spend an entire poll reading it back.
        self.back_scan_cap = int(cfg.get("back_scan_cap", 200))
        # How long a finished record is kept. Thirty days is long past every
        # deadline a loan has and long past when anyone reads back how one
        # ended, and short enough that a book serving a busy market does not
        # render years of history on every fetch.
        self.prune_after = int(cfg.get("prune_after", 30 * 24 * 3600))
        # A price older than this is not a price. Health, LTV and liquidatable
        # are withheld rather than computed from it.
        self.max_price_age = int(cfg.get("max_price_age", 600))
        self.trusted_proxies = set(str(p) for p in
                                   cfg.get("trusted_proxies",
                                           ["127.0.0.1", "::1"]))
        rpc = cfg.get("rpc")
        if rpc and rpc.get("url"):
            self.node = Node(url=rpc["url"], user=rpc.get("user"),
                             password=rpc.get("password"),
                             wallet=rpc.get("wallet") or None,
                             cookie_path=rpc.get("cookie"))
        # A Bitcoin node, if this book is asked to carry cross-chain messages.
        # It never spends anything: it is here so the book reads the parent
        # chain's tip itself rather than believing a height a client sent, and
        # so it can say plainly when it cannot see that chain at all.
        self.btc_node = None
        btc_rpc = cfg.get("btc_rpc")
        if btc_rpc and btc_rpc.get("url"):
            self.btc_node = Node(url=btc_rpc["url"], user=btc_rpc.get("user"),
                                 password=btc_rpc.get("password"),
                                 wallet=btc_rpc.get("wallet") or None,
                                 cookie_path=btc_rpc.get("cookie"))
        self.watcher = (VaultWatcher(self.node, min_depth=self.min_depth,
                                     rescan_depth=self.rescan_depth,
                                     back_scan_cap=self.back_scan_cap)
                        if self.node else None)
        # Which client registered which loan, so a flood of registrations that
        # never bury is charged to whoever made it. Swept as it is read.
        self._pending_by_client = {}
        # A vault address is a pure function of the terms, and deriving one is
        # script assembly plus an EC tweak. /v1/loans renders every loan on
        # every fetch, so the derivations are kept.
        self._spk_cache = {}
        self._lock = threading.Lock()
        # refresh_chain runs on the poll thread AND synchronously inside a POST
        # handler thread; VaultWatcher.poll mutates shared state, so the two
        # must never run at once.
        self._chain_lock = threading.Lock()
        self.block_seconds = int(cfg.get("block_seconds", 60))
        # The reference/numeraire the page quotes value in. Markets are
        # "COLLATERAL/DEBT"; the debt asset of a market whose debt ticker
        # matches this is worth 1 reference unit. Configurable, not hardcoded.
        self.reference_ticker = cfg.get("reference_ticker", "USDX")
        # Where a reader of this book's page should be sent to look at a
        # transaction, an explorer or the oracle's own log. Configurable
        # because a self-hosted book is not behind the testnet's reverse proxy
        # and its neighbours live somewhere else; the page falls back to its
        # own defaults for whatever is left unset, which is why these are empty
        # strings rather than guesses.
        self.explorer_url = cfg.get("explorer_url", "").rstrip("/")
        self.explorer_tx_url = cfg.get("explorer_tx_url", "")
        self.btc_explorer_tx_url = cfg.get("btc_explorer_tx_url", "")
        self.oracle_public_url = cfg.get("oracle_public_url", "").rstrip("/")
        # Where each oracle can be reached FROM OUTSIDE, in the same order as
        # `oracle` then `oracles`. These are almost never the addresses this
        # book uses: a book talks to its oracles over loopback, and serving
        # those to a browser is worse than serving nothing, because a page that
        # followed one would be fetching from the reader's own machine.
        #
        # It matters because a seizure is meant to be auditable by anyone: the
        # attestation behind it is published at that oracle's /v1/seizures, and
        # an m-of-n seizure is signed by oracles that are not the primary. A
        # threshold loan whose extra oracles are unreachable is one whose
        # seizure nobody outside can check, which is most of what bounds an
        # oracle's trust. Empty where none is configured, never a guess.
        pub = [str(u).rstrip("/") for u in cfg.get("oracle_public_urls", [])]
        if not pub and self.oracle_public_url:
            pub = [self.oracle_public_url]
        self.oracle_public_urls = pub

    # ------------------------------------------------------------- the oracle

    @staticmethod
    def _att_key(oracle_x, market):
        """How an attestation is filed: by key and market, both canonicalised.

        The covenant compares oracle keys as BYTES and derives the feed id from
        an upper-cased market name, so `AB…` and `ab…` are one oracle to it, and
        `gold/usdx` is one market. Keying on the strings verbatim would file a
        loan's own oracle under a name it never looks up, and the loan would
        show no price at all."""
        return (str(oracle_x).lower(), str(market).upper())

    def refresh_prices(self):
        if not self.oracle_urls:
            return
        primary = self.oracle_urls[0]
        key_by_url, errors = {}, []
        compromised, previous = set(), {}
        primary_err = None              # why the PRIMARY is out, when it is
        for url in self.oracle_urls:
            try:
                with urllib.request.urlopen(f"{url}/v1/pubkey", timeout=8) as r:
                    pub = json.loads(r.read().decode())
                served = str(pub["oracle_x"]).lower()
            except Exception as e:                  # noqa: BLE001
                errors.append(f"{url}: {type(e).__name__}: {e}")
                if url == primary:
                    primary_err = errors[-1]
                continue
            want = self.oracle_expected.get(url)
            if want and served != want:
                # Not adopted. The vaults bake the pinned key; a different one
                # here is a lost key recreated, or a substitution in transit.
                errors.append(f"{url}: serves key {served[:16]}…, not the "
                              f"{want[:16]}… this book is configured to expect; "
                              f"its attestations are ignored until the key "
                              f"is restored or the configuration changed")
                if url == primary:
                    primary_err = errors[-1]
                continue
            # Only the declarer's OWN keys: the one it serves and the ones it
            # used to sign with. Otherwise any quoted oracle -- or whatever
            # sits on the path to one -- could knock every other oracle out
            # of this book by naming its key.
            prev = [str(k).lower() for k in pub.get("previous") or []]
            own = set(prev) | {served}
            for k in pub.get("compromised") or []:
                k = str(k).lower()
                if k in own:
                    compromised.add(k)
                else:
                    errors.append(f"{url}: declares {k[:16]}… compromised but "
                                  f"does not list it as its own key or a "
                                  f"previous one; ignored")
            previous[url] = prev
            key_by_url[url] = served
        for url, ox in list(key_by_url.items()):
            if ox in compromised:
                errors.append(f"{url}: its key {ox[:16]}… is declared "
                              f"compromised; nothing it signs is accepted")
                if url == primary:
                    primary_err = errors[-1]
                del key_by_url[url]
        with self._lock:
            # Configured order, not the order they happened to answer in.
            self.oracle_keys = [key_by_url[u] for u in self.oracle_urls
                                if u in key_by_url]
            self.oracle_key_by_url = key_by_url
            self.compromised_keys = compromised
            self.previous_keys = previous
            # The FIRST oracle is the "primary": its price is what the page
            # shows and what a plain single-oracle loan bakes in. An outage of
            # the primary must NOT promote the next oracle. A vault verifies
            # against the key baked into it, so serving another signer's key
            # and attestations would hand lenders a key their loans do not use
            # and break the check on every loan already open. Keep the last key
            # known, stop updating the price, and say so in /healthz.
            if primary in key_by_url:
                self.oracle_x = key_by_url[primary]
        # Decided by the flag set where the primary was actually dropped,
        # not by scanning every note that names its URL: a declaration the
        # book merely ignored is a note, and read as an outage it took the
        # book to unhealthy with the primary answering and its prices fresh.
        self.oracle_error = (f"primary oracle unreachable: {primary_err}"
                             if primary_err else None)
        fresh_primary = 0
        for url in self.oracle_urls:
            ox = key_by_url.get(url)
            if not ox:
                continue
            self._refresh_oracle_precisions(url, ox, errors)
            for m in self.markets:
                try:
                    a_url = f"{url}/v1/attestation/{m.replace('/', '_')}"
                    with urllib.request.urlopen(a_url, timeout=8) as r:
                        att = O.Attestation.from_dict(
                            json.loads(r.read().decode()))
                except urllib.error.HTTPError as e:
                    # A 404 is "no attestation yet"; anything else -- a 429
                    # from the oracle's own archive ration, a 503 -- is the
                    # oracle erroring, and is said, not read as no price.
                    if e.code != 404:
                        errors.append(f"{url}: {m}: HTTP {e.code}")
                    continue
                except Exception:                   # noqa: BLE001
                    continue
                # Verify before storing. A price this service displays is a
                # price someone may act on, and an oracle is trusted for the
                # number, not for the transport that carried it.
                if not O.verify(ox, att):
                    continue
                # EVERY oracle's precisions, not only the primary's. A
                # secondary configured to 2 decimals where the registry says
                # 8 is a good signature over a number a millionfold off, and
                # the lowest price is what a threshold seizure presents first.
                said = self.oracle_precisions.get(self._att_key(ox, m))
                reg = self._registry_precisions(m)
                if said and reg and tuple(said) != reg:
                    errors.append(
                        f"{url}: prices {m} at precisions {list(said)}, but "
                        f"the registry denominates it at {list(reg)}; that "
                        f"attestation was dropped")
                    continue
                scale = int(att.price_scale)
                if url == primary:
                    self.price_scales[m] = scale
                elif self.price_scales.get(m) not in (None, scale):
                    # A good signature over a number that means something else.
                    # The scale is not in the signed message, so a second oracle
                    # quoting at ten times the primary's is not detectable from
                    # the signature -- and quoting one against the other would
                    # make a healthy loan look liquidatable by a factor of ten.
                    errors.append(
                        f"{url}: quotes {m} at price_scale {scale}, but the "
                        f"primary quotes it at {self.price_scales[m]}; that "
                        "attestation is not comparable and was dropped")
                    continue
                with self._lock:
                    self.att_by_oracle[self._att_key(ox, m)] = att
                    if url == primary:
                        fresh_primary += 1
                        self.attestations[m] = att
        if primary_err is None and self.markets and not fresh_primary:
            self.oracle_error = (f"primary oracle {primary} published no "
                                 "attestation this book could verify")
        self.oracle_errors = errors
        # With the prices in hand, remember which loans crossed and when.
        try:
            self.book.stamp_at_risk(self.price_for)
        except Exception as e:                          # noqa: BLE001
            print(f"at-risk stamp: {type(e).__name__}: {e}", file=sys.stderr)

    def _oracle_compromised(self, t):
        """Does this loan bake a key some quoted oracle has declared
        compromised? Whoever holds that key can sign any price for it."""
        keys = {str(t.oracle_x or "").lower()} | {
            str(k).lower() for k in (t.oracle_keys or ())}
        return bool(keys & self.compromised_keys)

    def _registry_precisions(self, market):
        """(collateral, debt) decimals the registry gives a market's assets,
        or None when either asset is unknown to it."""
        c_sym, _, d_sym = market.partition("/")
        c_asset, _ct = self.resolve(c_sym)
        d_asset, _dt = self.resolve(d_sym)
        if not c_asset or not d_asset:
            return None
        return (int((self.assets.get(c_asset) or {}).get("precision", 8)),
                int((self.assets.get(d_asset) or {}).get("precision", 8)))

    def _refresh_oracle_precisions(self, url, ox, errors):
        """What each oracle believes the two assets of a market are denominated
        in. The price is an integer scaled by exactly those two precisions, so
        an oracle configured against 8 decimals and a registry publishing 2
        agree about every label and disagree about the number by a millionfold.
        Nothing on chain would notice, which is why it is checked here."""
        try:
            with urllib.request.urlopen(f"{url}/v1/markets", timeout=8) as r:
                rows = json.loads(r.read().decode()).get("markets") or []
        except Exception as e:                      # noqa: BLE001
            errors.append(f"{url}/v1/markets: {type(e).__name__}: {e}")
            return
        with self._lock:
            for row in rows:
                m = str(row.get("market") or "")
                cp = row.get("collateral_precision")
                dp = row.get("debt_precision")
                if not m or cp is None or dp is None:
                    continue
                self.oracle_precisions[self._att_key(ox, m)] = (int(cp), int(dp))

    def fresh_prices(self):
        """The prices young enough to act on. A price kept from hours ago is
        still a verified attestation -- freshness is not a covenant property --
        but it is not something to compute a liquidation from."""
        now = int(time.time())
        return {m: a.price for m, a in self.attestations.items()
                if O.current(a, self.max_price_age, now)}

    def stale_markets(self):
        fresh = self.fresh_prices()
        return sorted(m for m in self.markets if m not in fresh)

    def price_for(self, t):
        """The price to judge ONE set of terms by, or None.

        Not the book's primary price: a vault verifies against the keys baked
        into it, so a loan built on another oracle -- one left behind by a key
        rotation, or a threshold set -- has to be shown at those keys' price or
        at no price at all. A number from a key the covenant will not accept is
        a number nobody can act on, and putting it on the page as health and
        LTV would invite exactly that.

        The attestations here are already verified against the key they are
        filed under; what is checked again is what depends on the terms rather
        than on the oracle: the scale the loan computes at, the timestamp floor
        the covenant enforces, and how many the vault requires. The covenant
        takes the MAXIMUM of the prices presented and the best a spender can
        present is the `threshold` lowest, so the threshold-th lowest is the
        number the script would actually compute with.
        """
        now = int(time.time())
        prices = []
        for k in t.oracle_keys:
            att = self.att_by_oracle.get(self._att_key(k, t.market))
            if att is None or not O.current(att, self.max_price_age, now):
                continue
            if int(att.price_scale) != int(t.price_scale):
                continue
            if att.timestamp < t.not_before:
                continue
            prices.append(att.price)
        if len(prices) < t.threshold:
            return None
        return sorted(prices)[t.threshold - 1]

    # ----------------------------------------------------------- the registry

    def refresh_registry(self, force=False):
        """Asset ids and precisions by ticker, from the registry's minimal
        index, with the node's own labels underneath it."""
        if not force and time.time() - self._registry_at < 600:
            return
        found = {}
        if self.node:
            try:
                for label, asset in (self.node.dumpassetlabels() or {}).items():
                    if len(asset) == 64 and label != "bitcoin":
                        found[asset] = {"ticker": label, "name": label,
                                        "precision": 8}
            except Exception:                       # noqa: BLE001
                pass
        if self.registry_url:
            try:
                with urllib.request.urlopen(
                        f"{self.registry_url}/index.minimal.json",
                        timeout=8) as r:
                    idx = json.loads(r.read().decode())
                for asset, row in idx.items():
                    if len(asset) != 64 or not isinstance(row, list):
                        continue
                    ticker = str(row[1] or "")
                    if not ticker:
                        continue
                    # 0..18, whatever the registry says: the page raises ten
                    # to this power, and a precision of -1 or 10**9 from one
                    # row took every panel down with it.
                    prec = int(row[3] or 0) if row[3] is not None else 8
                    found[asset] = {"ticker": ticker, "name": str(row[2] or ticker),
                                    "precision": min(18, max(0, prec))}
            except Exception as e:                  # noqa: BLE001
                print(f"registry: {type(e).__name__}: {e}", file=sys.stderr)
        if found:
            with self._lock:
                self.assets = found
                self.by_ticker = {v["ticker"].upper(): k for k, v in found.items()}
        self._registry_at = time.time()

    def resolve(self, symbol):
        """A market symbol -> (asset id | None, ticker)."""
        sym = symbol.strip()
        up = sym.upper()
        if up in NATIVE:
            return None, "BTC"
        ticker = ALIASES.get(up, sym)
        asset = self.by_ticker.get(ticker.upper()) or self.by_ticker.get(up)
        return asset, (self.assets[asset]["ticker"] if asset else ticker)

    # ------------------------------------------------------------ the watcher

    def refresh_fees(self):
        if not self.node:
            return
        try:
            self.fee_table = F.fee_table(self.node)
        except Exception as e:                      # noqa: BLE001
            print(f"fees: {type(e).__name__}: {e}", file=sys.stderr)

    def refresh_chain(self, wait=0):
        """Reconcile the book to the chain.

        `wait` is how long a caller will hold on for the poll thread to finish
        the reconciliation it is in the middle of. The loop passes nothing and
        skips -- it will come round again in a few seconds. A POST passes a
        second or two, because the record it is about to serve back is the one
        the caller will read the confirmations off: skipping there answers
        UNCONFIRMED for a funding the chain has already buried.
        """
        if not self.watcher:
            return
        w = self.watcher
        got = (self._chain_lock.acquire(timeout=wait) if wait > 0
               else self._chain_lock.acquire(blocking=False))
        if not got:
            return          # a reconcile is already in flight; let it finish
        try:
            self.height = self.node.getblockcount()
            # The first tip this process sees, against the last one the book
            # was reconciled to before it stopped. A backward walk is bounded
            # by `rescan_depth`, so a book away longer than that cannot see
            # what happened while it was away: offers quietly read as `gone`
            # and vaults as `SPENT_UNKNOWN`, which looks like chain trouble
            # rather than "run me once with --rescan-from". So it is said, in
            # /healthz and on the page, until that is done.
            if self.rescan_needed_from is None and not self._absence_checked:
                self._absence_checked = True
                last = int(self.book.meta.get("last_height") or 0)
                if last and self.height - last > self.rescan_depth:
                    self.rescan_needed_from = last
                    print(f"pignusd: the book was last reconciled at height "
                          f"{last} and the chain is at {self.height}, which "
                          f"is further back than the {self.rescan_depth} "
                          f"blocks a poll can look. Run once with "
                          f"--rescan-from {last} to read what happened in "
                          f"between; until then some offers and vaults may "
                          f"read as gone or unknown.", file=sys.stderr)
            self.book.meta["last_height"] = int(self.height)
            # One write for the whole reconciliation. Confirmations move on
            # every block for every live loan and every open offer, and each of
            # those used to rewrite the entire book under the lock the request
            # handlers take.
            with self.book.batch():
                self._reconcile(w)
        except Exception as e:                      # noqa: BLE001
            self.node_error = f"node: {type(e).__name__}: {e}"
        finally:
            self._chain_lock.release()

    def _track_new(self, w):
        """Hand the watcher every record the book holds and it does not."""
        for rec in self.book.list_loans():
            if rec["loan_id"] in w.vaults:
                continue
            try:
                terms = LoanTerms.from_json(rec["terms"])
            except Exception as e:              # noqa: BLE001
                print(f"loan {rec.get('loan_id','?')[:12]}: unparseable "
                      f"terms, skipping: {e}", file=sys.stderr)
                continue
            # Hand back what the book already knew. A vault re-tracked from
            # nothing looks like a funding that never buried, which is what
            # a reorged-away funding looks like too.
            w.track(rec["loan_id"], terms, rec["txid"], rec["vout"],
                    single_leaf=bool(rec.get("single_leaf")),
                    state=rec.get("state") or "UNCONFIRMED",
                    confirmations=rec.get("confirmations") or 0,
                    spent_by=rec.get("spent_by") or "",
                    spent_height=rec.get("spent_height") or 0,
                    note=rec.get("note") or "",
                    funding_height=rec.get("funding_height") or 0,
                    funding_block=rec.get("funding_block") or "")
        for off in self.book.list_offers(status="all"):
            # Ghosts too: a funding undone by a Bitcoin reorg can come back,
            # and only a watched offer can be seen to come back. Everything
            # else -- taken, withdrawn, gone -- is finished.
            if off.get("status", "open") not in ("open", "ghost", "delisted"):
                continue
            if off.get("kind") != "funded" or off["offer_id"] in w.offers:
                continue
            try:
                txid, _, vout = str(off["outpoint"]).rpartition(":")
                w.track_offer(off["offer_id"],
                              LoanTerms.from_json(off["terms"]), txid,
                              int(vout), int(off["principal"]),
                              int(off["collateral"]),
                              int(off["expiry_locktime"]),
                              confirmations=off.get("confirmations") or 0,
                              status=off.get("status") or "open",
                              value=int(off.get("funded_value") or 0))
            except Exception as e:              # noqa: BLE001
                print(f"offer {off.get('offer_id','?')[:12]}: cannot track, "
                      f"skipping: {e}", file=sys.stderr)

    def refresh_one(self, loan_id=None, offer_id=None, wait=2):
        """Reconcile ONE record, now, for the request about to serve it.

        A publish or a registration ran the whole reconciliation in its own
        thread and answered after every record's question -- fifteen seconds
        on a full book, longer than a reverse proxy waits. The record about
        to be served back is the only one the caller reads confirmations
        off; the rest are the poll's, a few seconds later.
        """
        if not self.watcher:
            return
        w = self.watcher
        if not self._chain_lock.acquire(timeout=wait):
            return          # a reconcile is in flight; it will cover this too
        try:
            tip = self.node.getblockcount()
            self.height = tip
            self._track_new(w)
            with self.book.batch():
                w.refresh_one(tip, loan_id=loan_id, offer_id=offer_id)
                if loan_id and loan_id in w.vaults:
                    self._reconcile_loan(w.vaults[loan_id])
                if offer_id and offer_id in w.offers:
                    self._reconcile_offer(w.offers[offer_id])
            self.node_error = None
        except Exception as e:                      # noqa: BLE001
            self.node_error = f"node: {type(e).__name__}: {e}"
        finally:
            self._chain_lock.release()

    def _reconcile(self, w):
        self._track_new(w)
        try:
            w.poll()
            self.node_error = None
        except Exception as e:                  # noqa: BLE001
            self.node_error = f"node: {type(e).__name__}: {e}"
        # Reconcile everything tracked, not only what this poll reported as
        # changed. A poll that raises part way has already moved vaults the
        # watcher will never report again, and the book would keep calling
        # a closed loan LIVE for as long as it ran.
        for v in list(w.vaults.values()):
            self._reconcile_loan(v)
        # Offer events are applied IN ORDER, and a failed one is retried on
        # the next poll rather than dropped: the event is what registers the
        # loan an offer opened, and the watcher has already moved on.
        #
        # But only a failure that could pass LATER is worth retrying. An event
        # this daemon can never apply -- terms it cannot parse, a take whose
        # shape it does not understand -- would otherwise sit at the head of
        # the queue for ever, and every later event, including every loan every
        # other offer opens, would queue up behind it without bound. So a
        # verdict about the event itself is final and the event is dropped,
        # loudly; only a book or node failure is retried.
        while w.events:
            ev = w.events[0]
            try:
                self._apply_offer_event(ev)
            except (ValueError, TypeError, KeyError, AssertionError) as e:
                # A judgement about this event, which will not change.
                self.event_errors.append(
                    f"offer {str(ev.get('offer_id', '?'))[:12]}: dropped the "
                    f"'{ev.get('kind')}' event: {type(e).__name__}: {e}")
                print(self.event_errors[-1], file=sys.stderr)
                w.events.pop(0)
                continue
            except Exception as e:              # noqa: BLE001
                print(f"offer {str(ev.get('offer_id', '?'))[:12]}: cannot "
                      f"apply the '{ev.get('kind')}' event yet, retrying "
                      f"next poll: {type(e).__name__}: {e}", file=sys.stderr)
                break
            w.events.pop(0)
        # A queue that keeps growing is a daemon falling behind its own chain.
        # Say so where an operator is already looking.
        self.event_backlog = len(w.events)
        for o in list(w.offers.values()):
            self._reconcile_offer(o)

    def _reconcile_loan(self, v):
        """Write what the watcher believes about a vault, key by key, so a
        record that missed one update is repaired by the next poll."""
        rec = self.book.loans.get(v.loan_id)
        if rec is None:
            return
        # funding_height and funding_block go with the rest. They are the pair
        # that tells a Bitcoin-driven reorg from a spend out of the backward
        # walk's reach; a restart that lost them re-tracks the vault as one that
        # never buried, and the reorg the book exists to report reads instead as
        # "spent by something we cannot see".
        want = {"state": v.state.value, "confirmations": v.confirmations,
                "spent_by": v.spent_by, "spent_height": v.spent_height,
                "note": v.note, "funding_height": v.funding_height,
                "funding_block": v.funding_block}
        diff = {k: val for k, val in want.items() if rec.get(k) != val}
        if not diff:
            return
        try:
            self.book.update_loan(v.loan_id, **diff)
        except Exception as e:                      # noqa: BLE001
            print(f"loan {v.loan_id[:12]}: cannot record "
                  f"{', '.join(sorted(diff))}: {type(e).__name__}: {e}",
                  file=sys.stderr)

    def _reconcile_offer(self, o):
        rec = self.book.offers.get(o.offer_id)
        if rec is None:
            return
        if o.status == "open":
            # The outpoint too: a partial take moves the coin to the remainder,
            # and a book left pointing at the spent one advertises an offer
            # nobody can take.
            want = {"outpoint": f"{o.txid}:{o.vout}",
                    "funded_value": str(o.value),
                    "confirmations": o.confirmations}
        elif rec.get("status", "open") == "open":
            # The coin is gone and the event that said so was lost. An offer
            # nobody can take must not stay on the page as one anybody can.
            want = {"status": o.status, "funded_value": "0"}
        else:
            return
        diff = {k: val for k, val in want.items() if rec.get(k) != val}
        if not diff:
            return
        try:
            self.book.update_offer(o.offer_id, **diff)
        except Exception as e:                      # noqa: BLE001
            print(f"offer {o.offer_id[:12]}: cannot record "
                  f"{', '.join(sorted(diff))}: {type(e).__name__}: {e}",
                  file=sys.stderr)

    def _apply_offer_event(self, ev):
        """What the chain did to an offer, written into the book."""
        off = self.book.offers.get(ev["offer_id"])
        if off is None:
            return
        kind = ev["kind"]
        if kind == "taken":
            base = json.loads(off["terms"])
            prog = ev["borrower_prog"]
            # Through the READER. It turns the oracle list back into a tuple
            # and coerces amounts a browser sent as decimal strings; the
            # constructor does neither, and the terms this writes into the book
            # are the ones every later reader rebuilds an address from.
            terms = LoanTerms.from_json(json.dumps(
                {**base, "borrower_prog": prog, "borrower_x": prog}))
            v = ev["vault"]
            rec = self.book.put_loan(terms.to_json(), v["txid"], v["vout"],
                                     single_leaf=True, offer_id=off["offer_id"],
                                     discovered="chain")
            if rec["loan_id"] not in self.watcher.vaults:
                self.watcher.track(rec["loan_id"], terms, v["txid"], v["vout"],
                                   single_leaf=True)
            print(f"offer {off['offer_id'][:12]} taken in {ev['txid'][:12]}: "
                  f"loan {rec['loan_id'][:12]} at {v['txid'][:12]}:{v['vout']}",
                  file=sys.stderr)
            rem = ev.get("remainder")
            # Counted once per TRANSACTION. A take seen in the mempool, then
            # in a block, then again after a reorg re-mines it is one take
            # reported three times, and a counter that grew on each report
            # drifted upward for the life of the offer.
            seen = list(off.get("taken_txids") or [])
            count = int(off.get("taken", 0))
            if ev["txid"] not in seen:
                seen.append(ev["txid"])
                count += 1
            if rem:
                self.book.update_offer(off["offer_id"],
                                       outpoint=f"{rem['txid']}:{rem['vout']}",
                                       funded_value=str(rem["value"]),
                                       confirmations=0,
                                       taken=count, taken_txids=seen)
            else:
                self.book.update_offer(off["offer_id"], status="taken",
                                       funded_value="0",
                                       taken=count, taken_txids=seen,
                                       spent_by=ev["txid"])
        elif kind == "withdrawn":
            self.book.update_offer(off["offer_id"], status="withdrawn",
                                   spent_by=ev["txid"])
        else:   # "gone" or "ghost"
            self.book.update_offer(off["offer_id"], status=kind,
                                   spent_by=ev.get("txid", ""),
                                   note=ev.get("note", ""))

    def prune_book(self):
        """Drop the records that have been finished for long enough.

        This decides nothing about a loan. The chain is the record and a pruned
        loan can be read back off it; what goes is an index entry, and what it
        stops costing is a full render of the whole book on every read.
        """
        if self.prune_after <= 0:
            return
        # The watcher is unwatched here too, and a POST can be reconciling on
        # another thread: hold the lock refresh_chain holds rather than take
        # records out from under a poll that is iterating them. A prune that
        # cannot get it waits for the next one; there is no hurry.
        if not self._chain_lock.acquire(timeout=5):
            return
        try:
            dropped = self.book.prune(self.prune_after)
            if not any(dropped.values()):
                return
            self._stop_watching(dropped)
        except Exception as e:                      # noqa: BLE001
            # A book that cannot be pruned is still a book worth serving.
            print(f"prune: {type(e).__name__}: {e}", file=sys.stderr)
            return
        finally:
            self._chain_lock.release()
        names = {"offers": "offer", "loans": "loan", "btc_takes": "BTC take"}
        counts = ", ".join(
            f"{len(v)} {names.get(k, k)}{'' if len(v) == 1 else 's'}"
            for k, v in dropped.items() if v)
        print(f"pignusd: pruned {counts} finished more than "
              f"{self.prune_after}s ago", file=sys.stderr)

    def _stop_watching(self, dropped):
        """Stop watching what the book has just forgotten.

        The watcher indexes what it tracks by the outpoint it rests in, and
        looks the record up through that index for every input of every block
        it scans. So the index entry has to go with the record: one left
        pointing at a vault that is no longer tracked raises inside the block
        scan, which is the one place this daemon must not fall over.
        """
        loans = set(dropped.get("loans") or ())
        offers = set(dropped.get("offers") or ())
        for rid in loans | offers:
            UNRENDERABLE.pop(rid, None)
        w = self.watcher
        if w is None:
            return
        for lid in loans:
            w.vaults.pop(lid, None)
            # The vault script cache is keyed by the loan's terms id, which
            # is its loan id; evicting here keeps it from ever reaching the
            # size at which it is cleared wholesale and every list pays the
            # covenant build again.
            for single in (True, False):
                self._spk_cache.pop((lid, single), None)
        for oid in offers:
            w.offers.pop(oid, None)
        for attr, gone in (("_by_outpoint", loans), ("_by_offer", offers)):
            index = getattr(w, attr, None)
            if not isinstance(index, dict):
                continue        # indexed some other way; leave it whole
            for key in [k for k, rid in index.items() if rid in gone]:
                del index[key]

    def expire_btc_offers(self):
        """Close cross-chain offers whose own deadlines have gone by.

        Nothing else ever ended one. They carry no coin, so a lender who walks
        away leaves theirs listed for ever -- and because publishing costs
        nothing but a self-signature, MAX_BTC_OFFERS was a one-way ceiling:
        one script could fill it with well-formed offers under keys it made up
        and Tier B would be closed to everyone, permanently.

        They need no expiry field. Every one of them names four deadlines, and
        an offer whose deadlines no longer leave both sides the margins
        `timelocks_sane` insists on is one no responder will answer and no
        borrower should take: the same rule that would refuse the take, applied
        one step earlier. Marked rather than deleted, so pruning collects it on
        the same schedule as every other dead record.
        """
        from pignus import btc_collateral as BC
        btc_h = self.btc_height()
        if btc_h is None or not self.height:
            return 0                    # no parent chain: judge nothing
        n = 0
        # One write for the sweep, not one per expired offer: see
        # `Book.stamp_at_risk`.
        with self.book.batch():
          for o in self.book.list_btc_offers(status="open"):
            # An offer is the LENDER's half of a loan. The borrower's key and
            # their origination commitment arrive on the take, so an offer
            # carries neither -- and `loan_from_dict` cannot build a BtcLoan
            # without `borrower_x` at all. Stand in for both.
            #
            # Neither stand-in is used as a value: `timelocks_sane` never reads
            # `borrower_x`, and reads `h_w` only for presence, to know the
            # origination is the abortable shape. What gets judged is the
            # offer's own four deadlines, which is the whole point. Standing in
            # for only one of them is worse than neither -- an offer with
            # `abort_after` and `d_refund` but no `h_w` collects "an abortable
            # origination needs h_w, abort_after and d_refund together", and
            # every honest offer on the board would be expired on the first
            # poll.
            loan = dict(o.get("loan") or {})
            loan["borrower_x"] = "00" * 32
            loan["h_w"] = "00" * 32
            try:
                problems = BC.timelocks_sane(BC.loan_from_dict(loan), btc_h,
                                             int(self.height))
            except Exception as e:                      # noqa: BLE001
                # LOUD. An offer this cannot read is one no borrower can take,
                # but it is not this sweep's to end. Passing it over in silence
                # is how this loop came to do nothing whatever, for every offer
                # on the book, while reporting success.
                print(f"pignusd: cross-chain offer {o.get('btc_offer_id')} "
                      f"cannot be judged: {type(e).__name__}: {e}",
                      file=sys.stderr)
                continue
            if not problems:
                continue
            self.book.update_btc_offer(o["btc_offer_id"], status="expired")
            print(f"pignusd: cross-chain offer {o['btc_offer_id']} has expired: "
                  + "; ".join(problems)[:200], file=sys.stderr)
            n += 1
        return n

    def loop(self, interval):
        first = True
        while True:
            started = time.time()
            # EACH STEP ITS OWN GUARD. One try around all six was enough to
            # keep the thread alive and not enough to keep the book working: an
            # oracle -- a third party by design -- serving a reply this code
            # did not anticipate would raise in `refresh_prices` and every
            # later step would be skipped for ever after. The chain would never
            # be reconciled again, so a repaid loan reads LIVE indefinitely
            # while /healthz reports only a poll error. A failing source must
            # cost its own step and nothing else.
            errors = {}
            for name, step in (("registry",
                                lambda: self.refresh_registry(force=first)),
                               ("prices", self.refresh_prices),
                               ("fees", self.refresh_fees),
                               ("chain", self.refresh_chain),
                               ("offer-expiry", self.expire_btc_offers),
                               ("prune", self.prune_book)):
                try:
                    step()
                except Exception as e:                  # noqa: BLE001
                    errors[name] = f"{type(e).__name__}: {e}"
                    print(f"pignusd: the {name} step of the poll failed: "
                          f"{errors[name]}", file=sys.stderr)
                self.view_epoch += 1
            self.poll_error = ("; ".join(f"{k}: {v}" for k, v in errors.items())
                               if errors else None)
            self.last_poll = int(time.time())
            if first:
                # The first pass re-tracks every loan and walks back for spends
                # it missed, which on a long-idle book takes a while. It runs
                # here rather than before the socket is bound, so the page and
                # /healthz answer -- and say "initial sync" -- throughout.
                print(f"pignusd: first sync done in "
                      f"{time.time() - started:.1f}s at height {self.height}",
                      file=sys.stderr)
                first = False
            time.sleep(max(2.0, interval - (time.time() - started)))

    # ----------------------------------------------------------- verification

    def verify_offer(self, body):
        """Refuse to publish an offer that is not actually funded where it says.

        The site checks an offer for itself before composing anything against
        it, so this is not what stands between a borrower and a loss. It is what
        stands between a borrower and a screen full of plausible fiction: an
        unchecked book can be filled with offers that do not exist, and every
        one of them wastes somebody's attention.
        """
        # A STRING, as the reference says and as every reader assumes: the
        # page does JSON.parse on it, and the watcher's event applier does
        # json.loads. An object stored verbatim rendered as "[object Object]"
        # in the browser, which took the whole offers panel down for every
        # visitor, and raised a TypeError in the applier that was read as a
        # permanent verdict -- so a take of that offer was never registered.
        if not isinstance(body.get("terms"), str):
            raise ValueError("terms must be the loan terms as a JSON STRING, "
                             "not an object")
        terms = LoanTerms.from_json(body["terms"])
        # Heights, only. An absolute locktime at or above LOCKTIME_THRESHOLD is
        # a Unix time to a node, and every comparison on the offer path -- the
        # page's "matured", the CLI's take and withdraw gates, this book's own
        # `expired` -- compares against a block height. A maturity written as
        # a time from the past passes all of them as "thousands of years away"
        # while the loan's DEFAULT leaf is already open the block it exists.
        from pignus import LOCKTIME_THRESHOLD
        for name, val in (("maturity", terms.maturity),
                          ("recover_after", terms.recover_after),
                          ("expiry_locktime",
                           body.get("expiry_locktime") or terms.maturity)):
            if int(val) >= LOCKTIME_THRESHOLD:
                raise ValueError(
                    f"this offer's {name} is {int(val)}, which a node reads as "
                    f"a Unix time rather than a block height. Offers are "
                    f"written in heights, because everything that checks a "
                    f"deadline on this path compares it to one.")
        if body.get("kind") != "funded":
            raise ValueError('an offer must be funded: set kind to "funded" '
                             "and name the outpoint its principal rests in")
        outpoint = str(body.get("outpoint") or "")
        if ":" not in outpoint:
            raise ValueError("a funded offer must name outpoint 'txid:vout'")
        txid, _, vout = outpoint.rpartition(":")
        if not self.node:
            raise NotOnChain(
                "this book has no node, so it cannot check that a funded offer "
                "is real, and it will not publish one it cannot check")
        # An offer has TWO principals and they must be the same number. The
        # one in the terms is decorative -- it enters no address -- while this
        # one is baked into the TAKE leaf as `taken == principal` and is the
        # only one the covenant enforces. Letting them differ publishes an
        # offer that says one thing on the page and pays another on the chain,
        # and the page reads whichever field happens to be nearest.
        principal = int(body.get("principal") or terms.principal)
        collateral = int(body.get("collateral") or terms.collateral_amount)
        if principal != int(terms.principal):
            raise ValueError(
                f"this offer's terms say a principal of {terms.principal} and "
                f"its listing says {principal}. Only the listing is enforced "
                f"by the covenant, so the two differing is an offer that "
                f"advertises one loan and makes another.")
        if collateral != int(terms.collateral_amount):
            raise ValueError(
                f"this offer's terms lock {terms.collateral_amount} of "
                f"collateral and its listing says {collateral}. They have to "
                f"agree: the listing is what the covenant enforces.")
        expiry = int(body.get("expiry_locktime") or terms.maturity)
        want = offer_address(terms, principal, collateral, expiry)
        got = self._outpoint_holds(txid, vout, want, "offer")
        if got["asset"] and got["asset"] != terms.debt_asset:
            raise NotOnChain(
                f"that outpoint holds {got['asset'][:16]}…, not the debt asset "
                f"{terms.debt_asset[:16]}…")
        if got["value"] < principal:
            raise NotOnChain(
                f"that outpoint holds {got['value']} atoms, less than one "
                f"principal of {principal}")
        body["principal"] = str(principal)
        body["collateral"] = str(collateral)
        body["expiry_locktime"] = expiry
        body["funded_value"] = str(got["value"])
        body["confirmations"] = got["confirmations"]

    def _outpoint_holds(self, txid, vout, want, what):
        """check_outpoint, with a confidential output turned into a sentence.

        A blinded output has no `value` to read, and the covenant needs an
        explicit amount anyway -- so this is a refusal with a reason, not the
        KeyError it used to be.
        """
        try:
            return check_outpoint(self.node, txid, vout, want, what)
        except KeyError:
            raise NotOnChain(f"that output is confidential; a {what} must hold "
                             "an explicit amount anyone can check")

    def verify_loan(self, body):
        """Refuse to track a loan whose outpoint is not the vault its terms
        compile to, or does not hold what the terms say it locks.

        Same reason as above: everything a reader is shown about a loan --
        health, LTV, what a liquidator would seize -- is computed from the
        terms, so a vault holding one atom of something else would put a whole
        fictional position on every screen and in every liquidator's feed.
        """
        terms = LoanTerms.from_json(body["terms"])
        if not self.node:
            raise NotOnChain(
                "this book has no node, so it cannot check that a loan is real")
        single = bool(body.get("single_leaf"))
        want = (offer_vault_address(terms) if single
                else terms.script_pubkey())
        got = self._outpoint_holds(body["txid"], body["vout"], want, "loan")
        if got["asset"] and got["asset"] != terms.collateral_asset:
            raise NotOnChain(
                f"that vault holds {got['asset'][:16]}…, not the collateral "
                f"asset {terms.collateral_asset[:16]}…")
        # Exactly, not at least: the covenant returns what is locked, so terms
        # that overstate or understate the coin misprice every figure derived
        # from them.
        if got["value"] != terms.collateral_amount:
            raise NotOnChain(
                f"that vault holds {got['value']} atoms, but the terms lock "
                f"{terms.collateral_amount}")
        return got

    # -------------------------------------------------------------- rendering

    def market_rows(self):
        rows = []
        fresh = self.fresh_prices()
        for m in self.markets:
            att = self.attestations.get(m)
            n_oracles = sum(1 for ox in self.oracle_keys
                            if self._att_key(ox, m) in self.att_by_oracle)
            c_sym, _, d_sym = m.partition("/")
            c_asset, c_tick = self.resolve(c_sym)
            d_asset, d_tick = self.resolve(d_sym)
            c_prec = (self.assets.get(c_asset) or {}).get("precision", 8)
            d_prec = (self.assets.get(d_asset) or {}).get("precision", 8)
            # What the PRIMARY oracle scaled the price by, against what the
            # registry says the two assets are denominated in. They are the same
            # two numbers, arrived at independently; when they differ the price
            # is out by a factor of ten per decimal and nothing else can see it.
            said = self.oracle_precisions.get(self._att_key(self.oracle_x, m))
            mismatch = bool(said and c_asset and d_asset
                            and said != (c_prec, d_prec))
            rows.append({
                "market": m, "feed_id": feed_id(m).hex(),
                "collateral_ticker": c_tick, "debt_ticker": d_tick,
                "collateral_asset": c_asset, "debt_asset": d_asset,
                "collateral_precision": c_prec,
                "debt_precision": d_prec,
                "oracle_precisions": list(said) if said else None,
                "precision_mismatch": mismatch,
                "cross_chain": c_sym.strip().upper() in NATIVE,
                # A cross-chain market is priced and watched here, but it is
                # not lendable through the covenant offers: its collateral is
                # on the parent chain, where none of the covenant runs. Saying
                # otherwise puts native Bitcoin in the lend form and in the
                # sentence about what the network enforces.
                "lendable": bool(c_asset and d_asset and m in fresh
                                 and not mismatch
                                 and c_sym.strip().upper() not in NATIVE),
                "stale": bool(att) and m not in fresh,
                "debt_is_reference": d_tick.upper() == self.reference_ticker.upper(),
                "price": att.price if att else None,
                "price_scale": att.price_scale if att else None,
                # Quoted per WHOLE unit of each asset, so the precisions the
                # price was scaled by have to come back out of it: with the
                # 8/8 default a market whose debt asset has two decimals reads
                # a million times too small.
                "unit_price": (O.unquote_price(att.price, c_prec, d_prec,
                                               price_scale=att.price_scale)
                               if att else None),
                "timestamp": att.timestamp if att else None,
                "age_seconds": O.age_of(att) if att else None,
                "oracles_available": n_oracles,
            })
        return rows

    def _tickers(self, out, t):
        c = self.assets.get(t.collateral_asset) or {}
        d = self.assets.get(t.debt_asset) or {}
        out["collateral_ticker"] = c.get("ticker") or t.collateral_asset[:8] + "…"
        out["debt_ticker"] = d.get("ticker") or t.debt_asset[:8] + "…"
        out["collateral_precision"] = c.get("precision", 8)
        out["debt_precision"] = d.get("precision", 8)

    def offer_view(self, rec):
        t = LoanTerms.from_json(rec["terms"])
        out = dict(rec)
        out.pop("manage_hash", None)             # never serve the manage secret
        self._tickers(out, t)
        out["collateral_asset"] = t.collateral_asset
        out["debt_asset"] = t.debt_asset
        out["debt"] = str(t.debt)
        out["strike"] = str(t.strike)
        out["maturity"] = t.maturity
        out["lender_prog"] = t.payout_programs[0]
        out["lender_ver"] = t.lender_ver
        out["oracle_x"] = t.oracle_x
        out["oracle_compromised"] = self._oracle_compromised(t)
        price = self.price_for(t)
        if rec.get("liquidatable_since"):
            out["liquidatable_since"] = int(rec["liquidatable_since"])
        if price is not None:
            out["price"] = price
            out["open_ltv"] = round(t.principal / max(1, t.collateral_value(price)), 4)
        if self.height is not None:
            out["expired"] = self.height >= int(rec.get("expiry_locktime") or 0)
            out["height"] = self.height
        try:
            out["lots_left"] = int(rec.get("funded_value") or 0) // int(rec["principal"])
        except (ValueError, ZeroDivisionError, KeyError):
            out["lots_left"] = None
        return out

    def vault_spk(self, terms, terms_id, single):
        key = (terms_id, bool(single))
        spk = self._spk_cache.get(key)
        if spk is None:
            spk = (offer_vault_address(terms) if single
                   else terms.script_pubkey()).hex()
            if len(self._spk_cache) > MAX_LOANS:
                self._spk_cache.clear()
            self._spk_cache[key] = spk
        return spk

    def loan_view(self, rec):
        t = LoanTerms.from_json(rec["terms"])
        # This loan's OWN oracles, not the book's primary: see price_for.
        price = self.price_for(t)
        out = dict(rec)
        self._tickers(out, t)
        out["collateral_asset"] = t.collateral_asset
        out["debt_asset"] = t.debt_asset
        out["principal"] = str(t.principal)
        out["debt"] = str(t.debt)
        out["collateral_amount"] = str(t.collateral_amount)
        out["strike"] = str(t.strike)
        out["price_scale"] = t.price_scale
        out["maturity"] = t.maturity
        out["recover_after"] = t.recover_after
        out["lender_prog"] = t.payout_programs[0]
        out["borrower_prog"] = t.payout_programs[1]
        out["lender_ver"] = t.lender_ver
        out["borrower_ver"] = t.borrower_ver
        out["oracle"] = f"{t.threshold}-of-{len(t.oracle_keys)}"
        out["oracle_x"] = t.oracle_x
        out["oracle_keys"] = list(t.oracle_keys)
        out["oracle_compromised"] = self._oracle_compromised(t)
        # A loan originated through a funded offer lives in the SINGLE-LEAF
        # vault; the browser needs to know which, because the two have
        # different addresses and different witnesses.
        single = bool(rec.get("single_leaf"))
        out["single_leaf"] = single
        out["vault_address"] = self.vault_spk(
            t, rec.get("terms_id") or t.loan_id(), single)
        if self.height is not None:
            out["height"] = self.height
            # In the LOCKTIME's own units: a deadline at or above 500,000,000
            # is a Unix time to a node, not a height, and comparing one to a
            # block height puts it thousands of years away. The chain's median
            # time is what consensus compares a time locktime to.
            mtp = self.median_time()
            out["past_maturity"] = _locktime_open(t.maturity, self.height, mtp)
            out["recover_open"] = _locktime_open(t.recover_after, self.height,
                                                 mtp)
        # How deep the CLOSE is, which is a different question from how deep the
        # funding was. A repayment or a liquidation one block old can still be
        # reorged out, and showing it as simply settled tells a borrower their
        # collateral is back when it may not be. Zero means either not closed at
        # all or closed only in the mempool.
        spent_height = int(rec.get("spent_height") or 0)
        out["spent_height"] = spent_height
        # The depth this book calls a funding buried at, beside the loan it was
        # applied to: a reader counting confirmations should not have to fetch
        # another endpoint to learn what number they are counting towards.
        out["min_depth"] = self.min_depth
        out["closed_confirmations"] = (self.height - spent_height + 1
                                       if self.height and spent_height else 0)
        if price is not None:
            out["price"] = price
            # Terms with a zero strike, or collateral worth nothing at this
            # price, make these infinite, and infinity is not JSON: the page
            # would lose the whole list rather than one number.
            if t.strike:
                out["health"] = round(t.health(price), 4)
            if t.collateral_value(price):
                out["ltv"] = round(t.ltv(price), 4)
            out["liquidatable"] = t.is_liquidatable(price)
            out["seizure_if_liquidated"] = str(t.seizure_at(price))
            out["surplus_if_liquidated"] = str(t.surplus_at(price))
        return out

    # ----------------------------------------------------- explaining a close

    def spending_tx(self, v):
        """The transaction that closed a vault, or None if it is out of reach.

        Never `getrawtransaction` by txid: the committee nodes run without a
        transaction index, so a mined transaction cannot be looked up by id at
        all and the call fails in a way that reads like the transaction not
        existing. A mined spend is read out of the block the watcher recorded
        it in; a spend still in the mempool is the one case the node can answer
        by id, because it is holding it.
        """
        if not self.node or not v.spent_by:
            return None
        try:
            if v.spent_height and int(v.spent_height) > 0:
                blk = self.node.getblock(
                    self.node.getblockhash(int(v.spent_height)), 2)
                for tx in blk.get("tx", []):
                    if tx.get("txid") == v.spent_by:
                        return tx
                return None
            if v.spent_by not in (self.node.getrawmempool() or []):
                return None
            return self.node.getrawtransaction(v.spent_by, True)
        except Exception as e:                      # noqa: BLE001
            print(f"loan {v.loan_id[:12]}: cannot read the closing "
                  f"transaction: {type(e).__name__}: {e}", file=sys.stderr)
            return None

    def explain_exit(self, loan_id):
        """Everything a closed loan's account can be rebuilt from, or None.

        A borrower who was liquidated is owed the price it happened at and the
        signature behind it, not a state tag -- so this returns the oracle
        evidence out of the witness, checked against the key the vault itself
        bakes in, beside what the terms say that price buys and what the
        transaction actually paid.

        CACHED, because the answer cannot change: a mined close is a fact about
        a block. Without a cache every unauthenticated GET pulled a whole
        verbose block from the node, which is by far the most expensive thing
        this daemon can be asked to do and is one request away from anybody.
        """
        v = self.watcher.vaults.get(loan_id) if self.watcher else None
        if v is None or not v.spent_by:
            return None
        # Keyed by the spend AND its height, so a close undone by a reorg and
        # replaced by a different one is a different key rather than a stale
        # answer.
        key = (loan_id, v.spent_by, int(v.spent_height or 0))
        hit = self._exit_cache.get(key)
        if hit is not None:
            return hit
        out = self._explain_exit_uncached(v, loan_id)
        # A close still in the mempool can change; only a mined one is a fact.
        if out is not None and int(v.spent_height or 0) > 0:
            if len(self._exit_cache) >= 512:
                self._exit_cache.clear()
            self._exit_cache[key] = out
        return out

    def _explain_exit_uncached(self, v, loan_id):
        tx = self.spending_tx(v)
        if tx is None:
            return {"loan_id": loan_id, "spent_by": v.spent_by,
                    "height": int(v.spent_height or 0), "exit": v.state.value,
                    "problems": ["the transaction that closed this loan is not "
                                 "in reach: it is neither in the block this "
                                 "book recorded nor in the mempool"]}
        want = (v.txid, int(v.vout))
        k = next((i for i, vin in enumerate(tx.get("vin", []))
                  if (vin.get("txid"), vin.get("vout")) == want), None)
        if k is None:
            return {"loan_id": loan_id, "spent_by": v.spent_by,
                    "height": int(v.spent_height or 0), "exit": v.state.value,
                    "problems": ["that transaction does not spend this vault; "
                                 "the book is pointing at the wrong one"]}
        return self.watcher.explain_exit(v, tx, k)

    # ------------------------------------------- registrations still to bury

    def pending_registrations(self, key):
        """How many loans this client has registered that have not buried.

        Counted by re-reading the book, and swept while counting: a loan that
        went LIVE or closed stops being charged to whoever registered it, so a
        borrower who opens loans one after another is never refused.
        """
        if key is None:
            return 0
        with self._lock:
            ids = self._pending_by_client.get(key) or []
            still = [i for i in ids
                     if (self.book.loans.get(i) or {}).get("state")
                     in PENDING_NAMES]
            if still:
                self._pending_by_client[key] = still
            else:
                self._pending_by_client.pop(key, None)
            return len(still)

    def note_registration(self, key, loan_id):
        if key is None:
            return
        with self._lock:
            ids = self._pending_by_client.setdefault(key, [])
            if loan_id not in ids:
                ids.append(loan_id)

    # ------------------------------------------------------- BTC collateral

    def btc_vault_outpoint(self, loan, prevault_txid, prevault_vout):
        """Where the collateral will be once the loan starts.

        The lender's release is signed against the VAULT, not the pre-vault, and
        the vault's outpoint is fixed the moment the borrower signs the one
        transaction that can create it -- which is why an id is known here
        before anything is broadcast."""
        from pignus import btc_collateral as BC
        return BC.upgrade_tx(loan, prevault_txid, int(prevault_vout)).txid(), 0

    def btc_reclaim_sighash(self, loan, vault_txid, vault_vout,
                            reclaim_dest_hex, fee):
        """Recompute the reclaim sighash the lender's RELEASE signature must
        commit to, so the relay never passes on a request to sign something
        other than the loan it claims to be."""
        from pignus import btc_collateral as BC
        tx = BC.reclaim_tx(loan, vault_txid, int(vault_vout),
                           bytes.fromhex(reclaim_dest_hex), int(fee))
        return BC.sighash_for(loan, tx, "reclaim").hex()

    def btc_check_take(self, offer, body):
        """Everything the relay can check about a take REQUEST.

        None of it protects the relay, which has nothing to lose. It protects
        the two parties from each other and from the relay: the borrower's
        payout is baked into the addresses, and the collateral's own address is
        rebuilt here rather than accepted.

        The vault, and the release that commits to it, cannot be checked yet:
        both depend on the hash of a secret the lender has not drawn. That is
        the next step of the handshake, not this one.
        """
        from pignus import btc_collateral as BC
        from pignus import btc_relay as R
        loan_d = dict(offer["loan"])
        loan_d["borrower_x"] = str(body["borrower_x"]).lower()
        loan_d["h_w"] = str(body["h_w"]).lower()
        loan_d["borrower_prog"] = R.check_program(body["borrower_prog"],
                                                  body.get("borrower_ver", 0))
        loan_d["borrower_ver"] = int(body.get("borrower_ver", 0))
        loan = BC.loan_from_dict(loan_d)

        if len(loan.borrower_x) != 64 or len(loan.h_w) != 64:
            raise ValueError("a take needs the borrower's key and their "
                             "origination commitment, both 32 bytes")
        prevault_txid = str(body["prevault_txid"]).lower()
        prevault_vout = int(body["prevault_vout"])
        if len(prevault_txid) != 64:
            raise ValueError("that is not a transaction id")
        # Both of these are decided again, atomically, by `put_btc_take`. They
        # are here so a caller is told which of their two fields is the problem
        # before any work is done on the rest of the request.
        clash = self.book.btc_take_by_funding(prevault_txid, prevault_vout)
        if clash is not None:
            raise ValueError("that outpoint already funds take "
                             f"{clash['take_id']}; one coin funds one loan")
        # And one secret opens one loan. Two takes committing to the same
        # secret share a pre-vault address and a disbursement address, and the
        # moment one of them publishes it the other's collateral can be moved
        # into a vault nobody paid for. Answered from the book's commitment
        # ledger rather than by scanning the takes, because a take is pruned
        # and a secret must never become reusable.
        who = self.book.btc_hash_in_use(loan.h_w, body.get("take_id"))
        if who is not None:
            raise ValueError(
                "that origination commitment is already in use by take "
                f"{who}. Draw a fresh one: two loans sharing a secret are one "
                "loan either party can settle twice")
        if int(body.get("prevault_value", 0)) != loan.prevault_value():
            raise ValueError("the funding named does not hold the collateral "
                             "plus the fee this loan needs")
        dest = str(body["reclaim_dest"]).lower()
        if len(dest) not in (44, 68) or any(c not in "0123456789abcdef"
                                            for c in dest):
            raise ValueError("that is not a Bitcoin output script")
        fee = int(body.get("reclaim_fee", 3000))
        # The release the lender is about to sign commits to a transaction
        # paying btc_amount - fee. A fee at or above the collateral makes that
        # output negative or dust, so the release verifies and the reclaim can
        # never be broadcast -- which a borrower finds out only when they try
        # to leave.
        if fee < 1 or int(loan.btc_amount) - fee < BTC_DUST:
            raise ValueError(
                f"a reclaim fee of {fee} satoshis leaves nothing spendable of "
                f"{loan.btc_amount}: the release would be one the borrower "
                "could never use")
        # And the same cap the page applies, so this relay never records a fee
        # its own page would refuse -- which would strand a take at the moment
        # the borrower is asked to commit Bitcoin, with nothing said about why.
        # The fee is not covered by the lender's offer signature, so this is
        # the only place it is bounded before a borrower sees it.
        cap = reclaim_fee_cap(int(loan.btc_amount))
        if fee > cap:
            raise ValueError(
                f"a reclaim fee of {fee} satoshis on {loan.btc_amount} of "
                f"collateral is more than the {cap} this book will record. "
                f"The reclaim is one transaction of about {RECLAIM_VSIZE} "
                f"vbytes")
        # And a FLOOR, priced from the parent chain. A fee too small to relay
        # makes the release the lender is about to sign worthless: it covers
        # exactly one transaction, which nobody can bump, so the collateral
        # would sit in the vault until the lender swept it at the timeout.
        floor = reclaim_fee_floor(self.btc_feerate())
        if fee < floor:
            raise ValueError(
                f"a reclaim fee of {fee} satoshis would not relay: a reclaim "
                f"is about {RECLAIM_VSIZE} vbytes and this chain wants around "
                f"{floor} for one. The release is signed over exactly that "
                f"transaction and can never be bumped, so the collateral "
                f"would never come back")

        # Heights this daemon read itself. A height a client sends is a height
        # an attacker chooses, and every cross-chain margin is measured against
        # it -- a borrower passing zero would make every Bitcoin-side deadline
        # look infinitely far away.
        problems = []
        btc_height = self.btc_height()
        if self.node is not None and btc_height is not None:
            try:
                problems = BC.timelocks_sane(loan, btc_height,
                                             int(self.height or 0))
            except Exception:                           # noqa: BLE001
                problems = []
        elif btc_height is None:
            problems = ["this book cannot see the parent chain, so it cannot "
                        "check this loan's Bitcoin deadlines: check them "
                        "yourself before funding anything"]
        return loan, dest, problems

    def btc_feerate(self):
        """Satoshis per vbyte on the parent chain, or None.

        None when there is no Bitcoin node or the node will not estimate:
        unknown is never a number here, because a page that shows a made-up
        feerate lets a borrower judge an unbumpable fee against nothing.
        Cached for a poll -- it moves with the mempool, not with a request.
        """
        btc = getattr(self, "btc_node", None)
        if btc is None:
            return None
        now = time.time()
        if self._btc_feerate_at and now - self._btc_feerate_at < 60:
            return self._btc_feerate
        try:
            from pignus.btc_collateral import btc_feerate as _rate  # noqa: PLC0415
            self._btc_feerate = round(_rate(btc, fallback=None), 3)
        except Exception:                               # noqa: BLE001
            self._btc_feerate = None
        self._btc_feerate_at = now
        return self._btc_feerate

    def btc_height(self):
        """The parent chain's tip, from this book's own Bitcoin node.

        None when it has none: a book without one can still carry messages, and
        says so rather than pretending to have checked a deadline it could not.
        """
        node = getattr(self, "btc_node", None)
        if node is None:
            return None
        # Remembered for a while, like the feerate. `/healthz` and
        # `/v1/markets` both carry it and every open tab polls both, so an
        # uncached call made a hung bitcoind block every health check for its
        # whole RPC timeout, thread by thread.
        now = time.time()
        if self._btc_height_at and now - self._btc_height_at < 20:
            return self._btc_height
        try:
            self._btc_height = int(node.getblockcount())
        except Exception:                               # noqa: BLE001
            self._btc_height = None
        self._btc_height_at = now
        return self._btc_height

    def depth_at(self, height, block_hash):
        """How deep a block is, or None if the chain no longer holds it there.

        Two RPCs at most, and no walk. The hash is checked against the height
        because that is the whole of the reorg test: a height still exists
        after a reorg, holding a different block, and a depth computed from it
        would be a confident answer about a transaction that is no longer in
        the chain.
        """
        if self.node is None or height is None or not block_hash:
            return None
        try:
            if self.node.getblockhash(int(height)) != block_hash:
                return None
            tip = int(self.node.getblockcount())
        except Exception:                               # noqa: BLE001
            return None
        return max(0, tip - int(height) + 1)

    def anchor_depth_at(self, height, conf):
        """`anchor_depth_of` for a block already located, so no walk is needed."""
        if self.node is None or height is None or not conf:
            return None
        btc = getattr(self, "btc_node", None)
        if btc is None:
            return None
        try:
            block = self.node.getblock(self.node.getblockhash(int(height)), 1)
        except Exception:                               # noqa: BLE001
            return None
        anchor = block.get("anchorhash") or ""
        if not anchor or int(anchor, 16) == 0:
            return None                 # a chain without anchoring
        try:
            return int(btc.getblockheader(anchor, True).get("confirmations", -1))
        except Exception:                               # noqa: BLE001
            return None

    def anchor_depth_of(self, txid, conf, vout_hint=0):
        """How deep the PARENT chain is behind the block a transaction is in.

        None when it cannot be answered -- no Bitcoin node, an unconfirmed
        transaction, a chain without anchoring, or an anchor this node cannot
        find. A caller must treat None as "not established", never as safe:
        the whole point of the number is that Sequentia depth does not measure
        a Bitcoin reorg.
        """
        if self.node is None or conf is None or int(conf) <= 0:
            return None
        btc = getattr(self, "btc_node", None)
        if btc is None:
            return None
        try:
            height = int(self.node.getblockcount()) - int(conf) + 1
            block = self.node.getblock(self.node.getblockhash(height), 1)
        except Exception:                               # noqa: BLE001
            return None
        anchor = block.get("anchorhash") or ""
        if not anchor or int(anchor, 16) == 0:
            return None                 # a chain without anchoring
        try:
            return int(btc.getblockheader(anchor, True).get("confirmations", -1))
        except Exception:                               # noqa: BLE001
            # Not in the parent chain this node follows: reorged away, or
            # unseen. Either way it is not a depth.
            return None

    def median_time(self):
        """The chain's median time past, which is what consensus compares a
        TIME locktime to.

        None when it cannot be read. Unknown is not "open": a loan reported
        callable when it is not sends a liquidator to broadcast a transaction
        the node rejects as `non-final`, and it pays for the preparing sends
        first. Cached for a poll, because it moves once a block.
        """
        if self.node is None:
            return None
        now = time.time()
        if self._mtp_at and now - self._mtp_at < 30:
            return self._mtp
        try:
            h = self.node.getbestblockhash()
            self._mtp = int(self.node.getblockheader(h)["mediantime"])
        except Exception:                               # noqa: BLE001
            self._mtp = None
        self._mtp_at = now
        return self._mtp

    def btc_take_loan(self, take, payment_hash=None):
        """The loan a take is for, as the RELAY understands it.

        Rebuilt from the offer that was signed plus the four things a taker
        chooses, so a take record cannot quietly say something the lender never
        offered. Every party does this for itself; the relay does it so that
        what it serves is at least self-consistent.
        """
        from pignus import btc_collateral as BC
        from pignus import btc_relay as R
        offer = self.book.btc_offers.get(take.get("btc_offer_id")) or {}
        d = dict(offer.get("loan") or {})
        got = dict(take.get("loan") or {})
        for k in ("borrower_x", "h_w", "borrower_prog", "borrower_ver"):
            d[k] = got.get(k, "")
        R.check_program(d["borrower_prog"], d.get("borrower_ver", 0))
        d["payment_hash"] = (payment_hash or got.get("payment_hash") or "")
        d["adaptor_point"] = got.get("adaptor_point", "")
        return BC.loan_from_dict(d)

    def health(self):
        """Healthy means the numbers being served are current. A price kept
        from an hour ago, a poll thread that stopped, or no node at all all
        leave the book answering with figures that look live and are not, so
        each of them says so here rather than only in a log nobody reads."""
        now = time.time()
        stale = self.stale_markets()
        btc_h = self.btc_height()
        reasons = []
        if self.node is None:
            reasons.append("no node configured; chain states are frozen")
        if self.node_error:
            reasons.append(self.node_error)
        if getattr(self, "btc_node", None) is not None and btc_h is None:
            # Configured and not answering. Without it every Bitcoin-side
            # deadline goes unchecked, the expiry sweep judges nothing, and
            # the page refuses to originate -- and none of that showed here.
            reasons.append("the Bitcoin node is not answering: cross-chain "
                           "deadlines are unchecked and the page will not "
                           "originate until it does")
        if self.oracle_error:
            reasons.append(self.oracle_error)
        if not self.last_poll:
            reasons.append("initial sync")
        elif now - self.last_poll > max(120, 3 * self.poll_interval):
            reasons.append(f"poll thread stalled: last finished "
                           f"{int(now - self.last_poll)}s ago")
        if getattr(self.watcher, "cut_short", False):
            reasons.append("the node stopped answering partway through the "
                           "last poll; chain states may lag until it answers")
        if self.poll_error:
            reasons.append(f"the last poll failed: {self.poll_error}")
        if stale:
            reasons.append("stale price: " + ", ".join(stale))
        # A market with a fresh, verified price that STILL cannot be lent
        # against. The usual cause is the registry: without it neither ticker
        # resolves to an asset id, so `lendable` is false for every market at
        # once, the lend form has nothing in it and the borrow tab is empty --
        # and every other health signal is green, because the prices are fine
        # and the chain is fine. Naming them is the difference between an
        # operator seeing a broken deployment and seeing a quiet one.
        try:
            dead = [r["market"] for r in self.market_rows()
                    if not r.get("cross_chain") and not r.get("lendable")
                    and not r.get("stale")]
        except Exception:                               # noqa: BLE001
            dead = []
        if dead:
            # WHY, distinguished. A book that has not fetched the registry yet
            # -- the first seconds after a start, or a registry that is not
            # answering -- has every market unlendable for a reason that is
            # about to fix itself, and sending an operator to "check the
            # registry" then is sending them to look at the one thing that is
            # not the problem. That is exactly when they are reading this.
            named = "priced but not lendable: " + ", ".join(sorted(dead)[:8])
            if not self._registry_at:
                why = (" -- this book has not read the asset registry yet, so "
                       "no ticker resolves to anything. A start-up clears this "
                       "on its own; if it does not, the registry is the thing "
                       "to look at")
            elif not self.assets:
                why = (" -- this book knows of NO assets at all: the registry "
                       "has been read and gave none, and no node answered "
                       "`dumpassetlabels` either. Nothing can be lent until "
                       "one of them does")
            else:
                why = (" -- their tickers do not resolve to any of the assets "
                       "this book knows, or the precisions disagree. Check "
                       "the registry")
            reasons.append(named + why)
        if self.rescan_needed_from is not None:
            reasons.append(
                f"this book was away for longer than it can look back: run "
                f"once with --rescan-from {self.rescan_needed_from}, or offers "
                f"and vaults that moved meanwhile read as gone or unknown")
        if UNRENDERABLE:
            # A record this book holds and cannot show. It is missing from
            # every listing, so a lender looking for their own offer simply
            # does not find it -- and the only trace was one log line, printed
            # once, whenever the message changed. Health goes false, and
            # `/v1/unrenderable` says which and why.
            reasons.append(
                f"{len(UNRENDERABLE)} record(s) cannot be rendered and are "
                f"missing from every listing; see /v1/unrenderable")
        if self.event_backlog:
            # Not fatal on its own -- one poll can leave an event waiting for a
            # book write -- but a backlog that persists means loans an offer
            # opened are not being registered, so it belongs in the reasons.
            reasons.append(f"{self.event_backlog} offer event(s) waiting to "
                           "be applied")
        return {
            "ok": not reasons,
            "error": "; ".join(reasons) or None,
            "version": __version__,
            "git_rev": GIT_REV,
            # How many golden vector cases the covenant tripwire checked in this
            # process. Zero would mean the builder was never loaded, so an
            # operator can confirm the check ran rather than assume it.
            "covenant_vectors": compat.verified_cases(),
            "height": self.height,
            "last_poll": self.last_poll, "markets": len(self.markets),
            # The height a rescan is needed from, when this book was away for
            # longer than a poll can look back; null otherwise.
            "rescan_needed_from": self.rescan_needed_from,
            "priced": len(self.fresh_prices()),
            "stale_markets": stale,
            "max_price_age": self.max_price_age,
            "min_depth": self.min_depth,
            "rescan_depth": self.rescan_depth,
            "back_scan_cap": self.back_scan_cap,
            "prune_after": self.prune_after,
            "offers": len(self.book.list_offers(status="open")),
            "loans": len(self.book.loans),
            "unrenderable": len(UNRENDERABLE),
            "assets": len(self.assets),
            "fee_rates": len(self.fee_table.get("rates", {})),
            "block_seconds": self.block_seconds,
            "reference_ticker": self.reference_ticker,
            "oracles": len(self.oracle_keys),
            "oracle_errors": list(self.oracle_errors),
            "event_errors": list(self.event_errors[-10:]),
            "event_backlog": self.event_backlog,
            "node": bool(self.node),
            # Whether this book can see the parent chain. Without it, a
            # cross-chain loan's Bitcoin-side deadlines are not checked here at
            # all, and every take it accepts says so in its warnings.
            "explorer_url": self.explorer_url,
            "oracle_public_url": self.oracle_public_url,
            "btc_node": btc_h is not None,
            # ...and the height itself, because every Bitcoin-side deadline a
            # borrower is shown or checks is measured against it. A page with
            # no parent-chain height cannot tell them when their collateral
            # becomes abortable, so it refuses to originate at all.
            "btc_height": btc_h,
            "compromised_keys": sorted(self.compromised_keys)}


class Server(ThreadingHTTPServer):
    """One thread per connection, up to `MAX_CONNECTIONS` of them.

    The stock server spawns a thread per accept with no ceiling, so a client
    holding a few hundred sockets open -- or a few hundred tabs -- is a few
    hundred threads, each able to ask for a full-book render. Past the cap an
    accepted socket is closed unread; a client that retries gets a slot when
    one frees.
    """
    # Threads that outlive the process are threads that keep it alive: a
    # shutdown should not wait on a client that has stopped talking.
    daemon_threads = True
    # The kernel's accept queue. The stock five means a burst of tab
    # reloads overflows it, the dropped ones retry a second later, and the
    # page reads as slow when the server was idle.
    request_queue_size = 128
    _slots = threading.BoundedSemaphore(MAX_CONNECTIONS)

    def process_request(self, request, client_address):
        if not self._slots.acquire(blocking=False):
            self.shutdown_request(request)
            return
        try:
            super().process_request(request, client_address)
        except Exception:
            self._slots.release()
            raise

    def process_request_thread(self, request, client_address):
        try:
            super().process_request_thread(request, client_address)
        finally:
            self._slots.release()


class Handler(BaseHTTPRequestHandler):
    svc = None
    server_version = "pignusd"
    protocol_version = "HTTP/1.1"
    # HTTP/1.1 means keep-alive, and a threading server gives every connection
    # a thread of its own. Without a timeout, a client that opens one and sends
    # nothing holds that thread for as long as it likes -- so a few hundred
    # idle sockets are a book that answers nobody, from a machine that has done
    # nothing wrong. Long enough for an ordinary slow request, short enough
    # that an abandoned socket is not a resource.
    timeout = 30

    # What every response from this daemon says about how a browser may use
    # it. The page has one inline <style> block and style attributes, no
    # inline scripts and no on*= handlers -- every handler is assigned in
    # app.js -- and fetches only same-origin relative paths; the wallet
    # extension's content script is not subject to a page's policy. So a
    # policy that forbids everything else costs the page nothing and takes
    # away: framing, which is how a visitor's own permissionless buttons --
    # repay for the borrower, liquidate, return to lender -- would be
    # clicked through an invisible overlay; scripts from anywhere but here;
    # and a browser guessing a JSON file is HTML.
    SECURITY_HEADERS = (
        ("Content-Security-Policy",
         "default-src 'self'; script-src 'self'; style-src 'self' "
         "'unsafe-inline'; img-src 'self' data:; connect-src 'self'; "
         "object-src 'none'; base-uri 'none'; form-action 'self'; "
         "frame-ancestors 'none'"),
        ("X-Frame-Options", "DENY"),
        ("X-Content-Type-Options", "nosniff"),
        ("Referrer-Policy", "no-referrer"),
    )

    def _security_headers(self):
        for k, v in self.SECURITY_HEADERS:
            self.send_header(k, v)

    def version_string(self):
        return "pignusd"          # not the Python version behind it

    def _send(self, code, payload, ctype="application/json", gz=None,
              extra=()):
        # Bytes pass through untouched -- a static .json file is already
        # serialised; only a Python object is turned into JSON. Serialising the
        # bytes (the old behaviour) 500'd every static .json the page fetched.
        # `gz` is the same bytes already gzipped, from the render cache; it is
        # what goes out when the client said it takes gzip.
        encoding = None
        if isinstance(payload, (bytes, bytearray)):
            body = bytes(payload)
            if gz is not None and "gzip" in (
                    self.headers.get("Accept-Encoding") or "").lower():
                body, encoding = gz, "gzip"
        elif ctype == "application/json":
            # allow_nan=False: `Infinity` is valid Python and invalid JSON, and
            # one of them in one record loses the whole document at the reader.
            body = json.dumps(payload, indent=2, allow_nan=False).encode()
        else:
            body = payload
        if not self._body_read:
            # Answering without having consumed the request body: the bytes
            # left on the socket would be read as the next request line, so
            # this connection ends here rather than desynchronising.
            self.close_connection = True
        self.send_response(code)
        if self.close_connection:
            self.send_header("Connection", "close")
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        if encoding:
            self.send_header("Content-Encoding", encoding)
            self.send_header("Vary", "Accept-Encoding")
        for k, v in extra:
            self.send_header(k, v)
        self._security_headers()
        # The page is served from here, so it is same-origin and needs none of
        # this. Reading the book from elsewhere is welcome; writing to it from
        # a visitor's browser on someone else's site is not, so the wildcard is
        # offered on reads only.
        if self.command in ("GET", "HEAD", "OPTIONS"):
            self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(body)
        return True

    def log_message(self, fmt, *args):
        # A manage token in a query string is a secret in the journal. The
        # header is the right place for it; redact it either way.
        line = re.sub(r"(?i)(token=)[^&\s\"]+", r"\1…", fmt % args)
        sys.stderr.write("%s - %s\n" % (self.address_string(), line))

    def _client_key(self):
        """Who to charge a write to.

        Behind a reverse proxy every request arrives from loopback, so keying
        the rate limit on the socket peer gives the whole internet one bucket
        and one flooder locks everybody out. The forwarded-for header is only
        believed from a peer configured as a proxy, and then only its LAST hop:
        earlier entries are whatever the client claimed. A request from a
        trusted peer with no such header is the box's own responder or CLI --
        it is not a public client, and it is not rate-limited.
        """
        peer = self.client_address[0] if self.client_address else ""
        if peer in self.svc.trusted_proxies:
            fwd = (self.headers.get("X-Forwarded-For")
                   or self.headers.get("X-Real-IP") or "")
            hop = fwd.split(",")[-1].strip()
            return hop or None
        return peer

    def _rate_limited(self):
        key = self._client_key()
        if key is None:
            return False
        now = time.time()
        return not WRITE_LIMIT.allow(key, now) or \
            not GLOBAL_WRITE_LIMIT.allow("all", now)

    def _list_limited(self):
        key = self._client_key()
        if key is None:
            return False
        return not LIST_LIMIT.allow(key, time.time())

    def _listing(self, name, key, build):
        """Answer a whole-book listing from the render cache, or render it.

        A listing is a function of the book (its `generation`), of what the
        poll last learned (`view_epoch`) and of the clock, because an
        attestation goes stale by age alone; the cache is keyed by all three
        and by the query. One render serves every tab that asks within the
        TTL. Renders run under one lock, so a burst of first-askers waits for
        one render instead of each making its own; past `RENDER_SLOTS` of
        them waiting, the answer is a 503 with `Retry-After` rather than a
        thread and tens of megabytes per request, without bound. The body is
        serialised once, compact, and gzipped once beside it.
        """
        if self._list_limited():
            return self._send(429, {"error": "too many listing requests: two "
                                             "a second per client with a burst "
                                             "of thirty; the page itself asks "
                                             "once every thirty seconds"})
        s = self.svc
        full = (name, s.book.generation, s.view_epoch,
                int(time.time() // RENDER_TTL)) + tuple(key)
        hit = s._render_cache.get(full)
        if hit is None:
            if not s._render_slots.acquire(blocking=False):
                return self._send(503, {"error": "the book is busy rendering; "
                                                 "ask again in a moment"},
                                  extra=(("Retry-After", "5"),))
            try:
                with s._render_lock:
                    hit = s._render_cache.get(full)
                    if hit is None:
                        raw = json.dumps(build(), allow_nan=False,
                                         separators=(",", ":")).encode()
                        hit = (raw, gzip.compress(raw, compresslevel=1))
                        if len(s._render_cache) >= 64:
                            s._render_cache.clear()
                        s._render_cache[full] = hit
            finally:
                s._render_slots.release()
        return self._send(200, hit[0], gz=hit[1])

    def _scan_limited(self):
        """A read that costs the NODE work, rather than one served from memory.

        `/v1/spend` walks up to `back_scan_cap` verbose blocks and `/v1/outpoint`
        asks the node per call. Anyone can reach both, and must be able to --
        a borrower recovering their own collateral has no account here -- so
        the answer is a ration rather than a gate.
        """
        key = self._client_key()
        if key is None:
            return False
        now = time.time()
        return not SCAN_LIMIT.allow(key, now) or \
            not GLOBAL_SCAN_LIMIT.allow("all", now)

    # The site is plain ES modules with no build step, so the browser fetches
    # them by name. Only files that actually exist in web/ are served, and only
    # by their bare name -- no path traversal, no directory listing, nothing
    # outside that one directory.
    STATIC = {".js": "application/javascript; charset=utf-8",
              ".json": "application/json",
              ".css": "text/css; charset=utf-8",
              ".svg": "image/svg+xml"}

    def _static(self, path):
        name = os.path.basename(path)
        if name != path.lstrip("/"):
            return False
        ext = os.path.splitext(name)[1]
        if ext not in self.STATIC or not name:
            return False
        full = os.path.realpath(os.path.join(WEB, name))
        if os.path.dirname(full) != os.path.realpath(WEB) or not os.path.isfile(full):
            return False
        with open(full, "rb") as f:
            return self._send(200, f.read(), self.STATIC[ext])

    def do_OPTIONS(self):
        self._body_read = True
        self.send_response(204)
        self.send_header("Content-Length", "0")
        self._security_headers()
        self.send_header("Cache-Control", "no-store")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()

    def do_HEAD(self):
        self.do_GET()

    # Connections are kept alive, so a request whose body is never read leaves
    # those bytes to be parsed as the next request line -- which surfaces as a
    # random 400 for whoever the proxy pools that connection to next.
    _body_read = True

    def _length(self):
        try:
            return int(self.headers.get("Content-Length") or 0)
        except ValueError:
            return -1

    def _cross_site(self, need_json=True):
        """A write from another origin, or one dressed as a plain form.

        Every state-changing endpoint needs a coin or a signature, so a
        cross-site write cannot move money; what it could do is spend a
        visitor's per-address rate-limit bucket from their own browser. A
        browser sends a cross-origin JSON POST only after a preflight, which
        the OPTIONS handler refuses -- unless the body is sent as text/plain,
        which needs none. So a POST must say it is JSON, and an Origin that
        is not this site is refused outright. A DELETE carries no body and
        already needs a custom header, which no browser sends unasked.
        """
        if need_json:
            ctype = (self.headers.get("Content-Type") or "").split(";")[0]
            if ctype.strip().lower() != "application/json":
                return "a write must carry Content-Type: application/json"
        origin = self.headers.get("Origin")
        if origin and origin != "null":
            host = (self.headers.get("Host") or "").lower()
            if origin.split("//", 1)[-1].lower() != host:
                return "a write from another origin is refused"
        return None

    def _body(self):
        n = self._length()
        if n < 0:
            self.close_connection = True
            raise ValueError("Content-Length must be a whole number of bytes")
        if n > MAX_BODY:
            self.close_connection = True
            raise ValueError("request body too large")
        raw = self.rfile.read(n)
        self._body_read = True
        return json.loads(raw.decode() or "{}")

    def _discard_body(self):
        """Read and throw away the body of a request being refused early."""
        n = self._length()
        if n < 0 or n > MAX_BODY:
            self.close_connection = True
            return
        if n:
            self.rfile.read(n)
        self._body_read = True

    def _get(self, path, q):
        s = self.svc
        if path in ("/", "/index.html"):
            with open(os.path.join(WEB, "index.html"), "rb") as f:
                return self._send(200, f.read(), "text/html; charset=utf-8")
        if self._static(path):
            return True
        if path == "/healthz":
            return self._send(200, s.health())
        if path == "/v1/markets":
            return self._send(200, {"markets": s.market_rows(),
                                    "height": s.height,
                                    # The parent chain's tip. A cross-chain
                                    # loan's abort and sweep are written in
                                    # Bitcoin blocks, so a page without this
                                    # cannot say when either opens.
                                    "btc_height": s.btc_height(),
                                    # What the parent chain currently charges,
                                    # in satoshis per vbyte. The one move a
                                    # cross-chain loan cannot fee-bump is
                                    # signed at origination, so a borrower
                                    # needs this BEFORE they commit anything:
                                    # a fee too low to confirm is a loan that
                                    # can never start, and the lender refuses
                                    # to pay a principal into one.
                                    "btc_feerate_sat_vb": s.btc_feerate(),
                                    "explorer_tx_url": s.explorer_tx_url,
                                    "btc_explorer_tx_url": s.btc_explorer_tx_url,
                                    "reference_ticker": s.reference_ticker,
                                    "block_seconds": s.block_seconds,
                                    "min_depth": s.min_depth,
                                    "max_price_age": s.max_price_age})
        if path == "/v1/assets":
            return self._send(200, {"assets": s.assets})
        if path == "/v1/fees":
            return self._send(200, s.fee_table)
        if path == "/v1/vectors":
            # The browser pins its own covenant implementation against
            # these before deriving any address a user will act on. Parsed
            # rather than streamed, so a corrupt file fails here and not in
            # a browser that is about to derive an address from it.
            try:
                with open(os.path.join(
                        os.path.dirname(WEB), "pignus", "vectors.json")) as f:
                    return self._send(200, json.load(f))
            except FileNotFoundError:
                return self._send(500, {"error": "pignus/vectors.json is "
                                        "missing; the browser has nothing to "
                                        "pin itself against"})
        if path == "/v1/oracle":
            # The PUBLIC address when the operator gave one. The address the
            # book reads from is loopback behind a proxy, and answering
            # "where is this oracle" with 127.0.0.1 sends a reader to their
            # own machine.
            public = s.oracle_public_urls[0] if s.oracle_public_urls else ""
            return self._send(200, {
                "oracle_x": s.oracle_x, "url": public or s.oracle_url,
                "note": "verify every attestation against the key the VAULT "
                        "bakes in, not against this one"})
        if path == "/v1/oracles":
            # The set of independent oracles this book quotes against, so a
            # lender can pick an m-of-n subset for a threshold loan.
            # PUBLIC urls, empty where none is configured. The book's own
            # addresses for its oracles are loopback, and answering "where is
            # this oracle" with 127.0.0.1 sends a reader to their own machine.
            #
            # Paired through the CONFIGURED url, never by position. `oracle_keys`
            # holds only the oracles that answered, so the moment one is
            # unreachable -- a restart is enough -- the two lists stop lining
            # up and every key after the gap is served with the next oracle's
            # address. An auditor sent to the wrong oracle finds no attestation
            # for a seizure and concludes it was never published, which is the
            # exact thing this endpoint exists to prevent.
            pairs = []
            for i, url in enumerate(s.oracle_urls):
                key = s.oracle_key_by_url.get(url)
                if not key:
                    continue            # has not answered; not something to
                                        # name an address for
                pairs.append((key, s.oracle_public_urls[i]
                              if i < len(s.oracle_public_urls) else ""))
            prev = sorted({k for ks in s.previous_keys.values() for k in ks})
            return self._send(200, {"oracles": [k for k, _u in pairs],
                                    "urls": [u for _k, u in pairs],
                                    # Keys these oracles USED to sign with,
                                    # so a loan baked to one reads as
                                    # rotated rather than as a stranger's,
                                    # and keys any of them has declared
                                    # compromised.
                                    "previous": prev,
                                    "compromised": sorted(s.compromised_keys)})
        if path == "/v1/offers":
            # A misspelt status returned an empty list, which reads as "there
            # are no offers" rather than "that is not a status" -- the same
            # trap `/v1/loans` already refuses for `state`.
            status = q.get("status", ["open"])[0]
            if status not in OFFER_STATUS_NAMES:
                raise ValueError("status must be one of "
                                 + ", ".join(sorted(OFFER_STATUS_NAMES)))
            # `expired` is derived from the chain, never stored, so it is
            # answered by filtering the open ones rather than by a stored
            # status nothing ever writes -- which answered every such query
            # with an empty list, the exact trap the check above exists for.
            market = q.get("market", [None])[0]
            _limit([], q)               # a bad limit is refused here, not cached
            limit = q.get("limit", [""])[0]

            def build():
                rows = s.book.list_offers(market=market,
                                          status="open" if status == "expired"
                                          else status)
                views = _views(s.offer_view, _limit(rows, q))
                if status == "expired":
                    views = [v for v in views if v.get("expired")]
                return {"offers": views}
            return self._listing("offers", (status, market, limit), build)
        if path.startswith("/v1/offer/") or path.startswith("/v1/offers/"):
            rec = s.book.offer(path.partition("/v1/")[2].partition("/")[2])
            if not rec:
                return self._send(404, {"error": "no such offer"})
            return self._send(200, s.offer_view(rec))
        if path == "/v1/loans":
            state = q.get("state", [None])[0]
            if state and state not in STATE_NAMES:
                # A misspelt state used to return an empty book, which reads as
                # "you have no loans" rather than "that is not a state".
                raise ValueError("state must be one of "
                                 + ", ".join(sorted(STATE_NAMES)))
            market = q.get("market", [None])[0]
            # By oracle key, so the loans still baked to a retired key can be
            # listed during a rotation -- that set is exactly what has to mature
            # before the old oracle can be switched off.
            ox = (q.get("oracle_x", [None])[0] or "").lower()
            _limit([], q)               # a bad limit is refused here, not cached
            limit = q.get("limit", [""])[0]

            def build():
                rows = s.book.list_loans(state=state, market=market)
                if ox:
                    rows = [r for r in rows if ox in _loan_oracle_keys(r)]
                return {"loans": _views(s.loan_view, _limit(rows, q))}
            return self._listing("loans", (state, market, ox, limit), build)
        if (path.startswith("/v1/loan/") or path.startswith("/v1/loans/")) \
                and path.endswith("/exit"):
            # How a loan ended, read back off the chain. A liquidated borrower
            # has no other way to see the price they were liquidated at.
            loan_id = path.partition("/v1/")[2].partition("/")[2][:-len("/exit")]
            if not s.book.loan(loan_id):
                return self._send(404, {"error": "no such loan"})
            if not s.watcher:
                return self._send(503, {"error": "no node configured; a close "
                                                 "cannot be read back"})
            out = s.explain_exit(loan_id)
            if out is not None:
                as_amounts(out)
            if out is None:
                return self._send(404, {"error": "that loan is still open, or "
                                                 "this book has not seen it "
                                                 "close"})
            return self._send(200, out)
        if path.startswith("/v1/loan/") or path.startswith("/v1/loans/"):
            rec = s.book.loan(path.partition("/v1/")[2].partition("/")[2])
            if not rec:
                return self._send(404, {"error": "no such loan"})
            return self._send(200, s.loan_view(rec))
        if path.startswith("/v1/outpoint/"):
            if self._scan_limited():
                return self._send(429, {
                    "error": "too many chain reads; slow down"})
            # What is actually at an outpoint right now, mempool included. The
            # browser uses this to confirm an offer still exists and holds what
            # the book says before composing anything against it.
            parts = path[len("/v1/outpoint/"):].split("/")
            if not s.node:
                return self._send(503, {"error": "no node configured; chain "
                                                 "lookups are unavailable"})
            if (len(parts) != 2 or not parts[1].isdigit()
                    or len(parts[0]) != 64 or not _is_hex(parts[0])):
                return self._send(400, {"error": "an outpoint is a 64-hex txid "
                                                 "and a vout number"})
            txid, vout = parts[0].lower(), int(parts[1])
            try:
                got = s.node.gettxout(txid, vout, True)
            except RpcError as e:
                return self._send(400, {"error": f"the node would not read "
                                                 f"that outpoint: {e}"})
            if got is None:
                return self._send(404, {"error": "unspent output not found; "
                                                 "it may already be spent"})
            if got.get("value") is None:
                return self._send(400, {"error": "that output is confidential; "
                                                 "its amount is not public"})
            return self._send(200, {
                "txid": txid, "vout": vout,
                "scriptPubKey": got["scriptPubKey"]["hex"],
                "asset": got.get("asset"),
                "value": str(atoms(got["value"])),
                "confirmations": got.get("confirmations", 0)})
        if path.startswith("/v1/scan/"):
            # Is there an unspent output at this address, in this asset, for
            # exactly this amount?
            #
            # The one question a browser cannot answer for itself and must be
            # able to: a payment whose report was lost -- the relay down, the
            # tab closed between broadcast and report -- is invisible to a page
            # rebuilding from the relay, and the page would then take the same
            # debt a second time. The address is derived from the terms, so
            # asking about it reveals nothing the asker did not already know.
            if self._scan_limited():
                return self._send(429, {
                    "error": "too many chain scans; slow down. This one walks "
                             "the whole UTXO set, so it is rationed"})
            if not s.node:
                return self._send(503, {"error": "no node configured; chain "
                                                 "lookups are unavailable"})
            spk = path[len("/v1/scan/"):]
            if not (2 <= len(spk) <= 200) or not _is_hex(spk):
                return self._send(400, {"error": "that is not a scriptPubKey"})
            asset = (q.get("asset", [None])[0] or "").lower()
            amount = q.get("amount", [None])[0]
            if asset and (len(asset) != 64 or not _is_hex(asset)):
                return self._send(400, {"error": "asset must be a 64-hex id"})
            if amount is not None and not str(amount).isdigit():
                return self._send(400, {"error": "amount is in atoms"})
            try:
                desc = s.node.getdescriptorinfo(f"raw({spk.lower()})")["descriptor"]
                res = s.node.scantxoutset("start", [desc]) or {}
            except RpcError as e:
                return self._send(400, {"error": f"the node would not scan "
                                                 f"that: {e}"})
            best = None
            for u in res.get("unspents", []):
                if asset and str(u.get("asset", "")).lower() != asset:
                    continue
                if amount is not None and atoms(u.get("amount", 0)) != int(amount):
                    continue
                if best is None or int(u.get("height", 0)) > int(best.get("height", 0)):
                    best = u
            if best is None:
                return self._send(200, {"found": False,
                                        "scriptPubKey": spk.lower()})
            tip = int(res.get("height", 0) or 0)
            return self._send(200, {
                "found": True, "scriptPubKey": spk.lower(),
                "txid": best["txid"], "vout": int(best["vout"]),
                "asset": best.get("asset"),
                "value": str(atoms(best.get("amount", 0))),
                "confirmations": (tip - int(best.get("height", 0)) + 1
                                  if tip and best.get("height") else 0)})
        if path.startswith("/v1/btc/outpoint/"):
            # Is a BITCOIN outpoint still unspent, and if not, what spent it?
            #
            # A cross-chain vault has three leaves and two of them belong to
            # the lender: SEIZE, which they and the oracle sign together with
            # no price test in any script, and TIMEOUT. Either empties the
            # vault at a moment the borrower is never told about -- so without
            # this a page shows a seized loan as live, with a Repay button, and
            # a borrower pays a debt for collateral that was taken before they
            # paid it.
            #
            # This book stays ignorant of the loan, exactly as `/v1/spend`
            # does: it returns the outpoint's state and the spend's witness,
            # and the page names the leaf from its own copy of the tree.
            if s.btc_node is None:
                return self._send(503, {"error": "this book has no Bitcoin "
                                                 "node, so it cannot say"})
            if self._scan_limited():
                return self._send(429, {
                    "error": "too many chain reads; slow down"})
            parts = path[len("/v1/btc/outpoint/"):].split("/")
            if len(parts) != 2 or len(parts[0]) != 64 \
                    or not parts[1].isdigit():
                return self._send(400, {"error": "expected "
                                                 "/v1/btc/outpoint/<txid>/<vout>"})
            btxid, bvout = parts[0].lower(), int(parts[1])
            from pignus import btc_collateral as BC
            try:
                got = s.btc_node.gettxout(btxid, bvout, True)
            except Exception as e:                      # noqa: BLE001
                return self._send(400, {"error": f"the Bitcoin node would not "
                                                 f"read that outpoint: {e}"})
            if got is not None:
                return self._send(200, {
                    "txid": btxid, "vout": bvout, "unspent": True,
                    "confirmations": int(got.get("confirmations", 0) or 0)})
            out = {"txid": btxid, "vout": bvout, "unspent": False,
                   "spend_txid": "", "witness": [], "confirmations": 0}
            try:
                found = BC.spend_witness(s.btc_node, btxid, bvout,
                                         scan_depth=s.back_scan_cap)
            except Exception:                           # noqa: BLE001
                found = None
            if found:
                out["spend_txid"], out["witness"], out["confirmations"] = found
            return self._send(200, out)
        if path.startswith("/v1/spend/"):
            if self._scan_limited():
                return self._send(429, {
                    "error": "too many chain scans; slow down. This one walks "
                             "blocks, so it is rationed"})
            # WHO spent an outpoint, and what they put in the witness to do it.
            #
            # This is how a borrower gets their collateral back when nobody
            # tells them the secret. The lender's claim of a repayment PUBLISHES
            # the preimage -- the covenant leaf forces it into the witness --
            # so the chain always has it even when the lender's relay report
            # never arrived, their node lost the claim, or they simply went
            # away. A page that could only read the relay's copy had one way to
            # learn it and no second one, and the collateral was the cost.
            from pignus import btc_collateral as BC     # noqa: PLC0415
            parts = path[len("/v1/spend/"):].split("/")
            if not s.node:
                return self._send(503, {"error": "no node configured; chain "
                                                 "lookups are unavailable"})
            if (len(parts) != 2 or not parts[1].isdigit()
                    or len(parts[0]) != 64 or not _is_hex(parts[0])):
                return self._send(400, {"error": "an outpoint is a 64-hex txid "
                                                 "and a vout number"})
            txid, vout = parts[0].lower(), int(parts[1])
            # A spend, once found, is a fact about a block: cache it. Without
            # this, a page polling for a secret walks the same blocks every few
            # seconds, and so does everybody else's.
            hit = s.spend_cache.get((txid, vout))
            if hit is not None:
                # The SPEND is a fact about a block and does not change. Its
                # DEPTH does, every minute -- and a browser gates a reclaim on
                # that depth, so serving a frozen one means the gate never
                # opens and a borrower can never take their collateral back.
                #
                # Recomputed from the BLOCK the spend landed in, which is what
                # was saved. Asking the node how deep the transaction is means
                # a `gettxout` that fails as soon as the spend's own outputs
                # are spent, and then a linear walk of the last few hundred
                # blocks -- on an unauthenticated endpoint that every open tab
                # polls, which made a cache hit ten times more expensive than
                # the miss it exists to avoid.
                out = dict(hit)
                height = out.pop("spend_height", None)
                out.pop("spend_block", None)
                conf = s.depth_at(height, hit.get("spend_block"))
                if conf is None:
                    # The block that held it is not at that height any more: a
                    # reorg. The cached fact is no longer a fact, so it goes,
                    # and the walk below finds where the spend is now.
                    s.spend_cache.pop((txid, vout), None)
                else:
                    out["confirmations"] = conf
                    out["anchor_confirmations"] = s.anchor_depth_at(
                        height, conf)
                    return self._send(200, out)
            try:
                from pignus import btc_collateral as BC
                if s.node.gettxout(txid, vout, True) is not None:
                    return self._send(404, {
                        "error": "that output is still unspent, so nothing has "
                                 "been published against it yet"})
                found = BC.spend_witness(s.node, txid, vout,
                                         scan_depth=s.back_scan_cap)
            except RpcError as e:
                return self._send(400, {"error": f"the node would not read "
                                                 f"that outpoint: {e}"})
            if not found:
                return self._send(404, {
                    "error": "the spend of that outpoint is not within this "
                             "book's scan window"})
            spend_txid, witness, conf = found
            # Every 32-byte item in the witness, with its hash. A caller knows
            # which hash its own loan commits to, so it can pick without this
            # book having to know anything about their loan.
            import hashlib
            # EVERY item is hashed, not only the 32-byte ones. The hashlock
            # leaf constrains the preimage's hash and nothing about its size,
            # so a claimant who chose a 33-byte secret spent the output just as
            # validly -- and a map that only held 32-byte items told the other
            # party they published nothing, which for a lender is a principal
            # gone with no way to start the loan and for a borrower collateral
            # they can never reclaim.
            preimages = {}
            for item in witness:
                try:
                    raw = bytes.fromhex(item)
                except ValueError:
                    continue
                if 0 < len(raw) <= 520:
                    preimages[hashlib.sha256(raw).hexdigest()] = item
            # And how deep the PARENT chain is behind it. Sequentia reorgs
            # whenever Bitcoin reorgs, so Sequentia confirmations measure the
            # wrong thing: six of them are six minutes, about six tenths of
            # one Bitcoin block, and an ordinary single-block Bitcoin reorg
            # undoes ten at once. What has to be buried is the Bitcoin header
            # the spend's own block anchored to, and this book holds a Bitcoin
            # node precisely so it can say.
            out = {"txid": txid, "vout": vout, "spend_txid": spend_txid,
                   "confirmations": conf,
                   "anchor_confirmations": s.anchor_depth_of(spend_txid, conf),
                   "preimages": preimages}
            # Only a MINED spend is a fact. One in the mempool can be replaced,
            # and caching that would serve a secret from a transaction that no
            # longer exists.
            if int(conf or 0) > 0:
                if len(s.spend_cache) >= 2048:
                    s.spend_cache.clear()
                # Without the depths -- they are recomputed on every read --
                # but WITH the block the spend landed in, which is what makes
                # recomputing them arithmetic rather than another walk. The
                # hash goes with the height so a reorg invalidates the entry
                # instead of quietly changing what it means.
                # From the NODE's tip, not the poll's. `conf` came back from
                # a walk that used the node's live height, and this book's own
                # `self.height` is whatever the last poll saw -- minutes old
                # after a restart, and arbitrarily old while a poll step is
                # failing. Subtracting one from the other names a block the
                # spend is not in, and `getblockhash` answers for it happily,
                # so the wrong block would be cached as a fact and every depth
                # computed from it afterwards would be wrong. That depth is
                # what a borrower's reclaim waits on.
                #
                # And then CHECKED: the block is fetched once, on the miss
                # path, and the spend must actually be in it. A height derived
                # by arithmetic is a guess until something confirms it.
                block_hash = None
                try:
                    height = int(s.node.getblockcount()) - int(conf) + 1
                    bh = s.node.getblockhash(height)
                    if spend_txid in (s.node.getblock(bh, 1).get("tx") or []):
                        block_hash = bh
                except Exception:                       # noqa: BLE001
                    block_hash = None
                if block_hash:
                    s.spend_cache[(txid, vout)] = {
                        **{k: v for k, v in out.items()
                           if k not in ("confirmations",
                                        "anchor_confirmations")},
                        "spend_height": height, "spend_block": block_hash}
            return self._send(200, out)
        if path.startswith("/v1/attestations/"):
            market = path[len("/v1/attestations/"):].replace("_", "/").upper()
            out = []
            for ox in s.oracle_keys:
                att = s.att_by_oracle.get(s._att_key(ox, market))
                if att is not None:
                    out.append(_attested(att, ox, s.max_price_age))
            return self._send(200, {"market": market, "attestations": out})
        if path.startswith("/v1/attestation/"):
            market = path[len("/v1/attestation/"):].replace("_", "/").upper()
            # `?oracle=` for one particular signer's, so a loan baked to a key
            # this book no longer calls primary can still be priced.
            ox = (q.get("oracle", [None])[0] or "").lower()
            att = (s.att_by_oracle.get(s._att_key(ox, market)) if ox
                   else s.attestations.get(market))
            if att is None:
                return self._send(404, {"error": f"no attestation for {market}"
                                        + (f" from oracle {ox[:16]}…" if ox
                                           else "")})
            return self._send(200, _attested(att, ox or s.oracle_x,
                                             s.max_price_age))
        if path == "/v1/btc/offers":
            status = q.get("status", ["open"])[0]
            if status not in OFFER_STATUS_NAMES:
                raise ValueError("status must be one of "
                                 + ", ".join(sorted(OFFER_STATUS_NAMES)))
            # COPIES. `list_btc_offers` hands out the stored records
            # themselves, so writing a computed field onto one wrote it into
            # the book -- and the next save persisted a number that is only
            # true for an instant.
            # One pass over the takes for every offer, not one per offer:
            # this listing is what every open tab polls.
            def build():
                left = s.book.btc_lots_left_all(BTC_TAKE_TTL, BTC_SIGNED_TAKE_TTL)
                offers = [served_loan(dict(o, lots_left=left.get(o["btc_offer_id"], 0)))
                    for o in s.book.list_btc_offers(status=status)]

                return {"offers": offers}
            return self._listing("btc_offers", (status,), build)
        if path.startswith("/v1/btc/offer/"):
            rec = s.book.btc_offer(path[len("/v1/btc/offer/"):])
            if not rec:
                return self._send(404, {"error": "no such BTC offer"})
            rec["lots_left"] = s.book.btc_offer_lots_left(
                rec["btc_offer_id"], BTC_TAKE_TTL, BTC_SIGNED_TAKE_TTL)
            return self._send(200, served_loan(rec))
        if path == "/v1/btc/takes":
            takes = s.book.list_btc_takes(
                offer_id=q.get("offer_id", [None])[0],
                status=q.get("status", [None])[0])
            # A browser's storage is not where a loan lives. Asking by borrower
            # key is what lets somebody who cleared it, or moved to another
            # machine, find their own loans again.
            who = (q.get("borrower_x", [None])[0] or "").lower()
            if who:
                takes = [t for t in takes if t.get("borrower_x", "") == who]
            # ...and by ORACLE, which is what an operator retiring a key needs.
            # `/v1/loans?oracle_x=` answers for the issued-asset tier only, so a
            # rotation checked against that alone reads "nothing depends on this
            # key" while live cross-chain loans still name it -- and those are
            # the ones that most need it, because on that tier the oracle's
            # signature IS the liquidation and no script can stand in for it.
            ox = (q.get("oracle_x", [None])[0] or "").lower()
            if ox:
                takes = [t for t in takes
                         if str((t.get("loan") or {}).get("oracle_x", "")).lower()
                         == ox]
            return self._send(200, {"takes": [served_loan(t) for t in takes]})
        if path.startswith("/v1/btc/take/"):
            # A COPY. Serialising the stored record hands `json.dumps` a dict
            # another thread is writing to, which raises mid-response on a
            # threaded server and answers a routine poll with a 500.
            rec = s.book.btc_take(path[len("/v1/btc/take/"):])
            if not rec:
                return self._send(404, {"error": "no such take"})
            return self._send(200, served_loan(rec))
        if path == "/v1/unrenderable":
            # WHICH records this book cannot show, and why. Counting them in
            # /healthz says something is wrong; this says what, so an operator
            # can go and look at the record rather than at the whole book.
            return self._send(200, {"unrenderable": dict(UNRENDERABLE)})
        if path == "/v1/stats":
            return self._listing("stats", (), lambda: s.book.stats(
                s.fresh_prices(), price_for=s.price_for))
        return self._send(404, {"error": "no such endpoint", "path": path})

    def do_GET(self):
        self._body_read = True          # a GET carries no body to drain
        u = urlparse(self.path)
        path = u.path.rstrip("/") or "/"
        try:
            self._get(path, parse_qs(u.query))
        except (BrokenPipeError, ConnectionResetError):
            # The client went away mid-response. There is nobody to tell.
            self.close_connection = True
        except FileNotFoundError:
            self._send(500, {"error": "web/index.html is missing"})
        except ValueError as e:
            self._send(400, {"error": str(e)})
        except Exception as e:                      # noqa: BLE001
            sys.stderr.write(f"GET {path}: {type(e).__name__}: {e}\n")
            self._send(500, {"error": f"{type(e).__name__}: {e}"})

    def do_POST(self):
        s = self.svc
        self._body_read = False
        path = urlparse(self.path).path.rstrip("/") or "/"
        why = self._cross_site()
        if why:
            self._discard_body()
            return self._send(400, {"error": why})
        if self._rate_limited():
            self._discard_body()
            return self._send(429, {"error": "too many writes; slow down"})
        try:
            body = self._body()
            if path == "/v1/offers":
                # Only what a publisher is entitled to say: everything else in
                # the body is dropped here rather than stored and served back
                # as if the book had checked it.
                offer = {k: body[k] for k in OFFER_FIELDS if k in body}
                # The cap counts LIVE listings, so a REPUBLISH of a coin that is
                # already listed adds none and is let through even when the book
                # is full: refusing a lender who is only refreshing their own
                # listing would read as a bug, and would leave a stale record on
                # the page instead of the corrected one.
                outpoint = str(offer.get("outpoint") or "")
                listed = any(o.get("outpoint") == outpoint
                             for o in s.book.list_offers(status="all"))
                if not listed and \
                        len(s.book.list_offers(status="open")) >= MAX_OFFERS:
                    return self._send(429, {"error": "the book is full"})
                s.verify_offer(offer)
                rec = s.book.put_offer(offer)     # carries manage_token once
                s.refresh_one(offer_id=rec["offer_id"])
                view = s.offer_view(s.book.offer(rec["offer_id"]) or rec)
                view["manage_token"] = rec.get("manage_token")
                return self._send(200, view)
            if path == "/v1/btc/offers":
                # A lender publishes a cross-chain offer. The relay is not
                # trusted for any of it, so the one thing it must establish is
                # that the lender really published it: an offer carrying
                # somebody else's key would make their own responder pay out.
                from pignus import btc_relay as R
                from pignus import btc_collateral as BC
                loan = dict(body.get("loan") or {})
                for k in ("btc_amount", "lender_x", "oracle_x", "recover_after",
                          "debt_asset", "debt", "repay_deadline", "abort_after",
                          "d_refund", "lender_prog", "upgrade_fee",
                          "market", "strike"):
                    if loan.get(k) in (None, ""):
                        raise ValueError(f"the offer's loan is missing {k}")
                # Every NUMBER is a number. These are stored verbatim and
                # served back to a page that does BigInt arithmetic on them, so
                # one offer carrying "lots" where an amount belongs throws out
                # of the render and takes every other offer on the page down
                # with it -- for every visitor, permanently, and it costs the
                # publisher nothing to leave there. Coerced here, at the door.
                for k in ("btc_amount", "debt", "principal", "repay_deadline",
                          "recover_after", "abort_after", "d_refund",
                          "upgrade_fee", "strike", "price_scale",
                          "lender_ver"):
                    if k not in loan or loan[k] in (None, ""):
                        continue
                    try:
                        n = int(str(loan[k]).strip())
                    except (TypeError, ValueError):
                        raise ValueError(
                            f"the offer's {k} is {loan[k]!r}, which is not a "
                            f"whole number") from None
                    if n < 0:
                        raise ValueError(f"the offer's {k} cannot be negative")
                    # Checked as a number, kept as the library says to keep it:
                    # the amounts that outgrow a browser's JSON parser go back
                    # out as decimal strings, the small ones as numbers.
                    loan[k] = str(n) if k in BC.BIG_LOAN_FIELDS else n
                market = str(body.get("market", ""))
                lots = max(1, int(body.get("lots", 1)))
                # A cap, the way `/v1/offers` has one. A cross-chain offer is
                # published with a signature and no chain state, so it costs
                # its publisher nothing -- and every one of them is
                # re-serialised, sorted and fsynced on every write this book
                # makes. Republishing an existing one adds none, so a lender
                # refreshing their own listing is never turned away.
                oid = R.offer_id(loan, market, lots)
                if oid not in s.book.btc_offers and \
                        len(s.book.list_btc_offers(status="open")) >= MAX_BTC_OFFERS:
                    return self._send(429, {
                        "error": "this book is holding as many cross-chain "
                                 "offers as it will. Withdraw one, or wait for "
                                 "an old one to be pruned."})
                if not R.verify_offer(loan, market, lots,
                                      str(body.get("offer_sig", ""))):
                    return self._send(403, {
                        "error": "that offer is not signed by the key it names "
                                 "as the lender. An unsigned offer would make "
                                 "somebody else's responder pay it out."})
                if int(loan["strike"]) <= 0:
                    raise ValueError(
                        "an offer must name the price a seizure would be "
                        "justified below. This tier's seizure is the lender "
                        "and the oracle signing together, with no price test "
                        "in the script, so the strike is the only thing that "
                        "makes one checkable afterwards")
                if loan.get("oracle_x") == loan.get("lender_x"):
                    raise ValueError("the lender cannot be their own oracle: "
                                     "that is collateral its holder may seize "
                                     "on their own say-so")
                if int(loan["upgrade_fee"]) < MIN_UPGRADE_FEE:
                    raise ValueError(
                        f"the fee carried for moving the collateral into the "
                        f"vault is {loan['upgrade_fee']} satoshis, under the "
                        f"{MIN_UPGRADE_FEE} this relay accepts. That "
                        "transaction is signed in advance and cannot be "
                        "replaced or fee-bumped, so a fee too low to confirm "
                        "is a loan that cannot start")
                # The four deadlines, against both tips, when this relay
                # has both. An offer past the margins was accepted with a
                # 200 and expired by the next poll, with a note only in this
                # daemon's own log; the lender saw an offer id.
                btc_h = s.btc_height()
                if btc_h is not None and s.height:
                    judged = dict(loan, borrower_x="00" * 32, h_w="00" * 32)
                    try:
                        problems = BC.timelocks_sane(
                            BC.loan_from_dict(judged), int(btc_h),
                            int(s.height))
                    except Exception as e:              # noqa: BLE001
                        raise ValueError(f"the offer's deadlines cannot be "
                                         f"judged: {type(e).__name__}: {e}")
                    if problems:
                        raise ValueError(
                            "the offer's deadlines leave no room, judged from "
                            f"Bitcoin block {btc_h} and Sequentia block "
                            f"{s.height}: " + "; ".join(problems))
                loan["lender_prog"] = R.check_program(loan["lender_prog"],
                                                      loan.get("lender_ver", 0))
                loan["lender_ver"] = int(loan.get("lender_ver", 0))
                rec = s.book.put_btc_offer({
                    "btc_offer_id": R.offer_id(loan, market, lots),
                    "loan": loan, "market": market, "lots": lots,
                    "offer_sig": str(body.get("offer_sig", "")),
                    "responder": body.get("responder", ""),
                    "note": str(body.get("note", ""))[:200]})
                return self._send(200, rec)
            if path.startswith("/v1/btc/offers/") and path.endswith("/resign"):
                # A lender replaces the SIGNATURE on an offer this book already
                # holds, and nothing else about it.
                #
                # An offer's signature is what makes it the lender's, and their
                # own responder checks it before acting on any take of the
                # offer -- so one that stops verifying stops every loan under
                # it, live ones included, and the collateral behind them is
                # never released. There are two ways that happens: a book that
                # altered the record, and an offer signed under an older
                # canonical form of the payload. The first cannot be repaired
                # here and should not be; the second is an upgrade, and a
                # lender who cannot repair it has to abandon the loans.
                #
                # It is safe to leave open because it changes nothing but a
                # signature, and only to one that verifies over the terms
                # ALREADY stored: an attacker with no key cannot produce one,
                # and the lender cannot alter a term with it.
                from pignus import btc_relay as R
                oid = path[len("/v1/btc/offers/"):-len("/resign")]
                offer = s.book.btc_offers.get(oid)
                if not offer:
                    return self._send(404, {"error": "no such offer"})
                sig = str(body.get("offer_sig", ""))
                if not R.verify_offer(offer["loan"], offer.get("market", ""),
                                      int(offer.get("lots", 1)), sig):
                    return self._send(403, {
                        "error": "that signature does not check out over the "
                                 "terms this book holds for that offer, under "
                                 "the key those terms name. Only the lender "
                                 "can re-sign an offer, and only over what is "
                                 "already stored"})
                s.book.update_btc_offer(oid, offer_sig=sig)
                return self._send(200, {"ok": True, "btc_offer_id": oid})
            if path.startswith("/v1/btc/offers/") and path.endswith("/withdraw"):
                from pignus import btc_relay as R
                oid = path[len("/v1/btc/offers/"):-len("/withdraw")]
                offer = s.book.btc_offers.get(oid)
                if not offer:
                    return self._send(404, {"error": "no such offer"})
                if not R.verify_report(offer["loan"]["lender_x"],
                                       R.WITHDRAW_TAG, oid,
                                       str(body.get("sig", ""))):
                    return self._send(403, {"error": "only the lender who "
                                            "published an offer can withdraw it"})
                s.book.update_btc_offer(oid, status="withdrawn")
                return self._send(200, {"ok": True, "btc_offer_id": oid})
            if path == "/v1/btc/take":
                # A borrower asks for a loan. Everything they claim is rebuilt
                # here from the offer's own terms -- but not the vault, which
                # depends on a secret the lender has not drawn yet. That is the
                # next step, and it is why this is a request rather than a take.
                offer = s.book.btc_offers.get(body.get("btc_offer_id"))
                if not offer:
                    return self._send(404, {"error": "no such BTC offer"})
                if offer.get("status", "open") != "open":
                    return self._send(409, {"error": "that offer is closed"})
                if s.book.btc_offer_lots_left(offer["btc_offer_id"],
                                              BTC_TAKE_TTL,
                                              BTC_SIGNED_TAKE_TTL) <= 0:
                    return self._send(409, {
                        "error": "every loan on that offer is already taken or "
                                 "spoken for; try again later or find another"})
                loan, dest, problems = s.btc_check_take(offer, body)
                from pignus import btc_collateral as BC
                from pignus import btc_relay as R
                # The borrower's signature over what they are taking. It is
                # what a seizure is later judged against, so it is required
                # here rather than left for a lender to ask for afterwards.
                take_auth = str(body.get("take_auth") or "")
                if not take_auth:
                    raise ValueError("a take needs `take_auth`: the borrower's "
                                     "signature over the offer id and the "
                                     "terms they chose")
                if not R.verify_take(loan.borrower_x, take_auth,
                                     btc_offer_id=offer["btc_offer_id"],
                                     borrower_x=loan.borrower_x, h_w=loan.h_w,
                                     borrower_prog=loan.borrower_prog,
                                     borrower_ver=loan.borrower_ver,
                                     prevault_txid=body["prevault_txid"],
                                     prevault_vout=body["prevault_vout"]):
                    raise ValueError("that take_auth does not verify against "
                                     "the borrower's key over this take")
                # The check above is a courtesy that answers early. This is the
                # one that decides: the last lot, the funding outpoint and the
                # commitment are all claimed under one lock, with the insert,
                # so two borrowers arriving together cannot both win.
                try:
                    rec = s.book.put_btc_take(
                        {"btc_offer_id": offer["btc_offer_id"],
                         "loan": BC.loan_to_dict(loan),
                         "borrower_x": loan.borrower_x,
                         "borrower_seq_spk": str(body.get("borrower_seq_spk", "")),
                         "w_seq": int(body.get("w_seq", 0)),
                         "prevault_txid": str(body["prevault_txid"]).lower(),
                         "prevault_vout": int(body["prevault_vout"]),
                         "prevault_value": str(loan.prevault_value()),
                         "prevault_spk": loan.prevault_spk().hex(),
                         "reclaim_dest": dest,
                         "reclaim_fee": int(body.get("reclaim_fee", 3000)),
                         "take_auth": take_auth,
                         "disbursement_spk": loan.disbursement_spk().hex(),
                         "status": "requested",
                         "warnings": problems},
                        lots_of=offer["btc_offer_id"], take_ttl=BTC_TAKE_TTL,
                        signed_ttl=BTC_SIGNED_TAKE_TTL)
                except ValueError as e:
                    if "lot" in str(e):
                        return self._send(409, {"error": str(e)})
                    raise
                return self._send(200, rec)
            if path == "/v1/btc/hash":
                # The lender draws this loan's secret and publishes its hash.
                # One secret per loan: sharing one across an offer's takes would
                # let any borrower's repayment release every other borrower's
                # collateral.
                from pignus import btc_relay as R
                take = s.book.btc_takes.get(body.get("take_id"))
                if not take:
                    return self._send(404, {"error": "no such take"})
                lender_x = take["loan"]["lender_x"]
                phash = str(body.get("payment_hash", "")).lower()
                point = str(body.get("adaptor_point", "")).lower()
                if len(phash) != 64:
                    raise ValueError("a payment hash is 32 bytes")
                if not R.verify_report(lender_x, R.HASH_TAG, take["take_id"],
                                       str(body.get("auth", "")),
                                       payment_hash=phash,
                                       adaptor_point=point):
                    return self._send(403, {"error": "that hash is not signed "
                                            "by this loan's lender"})
                if take.get("payment_hash") and take["payment_hash"] != phash:
                    return self._send(409, {"error": "this take already has a "
                                            "payment hash; it cannot change"})
                # One secret must never open both sides of a loan. If the hash
                # that releases the collateral is the hash that releases the
                # principal, the borrower's own claim publishes the secret that
                # frees their collateral and they keep both.
                if phash == str(take["loan"].get("h_w", "")).lower():
                    return self._send(400, {
                        "error": "that hash is the one the principal is locked "
                                 "to. One secret opening both sides would let "
                                 "the borrower keep the collateral as well."})
                # And it must not be another loan's. A hash shared between two
                # takes makes either borrower's repayment release the other's
                # collateral.
                if not s.book.claim_btc_hash(phash, take["take_id"]):
                    return self._send(409, {
                        "error": "that hash is already committed to another "
                                 "loan on this book"})
                loan = dict(take["loan"])
                loan["payment_hash"] = phash
                loan["adaptor_point"] = point
                from pignus import btc_collateral as BC
                ln = BC.loan_from_dict(loan)
                s.book.update_btc_take(
                    take["take_id"], loan=loan, payment_hash=phash,
                    adaptor_point=point, status="reserved",
                    hash_auth=str(body.get("auth", "")),
                    vault_txid=BC.upgrade_tx(ln, take["prevault_txid"],
                                             int(take["prevault_vout"])).txid(),
                    vault_vout=0,
                    repayment_spk=ln.repayment_spk().hex())
                return self._send(200, {"ok": True, "take_id": take["take_id"]})
            if path == "/v1/btc/presig":
                # The borrower signs the one transaction that can move their
                # collateral into the vault, now that they can derive it.
                from pignus import btc_relay as R
                from pignus import btc_collateral as BC
                take = s.book.btc_takes.get(body.get("take_id"))
                if not take:
                    return self._send(404, {"error": "no such take"})
                if not take.get("payment_hash"):
                    return self._send(409, {"error": "this take has no payment "
                                            "hash yet; there is no vault to "
                                            "sign for"})
                presig = str(body.get("upgrade_presig", "")).lower()
                loan = s.btc_take_loan(take)
                if not BC.check_upgrade_presig(loan, take["prevault_txid"],
                                               int(take["prevault_vout"]),
                                               bytes.fromhex(presig)):
                    raise ValueError("that signature does not move this "
                                     "collateral into this loan's vault")
                s.book.update_btc_take(take["take_id"], upgrade_presig=presig,
                                       status="pending")
                return self._send(200, {"ok": True, "take_id": take["take_id"]})
            if path == "/v1/btc/adaptor":
                # The lender returns the release: an ordinary signature over
                # the transaction that gives the collateral back, which the
                # borrower can check for themselves before committing anything.
                from pignus import btc_relay as R
                from pignus import btc_collateral as BC
                from pignus import adaptor as A
                take = s.book.btc_takes.get(body.get("take_id"))
                if not take:
                    return self._send(404, {"error": "no such take"})
                if not take.get("upgrade_presig"):
                    return self._send(409, {
                        "error": "this take has no advance signature yet, so "
                                 "there is nothing safe to release"})
                lender_x = take["loan"]["lender_x"]
                asig = str(body.get("release_sig")
                               or body.get("adaptor_sig", "")).lower()
                point = str(body.get("adaptor_point", take.get("adaptor_point", ""))).lower()
                phash = str(body.get("payment_hash", take.get("payment_hash", ""))).lower()
                if not R.verify_report(lender_x, R.ADAPTOR_TAG, take["take_id"],
                                       str(body.get("auth", "")),
                                       adaptor_point=point, payment_hash=phash,
                                       adaptor_sig=asig):
                    return self._send(403, {"error": "that reply is not signed "
                                            "by this loan's lender"})
                if phash != take.get("payment_hash"):
                    return self._send(409, {"error": "that release is for a "
                                            "different payment hash than this "
                                            "take reserved"})
                held = take.get("release_sig") or take.get("adaptor_sig")
                if held and held != asig:
                    return self._send(409, {"error": "this take already holds a "
                                            "release; it cannot be replaced"})
                ln = s.btc_take_loan(take)
                vault_txid = BC.upgrade_tx(ln, take["prevault_txid"],
                                           int(take["prevault_vout"])).txid()
                sighash = BC.sighash_for(
                    ln, BC.reclaim_tx(ln, vault_txid, 0,
                                      bytes.fromhex(take["reclaim_dest"]),
                                      int(take.get("reclaim_fee", 3000))),
                    "reclaim")
                if not A.verify(bytes.fromhex(lender_x), sighash,
                                bytes.fromhex(asig)):
                    raise ValueError("that release does not verify against the "
                                     "loan it claims to be for; refusing to "
                                     "store it")
                s.book.update_btc_take(
                    take["take_id"], release_sig=asig, adaptor_sig=asig,
                    status="signed",
                    # Kept, and served: it is what lets a borrower check that
                    # the payment hash their repayment commits to really came
                    # from the lender rather than from whoever runs this.
                    adaptor_auth=str(body.get("auth", "")),
                    vault_txid=vault_txid, vault_vout=0)
                return self._send(200, {"ok": True, "take_id": take["take_id"]})
            if path in ("/v1/btc/disbursed", "/v1/btc/upgraded",
                        "/v1/btc/claimed", "/v1/btc/refunded"):
                # Everything a lender's responder reports about a take. Each one
                # changes what a borrower believes and acts on, so each carries
                # the lender's signature over exactly the fields it reports.
                from pignus import btc_relay as R
                take = s.book.btc_takes.get(body.get("take_id"))
                if not take:
                    return self._send(404, {"error": "no such take"})
                lender_x = take["loan"]["lender_x"]
                kind = path.rsplit("/", 1)[1]
                tag, fields, update = {
                    "disbursed": (
                        R.DISBURSED_TAG,
                        {"txid": str(body.get("disbursement_txid", "")).lower(),
                         "vout": int(body.get("disbursement_vout", 0))},
                        lambda f: {"status": "disbursed",
                                   "disbursement_txid": f["txid"],
                                   "disbursement_vout": f["vout"]}),
                    "upgraded": (
                        R.UPGRADED_TAG,
                        {"txid": str(body.get("upgrade_txid", "")).lower()},
                        lambda f: {"status": "live", "upgrade_txid": f["txid"]}),
                    "claimed": (
                        R.CLAIMED_TAG,
                        {"txid": str(body.get("claim_txid", "")).lower(),
                         "secret": str(body.get("secret_t", "")).lower()},
                        # The secret is written only when there is one. A claim
                        # is reported twice -- once when it is made, again with
                        # the secret once it is buried -- and letting the empty
                        # first form overwrite the second would take back the
                        # one value that releases a borrower's collateral.
                        lambda f: ({"status": "claimed", "claim_txid": f["txid"]}
                                   | ({"secret_t": f["secret"]} if f["secret"]
                                      else {}))),
                    "refunded": (
                        R.REFUNDED_TAG,
                        {"txid": str(body.get("refund_txid", "")).lower()},
                        lambda f: {"status": "refunded",
                                   "refund_txid": f["txid"]}),
                }[kind]
                if not R.verify_report(lender_x, tag, take["take_id"],
                                       str(body.get("auth", "")), **fields):
                    return self._send(403, {
                        "error": "that report is not signed by this loan's "
                                 "lender, so it will not be believed"})
                if kind == "claimed" and fields["secret"]:
                    import hashlib
                    if hashlib.sha256(bytes.fromhex(fields["secret"])).hexdigest() \
                            != take["loan"].get("payment_hash"):
                        return self._send(400, {
                            "error": "that secret does not open this loan's "
                                     "repayment; refusing to publish it"})
                # A report always writes its FIELDS. It advances the status
                # only when that is forward -- so a `disbursed` report that
                # arrives (or is re-sent) after the loan is already `live`
                # records the disbursement rather than being refused, and
                # cannot drag the take backwards either. Without this a lender
                # whose report the relay refused once could never get it in,
                # and the borrower's page would show a step that had already
                # happened.
                wanted = update(fields)
                # KEEP the signature. The relay verifies it and used to throw
                # it away, which left a borrower with no way to check the one
                # thing a lender's report decides: the outpoint their claim is
                # built on. `hash_auth` and `adaptor_auth` are kept for exactly
                # this reason; these are the same kind of fact.
                wanted[f"{kind.replace('-', '_')}_auth"] = str(
                    body.get("auth", ""))
                s.book.update_btc_take(
                    take["take_id"],
                    **s.book.forward_only(take["take_id"], wanted))
                return self._send(200, {"ok": True, "take_id": take["take_id"]})
            if path in ("/v1/btc/claimed-principal", "/v1/btc/repaid"):
                # The BORROWER's own reports. Both are hints -- everything they
                # say is on chain anyway -- but an unauthenticated hint is a
                # denial of service: flipping a take out of the status a
                # responder scans would stop a principal being paid, or an
                # unclaimed one being taken back, for good. So the borrower
                # signs them with the key the take already names.
                from pignus import btc_relay as R
                take = s.book.btc_takes.get(body.get("take_id"))
                if not take:
                    return self._send(404, {"error": "no such take"})
                kind = path.rsplit("/", 1)[1]
                txid = str(body.get("txid") or body.get("claim_txid")
                           or body.get("repay_txid", ""))[:64].lower()
                vout = int(body.get("vout", 0))
                tag = (R.CLAIMED_PRINCIPAL_TAG if kind == "claimed-principal"
                       else R.REPAID_TAG)
                if not R.verify_report(take.get("borrower_x", ""), tag,
                                       take["take_id"],
                                       str(body.get("auth", "")),
                                       txid=txid, vout=vout):
                    return self._send(403, {
                        "error": "that report is not signed by this loan's "
                                 "borrower, so it will not be believed"})
                # A hint, and nothing more. It records where the borrower says
                # their payment landed and changes NO status. A status a
                # borrower can set is a status they can use: one that moved a
                # take out of the step a lender was acting on would leave the
                # principal paid and the collateral never taken into the vault.
                # It would also pin the offer's lot for ever, since a status
                # nothing else writes is one nothing else clears. The lender's
                # responder reads the chain and its own records instead, so
                # this field only ever saves it a scan.
                if kind == "claimed-principal":
                    s.book.update_btc_take(take["take_id"],
                                           principal_claim_txid=txid)
                else:
                    s.book.update_btc_take(take["take_id"], repay_txid=txid,
                                           repay_vout=vout)
                return self._send(200, {"ok": True})
            if path == "/v1/loans":
                open_loans = s.book.open_loans(CLOSED_NAMES)
                if open_loans >= MAX_LOANS:
                    return self._send(429, {"error": "the book is full"})
                # A loan is registered before its funding is buried, so a caller
                # can hand over a mempool transaction and then double-spend it.
                # Each of those leaves a GHOST behind, and nothing on chain was
                # ever spent to make one, so the only thing bounding them is
                # this: how many one client may have waiting at a time.
                who = self._client_key()
                waiting = s.pending_registrations(who)
                if waiting >= MAX_PENDING_PER_CLIENT:
                    return self._send(429, {
                        "error": f"you have {waiting} registrations still "
                                 "waiting for their funding to bury; wait for "
                                 "those to confirm before registering more"})
                got = s.verify_loan(body)
                # No caller-supplied state: a loan is not originated until its
                # funding is buried, and a register that could claim LIVE at
                # zero confirmations would say so on the page. The watcher
                # decides, from the chain, in the refresh two lines down.
                rec = s.book.put_loan(
                    body["terms"], body["txid"], body["vout"],
                    single_leaf=bool(body.get("single_leaf")),
                    confirmations=got["confirmations"],
                    offer_id=(body.get("offer_id")
                              if body.get("offer_id") in s.book.offers
                              else None))
                s.note_registration(who, rec["loan_id"])
                s.refresh_one(loan_id=rec["loan_id"])
                return self._send(200, s.loan_view(
                    s.book.loans.get(rec["loan_id"], rec)))
            return self._send(404, {"error": "no such endpoint", "path": path})
        except (BrokenPipeError, ConnectionResetError):
            self.close_connection = True
        except TakeMoved as e:
            # The state has moved on, which is what a replayed report meets.
            return self._send(409, {"error": str(e)})
        except OfferExists as e:
            # Not a bad request: the coin is listed and this caller did not
            # prove it owns the listing.
            return self._send(409, {"error": str(e)})
        except NotOnChain as e:
            return self._send(400, {"error": str(e)})
        except (KeyError, ValueError, TypeError) as e:
            # A 4xx body is read by a person, on the page, in a red box: the
            # Python class name in front of it told them nothing.
            sys.stderr.write(f"POST {path}: {type(e).__name__}: {e}\n")
            msg = (f"missing field {e.args[0]}" if isinstance(e, KeyError)
                   and e.args else str(e))
            return self._send(400, {"error": msg})
        except Exception as e:                      # noqa: BLE001
            sys.stderr.write(f"POST {path}: {type(e).__name__}: {e}\n")
            return self._send(500, {"error": f"{type(e).__name__}: {e}"})

    def do_DELETE(self):
        u = urlparse(self.path)
        self._body_read = False
        path = u.path.rstrip("/")
        why = self._cross_site(need_json=False)
        if why:
            self._discard_body()
            return self._send(400, {"error": why})
        if self._rate_limited():
            self._discard_body()
            return self._send(429, {"error": "too many writes; slow down"})
        self._discard_body()
        try:
            if path.startswith("/v1/offers/"):
                offer_id = path[len("/v1/offers/"):]
                token = (parse_qs(u.query).get("token", [None])[0]
                         or self.headers.get("X-Manage-Token"))
                rec = self.svc.book.offers.get(offer_id)
                if rec is None:
                    return self._send(404, {"removed": False})
                if not self.svc.book.manage_token_ok(offer_id, token):
                    # An anonymous flood cannot cancel a listing; only whoever
                    # published it holds the token. A stale listing whose coin
                    # is already spent is retired by the watcher regardless.
                    return self._send(403, {"error": "a listing can only be "
                                            "cancelled with the manage token "
                                            "returned when it was published"})
                gone = self.svc.book.delist_offer(offer_id)
                return self._send(200 if gone else 404, {
                    "removed": gone,
                    "note": "the listing is off the board; the record is kept, "
                            "because it is the only copy of the terms that can "
                            "withdraw the coin. Publish the coin again to "
                            "relist it, or withdraw it once the expiry opens"})
            return self._send(404, {"error": "no such endpoint"})
        except (BrokenPipeError, ConnectionResetError):
            self.close_connection = True
        except Exception as e:                      # noqa: BLE001
            sys.stderr.write(f"DELETE {path}: {type(e).__name__}: {e}\n")
            return self._send(500, {"error": f"{type(e).__name__}: {e}"})


def main():
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("--config", required=True)
    ap.add_argument("--once", action="store_true",
                    help="refresh once and print the state, without serving")
    ap.add_argument("--rescan-from", type=int, metavar="HEIGHT",
                    help="read the chain again from this height, forwards. "
                         "The backward walk a poll does is bounded, so a book "
                         "that was down for longer than rescan_depth blocks "
                         "cannot see what happened while it was away; this "
                         "does, a capped number of blocks per poll.")
    args = ap.parse_args()
    with open(args.config) as f:
        cfg = json.load(f)
    try:
        # Load the covenant before anything is served. Every vault address this
        # daemon shows is derived from it, and the golden vectors are checked on
        # the way through -- so a checkout that has drifted from the proven
        # builder stops here, rather than on the first offer somebody publishes.
        compat.load_covenant()
        svc = Service(cfg)
    except compat.CovenantUnavailable as e:
        print(f"error: {e}", file=sys.stderr)
        return 3

    if args.rescan_from is not None:
        if svc.watcher is None:
            print("error: --rescan-from needs a node; this config has no rpc",
                  file=sys.stderr)
            return 1
        svc.watcher.rescan_from(args.rescan_from)
        svc.rescan_needed_from = None
        svc._absence_checked = True
        print(f"pignusd: rescanning from height {args.rescan_from}",
              file=sys.stderr)

    if args.once:
        svc.refresh_registry(force=True)
        svc.refresh_prices()
        svc.refresh_fees()
        svc.refresh_chain()
        svc.last_poll = int(time.time())
        print(json.dumps({"markets": svc.market_rows(),
                          "stats": svc.book.stats(svc.fresh_prices()),
                          "health": svc.health()}, indent=2))
        return

    host, _, port = cfg.get("listen", "127.0.0.1:8741").rpartition(":")
    Handler.svc = svc
    srv = Server((host or "127.0.0.1", int(port)), Handler)
    # Bind and serve BEFORE the first sync. That sync re-tracks every loan and
    # can walk a long way back for spends it missed; done first, the page is a
    # proxy error for as long as it takes and nothing says why. Served first,
    # /healthz answers "initial sync" from the start.
    threading.Thread(target=svc.loop, args=(svc.poll_interval,),
                     daemon=True).start()
    print(f"pignusd: {len(svc.markets)} market(s), book {cfg['book']}, "
          f"serving on {host}:{port}; syncing", file=sys.stderr)
    srv.serve_forever()


if __name__ == "__main__":
    sys.exit(main())
