fix(horizon): honour shutdown, resume stream cursor, skip unhashed records, persist asset_issuer

[Manuel1234477]

Summary

Four defects in the Horizon settlement path. Three of them (#226, #228, #224) only
bite on a restart or an unexpected payload, which is why they survived; the fourth
(#223) is a missing column that makes historical rows unauditable.

Closes #226
Closes #228
Closes #224
Closes #223

Warning

This branch is implemented but untested. No tests were added and no test
suite was run — deliberately, per the request. cargo build, cargo clippy --all-targets and cargo fmt --check are clean, and the existing suite was
updated only where a struct literal needed the new field to keep compiling.
The tests each issue asks for (a multi-page catch-up against a mock Horizon, a
restart-gap stream test, a transaction_hash: None unit test) are not in
this branch and are still owed. Do not merge without running the suite.

---

#226 — poll_once now observes the shutdown signal

poll_once paged forward until caught up without ever checking shutdown. A
gateway down for a while faces a catch-up measured in minutes — each page is up
to 200 records, each meaning DB writes and an outbound webhook with retries —
and none of it saw SIGTERM. The task was killed mid-page once the 30 s grace
expired, replaying the unfinished page on the next boot and making the next
shutdown worse.

• The receiver is threaded into poll_once and checked at the top of every page
  iteration — i.e. immediately after the previous page's cursor checkpoint, so
  returning early never loses processed work.
• New POLL_MAX_PAGES_PER_CYCLE (default 50, 0 = unlimited) bounds one
  cycle even with no shutdown pending, so a catch-up cannot monopolise the
  poller task indefinitely. The next cycle resumes from the checkpoint.

#228 — the stream listener resumes from a persisted cursor

run_stream_listener hard-coded cursor = "now" on every process start,
discarding exactly the resume the poller works to persist. The advertised ~1 s
settlement latency therefore did not hold across a deploy: the gap was left to
the poller, whose catch-up is slower and which is disabled outright in
STELLAR_LISTENER_MODE=poll.

• The stream seeds from its own persisted cursor (horizon_stream_cursor),
  falls back to the poller's (horizon_payment_cursor), and only then to the
  live edge — so only a genuinely fresh database baselines at now.
• The two cursors live under separate kv_state keys, so neither listener drags
  the other backwards. The stream persists its cursor as each event advances it;
  a write failure is logged and tolerated (the poller remains the backstop).
• Replaying records the poller already saw is harmless — settlement is
  idempotent through processed_transactions.

#224 — records with no transaction_hash are skipped, never credited

reconcile_payment defaulted a missing hash to "" and used it as half of the
processed_transactions primary key. Two different unhashed transactions
therefore looked like the same transaction, and the second was discarded as
"already credited" when it never had been: money on chain, never credited to the
merchant. The empty string also landed in payments.tx_hash, pointing the
merchant-visible record at nothing.

• Such a record is now skipped with a warn!, and the poller re-sees it on the
  next cycle — so skipping is safe and self-healing.
• New counter stellargate_horizon_records_skipped_total{reason="missing_tx_hash"}
  on /metrics, so operators can see it happening.
• processed_transactions.tx_hash carries a CHECK (tx_hash <> '') for new
  databases, and record_processed_tx refuses an empty hash outright.
• Rows written before this fix are reported at startup, not deleted — the
  amount they carry was really received, and dropping the row would silently
  reduce an intent's paid total. They need reconciling by hand.

#223 — payments.asset_issuer

The table stored only the asset code, so which USDC a row referred to lived in
process configuration: editing ACCEPTED_ASSETS retroactively reinterpreted
every historical row, "asset": "USDC" in a webhook did not say which USDC,
and after an incident there was no way to prove what an intent was priced in.

• asset_issuer TEXT added to payments (nullable — NULL means native).
• payments::create resolves the whole AcceptedAsset and persists both halves.
• Exposed in GET /payments/:id (owner view) and in every webhook payload;
  documented in openapi.yaml, README.md and WEBHOOK_REFERENCE.md.
• verify() now matches against the issuer recorded on the intent, falling
  back to configuration only for pre-migration rows — so editing
  ACCEPTED_ASSETS no longer changes what a historical row means.
• Existing rows are backfilled once from the configured allow-list, guarded by a
  kv_state marker rather than re-running every boot (re-running would let a
  later config edit rewrite history a second time). Best-effort by nature, and
  documented as such.

---

Notes for review

• poll_once's signature changed to take &watch::Receiver<bool>; run_poller
  is the only caller.
• db::Payment, db::NewPayment, AppState and metrics::render each gained a
  field/parameter, which is most of the diff's line count in the test files.
• db::backfill_asset_issuers is called from main rather than db::migrate
  because it needs the accepted-asset allow-list.

[drips-wave]

@Manuel1234477 Great news! 🎉 Based on an automated assessment of this PR, the linked Wave issue(s) no longer count against your application limits.

You can now already apply to more issues while waiting for a review of this PR. Keep up the great work! 🚀

Learn more about application limits