crypto_watcher: wire wallet_dist.enabled into Monero + Doge factories

Until now the wallet_dist.enabled = true flag in production.ini was
ornamental — only the mps_wallet_dist_health CLI smoke test ever
constructed an ErlangDist client. crypto_watcher's hot polling path
called get_client_from_settings() / get_dogecoin_client_from_settings()
which unconditionally built local HTTP clients against monero.rpc_url
and dogecoin.rpc_url. The toggle did not toggle.

This commit wires both factories to honor wallet_dist.enabled. Three
transports per coin, checked in order:

  1. Mock (monero.mock / dogecoin.mock) — test-only, unchanged
  2. Cluster RPC (wallet_dist.enabled = true) — returns
     ErlangDistMoneroClient / ErlangDistDogecoinClient that forward
     calls to Wallet.Service on cammy via portal's Wallet.Bridge
  3. Local HTTP daemon — current default, backwards-compatible

Existing erldist_clients module already had the dist client classes
(drop-in replacement surfaces matching MoneroClient / DogecoinClient);
they just weren't wired anywhere except the smoke CLI.

wallet_dist_config.py gains load_wallet_dist_config_from_settings()
alongside load_wallet_dist_config() (ini path). Both call shared
_build_from_dict() so validation stays in one place. The factory uses
the settings variant since it has the Pyramid settings dict already
and doesn't want to re-parse the ini.

## Rollback procedure

Two layers; both designed to be safe and reversible.

### Layer 1 — code rollback

Set wallet_dist.enabled = false in production.ini and restart
crypto_watcher. Factories drop back to local-daemon path
transparently. No code revert needed; the toggle IS the rollback.

The local monero-wallet-rpc and dogecoind services on mps-uwsgi1 must
stay running through the experiment window — they are the rollback
target. Do not decommission them in the same change as the cutover.

### Layer 2 — data rollback

CryptoPayment rows created during the cluster-RPC period reference
(account_index, subaddress_index) coordinates in CAMMY'S wallet
address space. If the toggle is flipped back without further action,
those rows would resolve to wrong addresses in MPS's local wallet.

Mitigation: pre-cutover, confirm no in-flight CryptoPayment rows.
During the experiment window, treat any pending rows as
force-expirable. If rollback is needed:

  1. Flip toggle off
  2. Force-expire any CryptoPayment rows created post-cutover (their
     cammy-side addresses become orphaned but no funds are at risk
     since the experiment assumes pre-cutover zero open payments)
  3. Resume local-wallet operation

This rollback path is only viable while the local daemons are still
running. Decommissioning them is a separate, later step taken only
after the experiment has soaked.

## Tests

14 new tests in test_crypto_clients_factory.py covering routing
branches for both coins:

  - Mock overrides everything (precedence pinned)
  - wallet_dist.enabled = true returns dist client + kwargs flow through
  - wallet_dist.enabled = false falls back to local
  - Missing wallet_dist key falls back to local (backwards-compat)
  - Local path errors clearly when rpc_url is also missing
  - load_wallet_dist_config_from_settings parity with ini loader
    (disabled / missing-keys / cookie-reading)

Dist client tests use monkeypatch to stub the constructor — we don't
want the factory test opening a real socket to find a Wallet.Bridge.

All 110 existing crypto_watcher tests still green.

## What this does NOT do

  - Does not flip wallet_dist.enabled in production.ini. Cutover is a
    deliberate ops step taken after a no-open-payments confirmation.
  - Does not migrate any DB rows. The shop-to-subaddress mapping
    stays as-is; new payments simply land on cammy's wallet from the
    cutover point forward.
  - Does not decommission local daemons. They remain as rollback
    targets until the cluster path has soaked.

make_post_sell/lib/crypto_watcher/crypto_clients.py

@@ -523,15 +523,39 @@ class MockDogecoinClient:
         }
 
 
-def get_client_from_settings(settings) -> MoneroClient:
+def get_client_from_settings(settings):
     """
-    Helper to construct a client from Pyramid settings.
-    Expects keys:
-      monero.rpc_url, monero.rpc_user, monero.rpc_pass
+    Construct a Monero client from Pyramid settings.
+
+    Three transports, checked in order:
+
+      1. Mock mode (``monero.mock = true``) — returns ``MockMoneroClient``.
+         Test-only path.
+      2. Cluster RPC (``wallet_dist.enabled = true``) — returns an
+         ``ErlangDistMoneroClient`` that forwards calls to
+         ``Wallet.Service`` on cammy via the portal bridge. Consolidates
+         wallet hosting onto cammy; mps-uwsgi1 no longer needs a local
+         ``monero-wallet-rpc`` once this is on.
+      3. Local HTTP daemon (``monero.rpc_url``) — current default,
+         backwards-compatible.
+
+    Toggle (2) is the consolidation path. To roll back: set
+    ``wallet_dist.enabled = false`` in production.ini and restart
+    crypto_watcher; the factory drops back to the local daemon path
+    (which is why we don't decommission the local daemons in the same
+    cutover).
     """
     if str(settings.get("monero.mock", "false")).lower() in ("1", "true", "yes"):
         path = settings.get("monero.mock_transfers_file") or "mock_transfers.json"
         return MockMoneroClient(path)
