Skip to content

jmwalletd._backend

jmwalletd._backend

Blockchain backend factory.

Provides a get_backend() coroutine that creates and caches blockchain backend instances (Bitcoin Core RPC, descriptor wallet, Neutrino, etc.) based on application configuration.

Descriptor wallets are cached per JoinMarket wallet rather than as a single global instance. Bitcoin Core descriptor wallets carry per-wallet import state (imported xpubs, rescan cursor); sharing a single bitcoind descriptor wallet across multiple JM wallets means the second JM wallet's descriptors are never imported, so listunspent returns no UTXOs for it. The cache key is the deterministic wallet name derived from the JM wallet's BIP32 m/0 fingerprint (see :func:jmwallet.backends.descriptor_wallet.generate_wallet_name).

Classes

Functions:

get_backend(data_dir: Path, force_new: bool = False, *, mnemonic: str | None = None, network: str | None = None, wallet_service: WalletService | None = None) -> Any async

Return a blockchain backend instance.

For the descriptor backend, wallet_service or mnemonic must identify a specific JoinMarket wallet; the backend is built (and cached) with a per-wallet bitcoind descriptor wallet name derived from the wallet's BIP32 m/0 fingerprint.

Args: data_dir: Path to data directory (kept for API compatibility). force_new: If True, always build a fresh instance and do not cache it. Used by long-lived components (Maker, Taker) that close the backend on exit. mnemonic: BIP39 seed phrase. When given, network must also be provided. Takes precedence over wallet_service if both are supplied. network: Network name (mainnet / testnet / regtest). Required alongside mnemonic. wallet_service: Already-initialised wallet service. Its master key is used to derive the per-wallet bitcoind wallet name.

Returns: A backend instance implementing :class:BlockchainBackend.

Source code in jmwalletd/src/jmwalletd/_backend.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
async def get_backend(
    data_dir: Path,
    force_new: bool = False,
    *,
    mnemonic: str | None = None,
    network: str | None = None,
    wallet_service: WalletService | None = None,
) -> Any:
    """Return a blockchain backend instance.

    For the descriptor backend, ``wallet_service`` or ``mnemonic`` must
    identify a specific JoinMarket wallet; the backend is built (and cached)
    with a per-wallet bitcoind descriptor wallet name derived from the
    wallet's BIP32 m/0 fingerprint.

    Args:
        data_dir: Path to data directory (kept for API compatibility).
        force_new: If True, always build a fresh instance and do not cache
            it.  Used by long-lived components (Maker, Taker) that close
            the backend on exit.
        mnemonic: BIP39 seed phrase.  When given, ``network`` must also be
            provided.  Takes precedence over ``wallet_service`` if both are
            supplied.
        network: Network name (``mainnet`` / ``testnet`` / ``regtest``).
            Required alongside ``mnemonic``.
        wallet_service: Already-initialised wallet service.  Its master
            key is used to derive the per-wallet bitcoind wallet name.

    Returns:
        A backend instance implementing :class:`BlockchainBackend`.
    """
    from jmcore.settings import get_settings

    settings = get_settings()
    backend_type = settings.bitcoin.backend_type

    # Resolve descriptor wallet name for this JM wallet.
    descriptor_wallet_name: str | None = None
    if backend_type == "descriptor_wallet":
        if mnemonic is not None:
            if network is None:
                msg = "get_backend: network is required when mnemonic is passed"
                raise ValueError(msg)
            descriptor_wallet_name = _wallet_name_for_mnemonic(mnemonic, network)
        elif wallet_service is not None:
            descriptor_wallet_name = wallet_name_for_service(wallet_service)
        else:
            msg = (
                "get_backend: descriptor_wallet backend requires either a "
                "mnemonic or a wallet_service to derive the per-wallet "
                "bitcoind descriptor wallet name"
            )
            raise ValueError(msg)

    cache_key = descriptor_wallet_name

    if not force_new:
        cached = _backend_cache.get(cache_key)
        if cached is not None:
            return cached

    logger.info(
        "Initializing blockchain backend: {}{}{}",
        backend_type,
        f" (wallet={descriptor_wallet_name})" if descriptor_wallet_name else "",
        " (new instance)" if force_new else "",
    )

    rpc_url = settings.bitcoin.rpc_url
    rpc_user = settings.bitcoin.rpc_user
    # SecretStr -> str for backend constructors.
    raw_password = settings.bitcoin.rpc_password
    rpc_password: str = (
        raw_password.get_secret_value()
        if hasattr(raw_password, "get_secret_value")
        else str(raw_password)
    )

    instance: Any

    if backend_type == "descriptor_wallet":
        from jmwallet.backends.descriptor_wallet import DescriptorWalletBackend

        assert descriptor_wallet_name is not None  # narrowed above
        instance = DescriptorWalletBackend(
            rpc_url=rpc_url,
            rpc_user=rpc_user,
            rpc_password=rpc_password,
            wallet_name=descriptor_wallet_name,
        )
    elif backend_type == "neutrino":
        from jmwallet.backends.neutrino import NeutrinoBackend

        neutrino_url = getattr(settings.bitcoin, "neutrino_url", rpc_url)
        resolved_network = network or settings.network_config.network.value

        scan_lookback = getattr(
            settings.bitcoin,
            "neutrino_scan_lookback_blocks",
            settings.wallet.scan_lookback_blocks,
        )

        instance = NeutrinoBackend(
            neutrino_url=neutrino_url,
            network=resolved_network,
            scan_start_height=settings.wallet.scan_start_height,
            scan_lookback_blocks=scan_lookback,
            add_peers=settings.get_neutrino_add_peers(),
            tls_cert_path=settings.bitcoin.neutrino_tls_cert,
            auth_token=settings.bitcoin.neutrino_auth_token,
            include_mempool=settings.bitcoin.neutrino_include_mempool,
        )
    else:
        msg = f"Unknown backend type: {backend_type}"
        raise ValueError(msg)

    if force_new:
        return instance

    _backend_cache[cache_key] = instance
    return instance

reset_backend() -> None

Reset the backend cache (for testing).

Source code in jmwalletd/src/jmwalletd/_backend.py 
180
181
182
def reset_backend() -> None:
    """Reset the backend cache (for testing)."""
    _backend_cache.clear()

wallet_name_for_service(wallet_service: WalletService) -> str

Return the bitcoind descriptor wallet name for a WalletService.

Mirrors :func:jmwallet.backends.descriptor_wallet.generate_wallet_name but derives the fingerprint from the already-initialised master key on wallet_service so callers that do not hold the raw mnemonic (e.g. helpers that only receive a :class:WalletService) can still address the correct backend.

Source code in jmwalletd/src/jmwalletd/_backend.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
def wallet_name_for_service(wallet_service: WalletService) -> str:
    """Return the bitcoind descriptor wallet name for a ``WalletService``.

    Mirrors :func:`jmwallet.backends.descriptor_wallet.generate_wallet_name`
    but derives the fingerprint from the already-initialised master key on
    ``wallet_service`` so callers that do not hold the raw mnemonic (e.g.
    helpers that only receive a :class:`WalletService`) can still address
    the correct backend.
    """
    from jmwallet.backends.descriptor_wallet import generate_wallet_name

    fingerprint = wallet_service.master_key.derive("m/0").fingerprint.hex()
    return generate_wallet_name(fingerprint, wallet_service.network)