+
+    if _wallet_dist_enabled(settings):
+        from .erldist_clients import ErlangDistMoneroClient
+        from .wallet_dist_config import load_wallet_dist_config_from_settings
+
+        cfg = load_wallet_dist_config_from_settings(settings)
+        return ErlangDistMoneroClient(**cfg.client_kwargs())
+
     rpc_url = settings.get("monero.rpc_url")
     if not rpc_url:
         raise RuntimeError("monero.rpc_url not configured")
@@ -544,17 +568,27 @@ def get_client_from_settings(settings) -> MoneroClient:
 
 def get_dogecoin_client_from_settings(settings):
     """
-    Helper to construct a Dogecoin client from Pyramid settings.
+    Construct a Dogecoin client from Pyramid settings.
 
-    Uses standard Dogecoin Core RPC - works with both full and pruned nodes.
-    Pruned mode recommended: only ~2GB storage vs 50GB for full node.
+    Same three-transport branching as ``get_client_from_settings``:
+    mock → cluster RPC → local daemon. The cluster RPC path returns
+    an ``ErlangDistDogecoinClient`` that routes through
+    ``Wallet.Service`` on cammy via the portal bridge.
 
-    Settings:
-      dogecoin.rpc_url, dogecoin.rpc_user, dogecoin.rpc_pass
+    Local-daemon mode (``dogecoin.rpc_url``) stays the default until
+    ``wallet_dist.enabled`` is flipped on. Pruned local mode is
+    recommended for the legacy path (~2GB vs 50GB).
     """
     if str(settings.get("dogecoin.mock", "false")).lower() in ("1", "true", "yes"):
         return MockDogecoinClient()
 
+    if _wallet_dist_enabled(settings):
+        from .erldist_clients import ErlangDistDogecoinClient
+        from .wallet_dist_config import load_wallet_dist_config_from_settings
+
+        cfg = load_wallet_dist_config_from_settings(settings)
+        return ErlangDistDogecoinClient(**cfg.client_kwargs())
+
     rpc_url = settings.get("dogecoin.rpc_url")
     if not rpc_url:
         raise RuntimeError("dogecoin.rpc_url not configured")
@@ -564,3 +598,12 @@ def get_dogecoin_client_from_settings(settings):
         rpc_user=settings.get("dogecoin.rpc_user"),
         rpc_pass=settings.get("dogecoin.rpc_pass"),
     )
+
+
+def _wallet_dist_enabled(settings) -> bool:
+    return str(settings.get("wallet_dist.enabled", "false")).strip().lower() in (
+        "1",
+        "true",
+        "yes",
+        "on",
+    )

make_post_sell/lib/crypto_watcher/wallet_dist_config.py

@@ -75,6 +75,28 @@ def load_wallet_dist_config(ini_path: str) -> WalletDistConfig:
         raise ValueError(f"{ini_path} has no [app:main] section")
     app = cp["app:main"]
 
+    return _build_from_dict(dict(app), source=ini_path)
+
+
+def load_wallet_dist_config_from_settings(settings) -> WalletDistConfig:
+    """Same shape as ``load_wallet_dist_config`` but reads from a Pyramid
+    ``settings`` dict instead of an ini path.
+
+    Used by the crypto_watcher client factory so the running app can
+    branch on ``wallet_dist.enabled`` without re-parsing the ini.
+    """
+    # Filter to wallet_dist.* keys so unrelated settings can't trip
+    # validation. ConfigParser-style keys come through as flat dotted
+    # strings in Pyramid's settings dict — same shape as `app` above.
+    relevant = {k: v for k, v in settings.items() if k.startswith("wallet_dist.")}
+    return _build_from_dict(relevant, source="pyramid settings")
+
+
+def _build_from_dict(app: dict, *, source: str) -> WalletDistConfig:
+    """Shared validation/assembly. ``app`` is a dict-like of dotted
+    ``wallet_dist.*`` keys (the [app:main] section, or a slice of
+    Pyramid settings). ``source`` shows up in error messages.
+    """
     enabled = _truthy(app.get("wallet_dist.enabled"))
     if not enabled:
         return WalletDistConfig(
@@ -105,9 +127,11 @@ def load_wallet_dist_config(ini_path: str) -> WalletDistConfig:
     try:
         cookie = Path(cookie_file).read_text().strip()
     except OSError as e:
-        raise ValueError(f"cannot read wallet_dist.cookie_file {cookie_file!r}: {e}")
+        raise ValueError(
+            f"cannot read wallet_dist.cookie_file {cookie_file!r} (from {source}): {e}"
+        )
     if not cookie:
-        raise ValueError(f"wallet_dist.cookie_file is empty: {cookie_file}")
+        raise ValueError(f"wallet_dist.cookie_file is empty: {cookie_file} (from {source})")
 
     # TLS material — all three or none. None = plain inet_tcp_dist
     # (acceptable when peer also runs plaintext, e.g. dev). All three =

make_post_sell/tests/test_crypto_clients_factory.py

@@ -0,0 +1,205 @@
+"""Tests for ``get_client_from_settings`` / ``get_dogecoin_client_from_settings``
+transport branching.
+
+Three transports per coin:
+  - mock mode (overrides everything else)
+  - cluster RPC via ErlangDist client (when wallet_dist.enabled = true)
+  - local HTTP daemon (default fallback)
+
+These tests pin the routing logic — adding a new transport or changing
+the precedence rules without updating these tests becomes a test
+failure rather than a silent regression on the cutover.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from make_post_sell.lib.crypto_watcher.crypto_clients import (
+    DogecoinClient,
+    MockDogecoinClient,
+    MockMoneroClient,
+    MoneroClient,
+    get_client_from_settings,
+    get_dogecoin_client_from_settings,
+)
+
+
+def _wallet_dist_settings(tmp_path: Path, extra=None) -> dict:
+    """Build a settings dict with a real cookie file so the factory's
+    cluster-RPC branch can fully construct an ErlangDist client."""
+    cookie = tmp_path / "cookie"
+    cookie.write_text("TEST_COOKIE_VALUE")
+    base = {
+        "wallet_dist.enabled": "true",
+        "wallet_dist.our_node_name": "mps@test.local",
+        "wallet_dist.peer_name": "portal",
+        "wallet_dist.peer_host": "unsandbox.com",
+        "wallet_dist.registered_name": "Elixir.Wallet.Bridge",
+        "wallet_dist.cookie_file": str(cookie),
+    }
+    if extra:
+        base.update(extra)
+    return base
+
+
+class TestMoneroFactoryRouting:
+    def test_mock_overrides_everything(self, tmp_path):
+        # Mock takes precedence over both cluster and local.
+        cookie = tmp_path / "cookie"
+        cookie.write_text("x")
+        mock_file = tmp_path / "transfers.json"
+        mock_file.write_text("[]")
+
+        settings = {
+            "monero.mock": "true",
+            "monero.mock_transfers_file": str(mock_file),
+            "monero.rpc_url": "http://local:18083/json_rpc",
+            "wallet_dist.enabled": "true",
+        }
+        client = get_client_from_settings(settings)
+        assert isinstance(client, MockMoneroClient)
+
+    def test_wallet_dist_enabled_returns_dist_client(self, tmp_path, monkeypatch):
+        settings = _wallet_dist_settings(tmp_path)
+        # Cluster transport takes precedence over local rpc_url.
+        settings["monero.rpc_url"] = "http://should-not-be-used:18083/json_rpc"
+
+        # Patch the dist client so the factory test doesn't open a real
+        # socket. We're testing routing, not network plumbing.
+        captured = {}
+
+        class _StubDist:
+            def __init__(self, **kwargs):
+                captured["kwargs"] = kwargs
+
+        monkeypatch.setattr(
+            "make_post_sell.lib.crypto_watcher.erldist_clients.ErlangDistMoneroClient",
+            _StubDist,
+        )
+
+        client = get_client_from_settings(settings)
+
+        assert isinstance(client, _StubDist)
+        # Verify the cluster-RPC kwargs actually flowed through from
+        # settings (cookie, node names, registered_name).
+        assert captured["kwargs"]["our_name"] == "mps@test.local"
+        assert captured["kwargs"]["peer_name"] == "portal"
+        assert captured["kwargs"]["registered_name"] == "Elixir.Wallet.Bridge"
+        assert captured["kwargs"]["cookie"] == "TEST_COOKIE_VALUE"
+
+    def test_wallet_dist_false_falls_back_to_local(self):
+        settings = {
+            "wallet_dist.enabled": "false",
+            "monero.rpc_url": "http://localhost:18083/json_rpc",
+        }
+        client = get_client_from_settings(settings)
+        assert isinstance(client, MoneroClient)
+        assert not isinstance(client, MockMoneroClient)
+
+    def test_no_wallet_dist_key_falls_back_to_local(self):
+        # Backwards-compat: pre-existing deployments without wallet_dist
+        # keys at all keep using local daemon transparently.
+        settings = {"monero.rpc_url": "http://localhost:18083/json_rpc"}
+        client = get_client_from_settings(settings)
+        assert isinstance(client, MoneroClient)
+
+    def test_local_path_errors_when_no_rpc_url(self):
+        # Falling back to local with no rpc_url configured is a hard error
+        # — better to crash than to silently use a default that might be
+        # pointing at the wrong daemon.
+        with pytest.raises(RuntimeError, match="monero.rpc_url"):
+            get_client_from_settings({"wallet_dist.enabled": "false"})
+
+
+class TestDogecoinFactoryRouting:
+    def test_mock_overrides_everything(self):
+        settings = {
+            "dogecoin.mock": "true",
+            "dogecoin.rpc_url": "http://local:22555/",
+            "wallet_dist.enabled": "true",
+        }
+        client = get_dogecoin_client_from_settings(settings)
+        assert isinstance(client, MockDogecoinClient)
+
+    def test_wallet_dist_enabled_returns_dist_client(self, tmp_path, monkeypatch):
+        settings = _wallet_dist_settings(tmp_path)
+        settings["dogecoin.rpc_url"] = "http://should-not-be-used:22555/"
+
+        captured = {}
+
+        class _StubDist:
+            def __init__(self, **kwargs):
+                captured["kwargs"] = kwargs
+
+        monkeypatch.setattr(
+            "make_post_sell.lib.crypto_watcher.erldist_clients.ErlangDistDogecoinClient",
+            _StubDist,
+        )
+
+        client = get_dogecoin_client_from_settings(settings)
+
+        assert isinstance(client, _StubDist)
+        assert captured["kwargs"]["our_name"] == "mps@test.local"
+        assert captured["kwargs"]["peer_name"] == "portal"
+
+    def test_wallet_dist_false_falls_back_to_local(self):
+        settings = {
+            "wallet_dist.enabled": "false",
+            "dogecoin.rpc_url": "http://localhost:22555/",
+        }
+        client = get_dogecoin_client_from_settings(settings)
+        assert isinstance(client, DogecoinClient)
+
+    def test_local_path_errors_when_no_rpc_url(self):
+        with pytest.raises(RuntimeError, match="dogecoin.rpc_url"):
+            get_dogecoin_client_from_settings({"wallet_dist.enabled": "false"})
+
+
+class TestWalletDistSettingsLoader:
+    """Pin the new from-settings loader's parity with the ini loader."""
+
+    def test_load_from_settings_returns_disabled_when_flag_false(self):
+        from make_post_sell.lib.crypto_watcher.wallet_dist_config import (
+            load_wallet_dist_config_from_settings,
+        )
+
+        cfg = load_wallet_dist_config_from_settings({"wallet_dist.enabled": "false"})
+        assert cfg.enabled is False
+
+    def test_load_from_settings_returns_disabled_when_no_keys(self):
+        from make_post_sell.lib.crypto_watcher.wallet_dist_config import (
+            load_wallet_dist_config_from_settings,
+        )
+
+        cfg = load_wallet_dist_config_from_settings({"unrelated": "key"})
+        assert cfg.enabled is False
+
+    def test_load_from_settings_validates_required_keys(self, tmp_path):
+        from make_post_sell.lib.crypto_watcher.wallet_dist_config import (
+            load_wallet_dist_config_from_settings,
+        )
+
+        # Enabled but missing required keys → ValueError, same as ini loader.
+        with pytest.raises(ValueError, match="missing keys"):
+            load_wallet_dist_config_from_settings({"wallet_dist.enabled": "true"})
+
+    def test_load_from_settings_reads_cookie_file(self, tmp_path):
+        from make_post_sell.lib.crypto_watcher.wallet_dist_config import (
+            load_wallet_dist_config_from_settings,
+        )
+
+        cookie = tmp_path / "cookie"
+        cookie.write_text("ROLLBACK_TEST_COOKIE")
+
+        cfg = load_wallet_dist_config_from_settings({
+            "wallet_dist.enabled": "true",
+            "wallet_dist.our_node_name": "mps@test",
+            "wallet_dist.peer_name": "portal",
+            "wallet_dist.peer_host": "unsandbox.com",
+            "wallet_dist.cookie_file": str(cookie),
+        })
+        assert cfg.enabled is True
+        assert cfg.cookie == "ROLLBACK_TEST_COOKIE"