jmwallet.backends.base

`jmwallet.backends.base`

Base blockchain backend interface.

Attributes

`logger = logging.getLogger(name)` `module-attribute`

Classes

`BlockchainBackend`

Bases: ABC

Abstract blockchain backend interface. Implementations provide access to blockchain data without requiring Bitcoin Core wallet functionality (avoiding BerkeleyDB issues).

Source code in jmwallet/src/jmwallet/backends/base.py

class BlockchainBackend(ABC):
    """
    Abstract blockchain backend interface.
    Implementations provide access to blockchain data without requiring
    Bitcoin Core wallet functionality (avoiding BerkeleyDB issues).
    """

    supports_descriptor_scan: bool = False
    """Whether this backend supports efficient descriptor-based UTXO scanning.

    Backends that override ``scan_descriptors()`` with a real implementation
    (e.g. Bitcoin Core's ``scantxoutset``) should set this to ``True``.
    Light-client backends (Neutrino) leave it at the default ``False`` so that
    ``sync_all()`` does not attempt descriptor scanning and fall back with a
    confusing warning.
    """

    supports_watch_address: bool = False
    """Whether this backend requires addresses to be pre-registered via ``add_watch_address()``.

    Light-client backends (Neutrino) must be told which addresses to watch before
    a rescan.  Full-node backends (Bitcoin Core, descriptor wallet) can query any
    address on demand and do not need pre-registration.
    Set to ``True`` only in backends that implement ``add_watch_address()``.
    """

    async def add_watch_address(self, address: str) -> None:
        """Register an address for watching.

        Only meaningful for backends with ``supports_watch_address = True``
        (i.e. light-client backends that need explicit address registration
        before a rescan).  Full-node backends can ignore this.
        """

    def set_wallet_creation_height(self, height: int | None) -> None:
        """Provide the block height at which the wallet was created.

        Backends can use this as a hint to skip scanning blocks before
        the wallet existed, avoiding unnecessary work during initial sync.
        Only used when no explicit ``scan_start_height`` is configured.

        Passing ``None`` clears any previously set hint.

        The default implementation is a no-op; backends that support
        scan optimisation override this method.
        """

    @abstractmethod
    async def get_utxos(self, addresses: list[str]) -> list[UTXO]:
        """Get UTXOs for given addresses"""

    @abstractmethod
    async def get_address_balance(self, address: str) -> int:
        """Get balance for an address in satoshis"""

    @abstractmethod
    async def broadcast_transaction(self, tx_hex: str) -> str:
        """Broadcast transaction, returns txid"""

    @abstractmethod
    async def get_transaction(self, txid: str) -> Transaction | None:
        """Get transaction by txid"""

    @abstractmethod
    async def estimate_fee(self, target_blocks: int) -> float:
        """Estimate fee in sat/vbyte for target confirmation blocks.

        Returns:
            Fee rate in sat/vB. Can be fractional (e.g., 0.5 sat/vB).
        """

    async def get_mempool_min_fee(self) -> float | None:
        """Get the minimum fee rate (in sat/vB) for transaction to be accepted into mempool.

        This is used as a floor for fee estimation to ensure transactions are
        relayed and accepted into the mempool. Returns None if not supported
        or unavailable (e.g., light clients).

        Returns:
            Minimum fee rate in sat/vB, or None if unavailable.
        """
        return None

    def can_estimate_fee(self) -> bool:
        """Check if this backend can perform fee estimation.

        Full node backends (Bitcoin Core) can estimate fees.
        Light client backends (Neutrino) typically cannot.

        Returns:
            True if backend supports fee estimation, False otherwise.
        """
        return True

    def has_mempool_access(self) -> bool:
        """Check if this backend can access unconfirmed transactions in the mempool.

        Full node backends (Bitcoin Core) and mempool API backends have
        mempool access and can verify transactions immediately after broadcast.

        Light client backends (Neutrino using BIP157/158) cannot access the mempool
        and can only see transactions after they're confirmed in a block. This
        affects broadcast verification strategy - see BroadcastPolicy docs.

        Returns:
            True if backend can see unconfirmed transactions, False otherwise.
        """
        return True

    @abstractmethod
    async def get_block_height(self) -> int:
        """Get current blockchain height"""

    @abstractmethod
    async def get_block_time(self, block_height: int) -> int:
        """Get block time (unix timestamp) for given height"""

    @abstractmethod
    async def get_block_hash(self, block_height: int) -> str:
        """Get block hash for given height"""

    @abstractmethod
    async def get_utxo(self, txid: str, vout: int) -> UTXO | None:
        """Get a specific UTXO from the blockchain UTXO set (gettxout).
        Returns None if the UTXO does not exist or has been spent."""

    async def scan_descriptors(
        self, descriptors: Sequence[str | dict[str, Any]]
    ) -> dict[str, Any] | None:
        """
        Scan the UTXO set using output descriptors.

        This is an efficient alternative to scanning individual addresses,
        especially useful for HD wallets where xpub descriptors with ranges
        can scan thousands of addresses in a single UTXO set pass.

        Example descriptors:
            - "addr(bc1q...)" - single address
            - "wpkh(xpub.../0/*)" - HD wallet addresses (default range 0-1000)
            - {"desc": "wpkh(xpub.../0/*)", "range": [0, 999]} - explicit range

        Args:
            descriptors: List of output descriptors (strings or dicts with range)

        Returns:
            Scan result dict with:
                - success: bool
                - unspents: list of found UTXOs
                - total_amount: sum of all found UTXOs
            Returns None if not supported or on failure.

        Note:
            Not all backends support descriptor scanning. The default implementation
            returns None. Override in backends that support it (e.g., Bitcoin Core).
        """
        # Default: not supported
        return None

    async def verify_utxo_with_metadata(
        self,
        txid: str,
        vout: int,
        scriptpubkey: str,
        blockheight: int,
    ) -> UTXOVerificationResult:
        """
        Verify a UTXO using provided metadata (neutrino_compat feature).

        This method allows light clients to verify UTXOs without needing
        arbitrary blockchain queries by using metadata provided by the peer.

        The implementation should:
        1. Use scriptpubkey to add the UTXO to watch list (for Neutrino)
        2. Use blockheight as a hint for efficient rescan
        3. Verify the UTXO exists with matching scriptpubkey
        4. Return the UTXO value and confirmations

        Default implementation falls back to get_utxo() for full node backends.

        Args:
            txid: Transaction ID
            vout: Output index
            scriptpubkey: Expected scriptPubKey (hex)
            blockheight: Block height where UTXO was confirmed

        Returns:
            UTXOVerificationResult with verification status and UTXO data
        """
        # Default implementation for full node backends
        # Just uses get_utxo() directly since we can query any UTXO
        utxo = await self.get_utxo(txid, vout)

        if utxo is None:
            return UTXOVerificationResult(
                valid=False,
                error="UTXO not found or spent",
            )

        # Verify scriptpubkey matches
        scriptpubkey_matches = utxo.scriptpubkey.lower() == scriptpubkey.lower()

        if not scriptpubkey_matches:
            return UTXOVerificationResult(
                valid=False,
                value=utxo.value,
                confirmations=utxo.confirmations,
                error="ScriptPubKey mismatch",
                scriptpubkey_matches=False,
            )

        return UTXOVerificationResult(
            valid=True,
            value=utxo.value,
            confirmations=utxo.confirmations,
            scriptpubkey_matches=True,
        )

    def requires_neutrino_metadata(self) -> bool:
        """
        Check if this backend requires Neutrino-compatible metadata for UTXO verification.

        Full node backends can verify any UTXO directly.
        Light client backends need scriptpubkey and blockheight hints.

        Returns:
            True if backend requires metadata for verification
        """
        return False

    def can_provide_neutrino_metadata(self) -> bool:
        """
        Check if this backend can provide Neutrino-compatible metadata for its own UTXOs.

        This determines whether to advertise neutrino_compat feature to the network.
        Backends should return True if they can provide extended UTXO format with
        scriptpubkey and blockheight fields for their own wallet UTXOs in !ioauth.

        This is distinct from requires_neutrino_metadata(): a backend may both
        require metadata from counterparties (to verify their UTXOs) AND provide
        metadata about its own UTXOs. Both full node and neutrino backends can
        provide their own UTXO metadata since they know their wallet's scriptpubkeys
        and block heights.

        Returns:
            True if backend can provide scriptpubkey and blockheight for its UTXOs
        """
        # Default: all backends can provide metadata for their own UTXOs
        return True

    async def verify_tx_output(
        self,
        txid: str,
        vout: int,
        address: str,
        start_height: int | None = None,
    ) -> bool:
        """
        Verify that a specific transaction output exists (was broadcast and confirmed).

        This is useful for verifying a transaction was successfully broadcast when
        we know at least one of its output addresses (e.g., our coinjoin destination).

        For full node backends, this uses get_transaction().
        For light clients (neutrino), this uses UTXO lookup with the address hint.

        Args:
            txid: Transaction ID to verify
            vout: Output index to check
            address: The address that should own this output
            start_height: Optional block height hint for light clients (improves performance)

        Returns:
            True if the output exists (transaction was broadcast), False otherwise
        """
        # Default implementation for full node backends
        tx = await self.get_transaction(txid)
        return tx is not None

    async def verify_bonds(
        self,
        bonds: list[BondVerificationRequest],
    ) -> list[BondVerificationResult]:
        """Verify multiple fidelity bond UTXOs in bulk.

        This is the primary method for verifying fidelity bonds. Each backend can
        override this for optimal performance:
        - Bitcoin Core: JSON-RPC batch of gettxout calls (2 HTTP requests total)
        - Neutrino: batch address rescan + individual UTXO lookups
        - Mempool: parallel HTTP requests

        The default implementation calls get_utxo() sequentially with a semaphore.

        Args:
            bonds: List of bond verification requests with pre-computed addresses

        Returns:
            List of verification results, one per input bond (same order)
        """
        if not bonds:
            return []

        current_height = await self.get_block_height()
        semaphore = asyncio.Semaphore(10)

        async def _verify_one(bond: BondVerificationRequest) -> BondVerificationResult:
            async with semaphore:
                try:
                    utxo = await self.get_utxo(bond.txid, bond.vout)
                    if utxo is None:
                        return BondVerificationResult(
                            txid=bond.txid,
                            vout=bond.vout,
                            value=0,
                            confirmations=0,
                            block_time=0,
                            valid=False,
                            error="UTXO not found or spent",
                        )
                    if utxo.confirmations <= 0:
                        return BondVerificationResult(
                            txid=bond.txid,
                            vout=bond.vout,
                            value=utxo.value,
                            confirmations=0,
                            block_time=0,
                            valid=False,
                            error="UTXO unconfirmed",
                        )
                    # Get the block time for the confirmation block
                    conf_height = current_height - utxo.confirmations + 1
                    block_time = await self.get_block_time(conf_height)
                    return BondVerificationResult(
                        txid=bond.txid,
                        vout=bond.vout,
                        value=utxo.value,
                        confirmations=utxo.confirmations,
                        block_time=block_time,
                        valid=True,
                    )
                except Exception as e:
                    logger.warning(
                        "Bond verification failed for %s:%d: %s",
                        bond.txid,
                        bond.vout,
                        e,
                    )
                    return BondVerificationResult(
                        txid=bond.txid,
                        vout=bond.vout,
                        value=0,
                        confirmations=0,
                        block_time=0,
                        valid=False,
                        error=str(e),
                    )

        results = await asyncio.gather(*[_verify_one(b) for b in bonds])
        return list(results)

    async def close(self) -> None:
        """Close backend connection"""
        pass

Attributes

`supports_descriptor_scan: bool = False` `class-attribute` `instance-attribute`

Whether this backend supports efficient descriptor-based UTXO scanning.

Backends that override scan_descriptors() with a real implementation (e.g. Bitcoin Core's scantxoutset) should set this to True. Light-client backends (Neutrino) leave it at the default False so that sync_all() does not attempt descriptor scanning and fall back with a confusing warning.

`supports_watch_address: bool = False` `class-attribute` `instance-attribute`

Whether this backend requires addresses to be pre-registered via add_watch_address().

Light-client backends (Neutrino) must be told which addresses to watch before a rescan. Full-node backends (Bitcoin Core, descriptor wallet) can query any address on demand and do not need pre-registration. Set to True only in backends that implement add_watch_address().

Functions

`add_watch_address(address: str) -> None` `async`

Register an address for watching.

Only meaningful for backends with supports_watch_address = True (i.e. light-client backends that need explicit address registration before a rescan). Full-node backends can ignore this.

Source code in jmwallet/src/jmwallet/backends/base.py

async def add_watch_address(self, address: str) -> None:
    """Register an address for watching.

    Only meaningful for backends with ``supports_watch_address = True``
    (i.e. light-client backends that need explicit address registration
    before a rescan).  Full-node backends can ignore this.
    """

`broadcast_transaction(tx_hex: str) -> str` `abstractmethod` `async`

Broadcast transaction, returns txid

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def broadcast_transaction(self, tx_hex: str) -> str:
    """Broadcast transaction, returns txid"""

`can_estimate_fee() -> bool`

Check if this backend can perform fee estimation.

Full node backends (Bitcoin Core) can estimate fees. Light client backends (Neutrino) typically cannot.

Returns: True if backend supports fee estimation, False otherwise.

Source code in jmwallet/src/jmwallet/backends/base.py

def can_estimate_fee(self) -> bool:
    """Check if this backend can perform fee estimation.

    Full node backends (Bitcoin Core) can estimate fees.
    Light client backends (Neutrino) typically cannot.

    Returns:
        True if backend supports fee estimation, False otherwise.
    """
    return True

`can_provide_neutrino_metadata() -> bool`

Check if this backend can provide Neutrino-compatible metadata for its own UTXOs.

This determines whether to advertise neutrino_compat feature to the network. Backends should return True if they can provide extended UTXO format with scriptpubkey and blockheight fields for their own wallet UTXOs in !ioauth.

This is distinct from requires_neutrino_metadata(): a backend may both require metadata from counterparties (to verify their UTXOs) AND provide metadata about its own UTXOs. Both full node and neutrino backends can provide their own UTXO metadata since they know their wallet's scriptpubkeys and block heights.

Returns: True if backend can provide scriptpubkey and blockheight for its UTXOs

Source code in jmwallet/src/jmwallet/backends/base.py

def can_provide_neutrino_metadata(self) -> bool:
    """
    Check if this backend can provide Neutrino-compatible metadata for its own UTXOs.

    This determines whether to advertise neutrino_compat feature to the network.
    Backends should return True if they can provide extended UTXO format with
    scriptpubkey and blockheight fields for their own wallet UTXOs in !ioauth.

    This is distinct from requires_neutrino_metadata(): a backend may both
    require metadata from counterparties (to verify their UTXOs) AND provide
    metadata about its own UTXOs. Both full node and neutrino backends can
    provide their own UTXO metadata since they know their wallet's scriptpubkeys
    and block heights.

    Returns:
        True if backend can provide scriptpubkey and blockheight for its UTXOs
    """
    # Default: all backends can provide metadata for their own UTXOs
    return True

`close() -> None` `async`

Close backend connection

Source code in jmwallet/src/jmwallet/backends/base.py

async def close(self) -> None:
    """Close backend connection"""
    pass

`estimate_fee(target_blocks: int) -> float` `abstractmethod` `async`

Estimate fee in sat/vbyte for target confirmation blocks.

Returns: Fee rate in sat/vB. Can be fractional (e.g., 0.5 sat/vB).

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def estimate_fee(self, target_blocks: int) -> float:
    """Estimate fee in sat/vbyte for target confirmation blocks.

    Returns:
        Fee rate in sat/vB. Can be fractional (e.g., 0.5 sat/vB).
    """

`get_address_balance(address: str) -> int` `abstractmethod` `async`

Get balance for an address in satoshis

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_address_balance(self, address: str) -> int:
    """Get balance for an address in satoshis"""

`get_block_hash(block_height: int) -> str` `abstractmethod` `async`

Get block hash for given height

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_block_hash(self, block_height: int) -> str:
    """Get block hash for given height"""

`get_block_height() -> int` `abstractmethod` `async`

Get current blockchain height

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_block_height(self) -> int:
    """Get current blockchain height"""

`get_block_time(block_height: int) -> int` `abstractmethod` `async`

Get block time (unix timestamp) for given height

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_block_time(self, block_height: int) -> int:
    """Get block time (unix timestamp) for given height"""

`get_mempool_min_fee() -> float | None` `async`

Get the minimum fee rate (in sat/vB) for transaction to be accepted into mempool.

This is used as a floor for fee estimation to ensure transactions are relayed and accepted into the mempool. Returns None if not supported or unavailable (e.g., light clients).

Returns: Minimum fee rate in sat/vB, or None if unavailable.

Source code in jmwallet/src/jmwallet/backends/base.py

async def get_mempool_min_fee(self) -> float | None:
    """Get the minimum fee rate (in sat/vB) for transaction to be accepted into mempool.

    This is used as a floor for fee estimation to ensure transactions are
    relayed and accepted into the mempool. Returns None if not supported
    or unavailable (e.g., light clients).

    Returns:
        Minimum fee rate in sat/vB, or None if unavailable.
    """
    return None

`get_transaction(txid: str) -> Transaction | None` `abstractmethod` `async`

Get transaction by txid

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_transaction(self, txid: str) -> Transaction | None:
    """Get transaction by txid"""

`get_utxo(txid: str, vout: int) -> UTXO | None` `abstractmethod` `async`

Get a specific UTXO from the blockchain UTXO set (gettxout). Returns None if the UTXO does not exist or has been spent.

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_utxo(self, txid: str, vout: int) -> UTXO | None:
    """Get a specific UTXO from the blockchain UTXO set (gettxout).
    Returns None if the UTXO does not exist or has been spent."""

`get_utxos(addresses: list[str]) -> list[UTXO]` `abstractmethod` `async`

Get UTXOs for given addresses

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_utxos(self, addresses: list[str]) -> list[UTXO]:
    """Get UTXOs for given addresses"""

`has_mempool_access() -> bool`

Check if this backend can access unconfirmed transactions in the mempool.

Full node backends (Bitcoin Core) and mempool API backends have mempool access and can verify transactions immediately after broadcast.

Light client backends (Neutrino using BIP157/158) cannot access the mempool and can only see transactions after they're confirmed in a block. This affects broadcast verification strategy - see BroadcastPolicy docs.

Returns: True if backend can see unconfirmed transactions, False otherwise.

Source code in jmwallet/src/jmwallet/backends/base.py

def has_mempool_access(self) -> bool:
    """Check if this backend can access unconfirmed transactions in the mempool.

    Full node backends (Bitcoin Core) and mempool API backends have
    mempool access and can verify transactions immediately after broadcast.

    Light client backends (Neutrino using BIP157/158) cannot access the mempool
    and can only see transactions after they're confirmed in a block. This
    affects broadcast verification strategy - see BroadcastPolicy docs.

    Returns:
        True if backend can see unconfirmed transactions, False otherwise.
    """
    return True

`requires_neutrino_metadata() -> bool`

Check if this backend requires Neutrino-compatible metadata for UTXO verification.

Full node backends can verify any UTXO directly. Light client backends need scriptpubkey and blockheight hints.

Returns: True if backend requires metadata for verification

Source code in jmwallet/src/jmwallet/backends/base.py

def requires_neutrino_metadata(self) -> bool:
    """
    Check if this backend requires Neutrino-compatible metadata for UTXO verification.

    Full node backends can verify any UTXO directly.
    Light client backends need scriptpubkey and blockheight hints.

    Returns:
        True if backend requires metadata for verification
    """
    return False

`scan_descriptors(descriptors: Sequence[str | dict[str, Any]]) -> dict[str, Any] | None` `async`

Scan the UTXO set using output descriptors.

This is an efficient alternative to scanning individual addresses, especially useful for HD wallets where xpub descriptors with ranges can scan thousands of addresses in a single UTXO set pass.

Example descriptors: - "addr(bc1q...)" - single address - "wpkh(xpub.../0/)" - HD wallet addresses (default range 0-1000) - {"desc": "wpkh(xpub.../0/)", "range": [0, 999]} - explicit range

Args: descriptors: List of output descriptors (strings or dicts with range)

Returns: Scan result dict with: - success: bool - unspents: list of found UTXOs - total_amount: sum of all found UTXOs Returns None if not supported or on failure.

Note: Not all backends support descriptor scanning. The default implementation returns None. Override in backends that support it (e.g., Bitcoin Core).

Source code in jmwallet/src/jmwallet/backends/base.py

async def scan_descriptors(
    self, descriptors: Sequence[str | dict[str, Any]]
) -> dict[str, Any] | None:
    """
    Scan the UTXO set using output descriptors.

    This is an efficient alternative to scanning individual addresses,
    especially useful for HD wallets where xpub descriptors with ranges
    can scan thousands of addresses in a single UTXO set pass.

    Example descriptors:
        - "addr(bc1q...)" - single address
        - "wpkh(xpub.../0/*)" - HD wallet addresses (default range 0-1000)
        - {"desc": "wpkh(xpub.../0/*)", "range": [0, 999]} - explicit range

    Args:
        descriptors: List of output descriptors (strings or dicts with range)

    Returns:
        Scan result dict with:
            - success: bool
            - unspents: list of found UTXOs
            - total_amount: sum of all found UTXOs
        Returns None if not supported or on failure.

    Note:
        Not all backends support descriptor scanning. The default implementation
        returns None. Override in backends that support it (e.g., Bitcoin Core).
    """
    # Default: not supported
    return None

`set_wallet_creation_height(height: int | None) -> None`

Provide the block height at which the wallet was created.

Backends can use this as a hint to skip scanning blocks before the wallet existed, avoiding unnecessary work during initial sync. Only used when no explicit scan_start_height is configured.

Passing None clears any previously set hint.

The default implementation is a no-op; backends that support scan optimisation override this method.

Source code in jmwallet/src/jmwallet/backends/base.py

def set_wallet_creation_height(self, height: int | None) -> None:
    """Provide the block height at which the wallet was created.

    Backends can use this as a hint to skip scanning blocks before
    the wallet existed, avoiding unnecessary work during initial sync.
    Only used when no explicit ``scan_start_height`` is configured.

    Passing ``None`` clears any previously set hint.

    The default implementation is a no-op; backends that support
    scan optimisation override this method.
    """

`verify_bonds(bonds: list[BondVerificationRequest]) -> list[BondVerificationResult]` `async`

Verify multiple fidelity bond UTXOs in bulk.

This is the primary method for verifying fidelity bonds. Each backend can override this for optimal performance: - Bitcoin Core: JSON-RPC batch of gettxout calls (2 HTTP requests total) - Neutrino: batch address rescan + individual UTXO lookups - Mempool: parallel HTTP requests

The default implementation calls get_utxo() sequentially with a semaphore.

Args: bonds: List of bond verification requests with pre-computed addresses

Returns: List of verification results, one per input bond (same order)

Source code in jmwallet/src/jmwallet/backends/base.py

async def verify_bonds(
    self,
    bonds: list[BondVerificationRequest],
) -> list[BondVerificationResult]:
    """Verify multiple fidelity bond UTXOs in bulk.

    This is the primary method for verifying fidelity bonds. Each backend can
    override this for optimal performance:
    - Bitcoin Core: JSON-RPC batch of gettxout calls (2 HTTP requests total)
    - Neutrino: batch address rescan + individual UTXO lookups
    - Mempool: parallel HTTP requests

    The default implementation calls get_utxo() sequentially with a semaphore.

    Args:
        bonds: List of bond verification requests with pre-computed addresses

    Returns:
        List of verification results, one per input bond (same order)
    """
    if not bonds:
        return []

    current_height = await self.get_block_height()
    semaphore = asyncio.Semaphore(10)

    async def _verify_one(bond: BondVerificationRequest) -> BondVerificationResult:
        async with semaphore:
            try:
                utxo = await self.get_utxo(bond.txid, bond.vout)
                if utxo is None:
                    return BondVerificationResult(
                        txid=bond.txid,
                        vout=bond.vout,
                        value=0,
                        confirmations=0,
                        block_time=0,
                        valid=False,
                        error="UTXO not found or spent",
                    )
                if utxo.confirmations <= 0:
                    return BondVerificationResult(
                        txid=bond.txid,
                        vout=bond.vout,
                        value=utxo.value,
                        confirmations=0,
                        block_time=0,
                        valid=False,
                        error="UTXO unconfirmed",
                    )
                # Get the block time for the confirmation block
                conf_height = current_height - utxo.confirmations + 1
                block_time = await self.get_block_time(conf_height)
                return BondVerificationResult(
                    txid=bond.txid,
                    vout=bond.vout,
                    value=utxo.value,
                    confirmations=utxo.confirmations,
                    block_time=block_time,
                    valid=True,
                )
            except Exception as e:
                logger.warning(
                    "Bond verification failed for %s:%d: %s",
                    bond.txid,
                    bond.vout,
                    e,
                )
                return BondVerificationResult(
                    txid=bond.txid,
                    vout=bond.vout,
                    value=0,
                    confirmations=0,
                    block_time=0,
                    valid=False,
                    error=str(e),
                )

    results = await asyncio.gather(*[_verify_one(b) for b in bonds])
    return list(results)

`verify_tx_output(txid: str, vout: int, address: str, start_height: int | None = None) -> bool` `async`

Verify that a specific transaction output exists (was broadcast and confirmed).

This is useful for verifying a transaction was successfully broadcast when we know at least one of its output addresses (e.g., our coinjoin destination).

For full node backends, this uses get_transaction(). For light clients (neutrino), this uses UTXO lookup with the address hint.

Args: txid: Transaction ID to verify vout: Output index to check address: The address that should own this output start_height: Optional block height hint for light clients (improves performance)

Returns: True if the output exists (transaction was broadcast), False otherwise

Source code in jmwallet/src/jmwallet/backends/base.py

async def verify_tx_output(
    self,
    txid: str,
    vout: int,
    address: str,
    start_height: int | None = None,
) -> bool:
    """
    Verify that a specific transaction output exists (was broadcast and confirmed).

    This is useful for verifying a transaction was successfully broadcast when
    we know at least one of its output addresses (e.g., our coinjoin destination).

    For full node backends, this uses get_transaction().
    For light clients (neutrino), this uses UTXO lookup with the address hint.

    Args:
        txid: Transaction ID to verify
        vout: Output index to check
        address: The address that should own this output
        start_height: Optional block height hint for light clients (improves performance)

    Returns:
        True if the output exists (transaction was broadcast), False otherwise
    """
    # Default implementation for full node backends
    tx = await self.get_transaction(txid)
    return tx is not None

`verify_utxo_with_metadata(txid: str, vout: int, scriptpubkey: str, blockheight: int) -> UTXOVerificationResult` `async`

Verify a UTXO using provided metadata (neutrino_compat feature).

This method allows light clients to verify UTXOs without needing arbitrary blockchain queries by using metadata provided by the peer.

The implementation should: 1. Use scriptpubkey to add the UTXO to watch list (for Neutrino) 2. Use blockheight as a hint for efficient rescan 3. Verify the UTXO exists with matching scriptpubkey 4. Return the UTXO value and confirmations

Default implementation falls back to get_utxo() for full node backends.

Args: txid: Transaction ID vout: Output index scriptpubkey: Expected scriptPubKey (hex) blockheight: Block height where UTXO was confirmed

Returns: UTXOVerificationResult with verification status and UTXO data

Source code in jmwallet/src/jmwallet/backends/base.py

async def verify_utxo_with_metadata(
    self,
    txid: str,
    vout: int,
    scriptpubkey: str,
    blockheight: int,
) -> UTXOVerificationResult:
    """
    Verify a UTXO using provided metadata (neutrino_compat feature).

    This method allows light clients to verify UTXOs without needing
    arbitrary blockchain queries by using metadata provided by the peer.

    The implementation should:
    1. Use scriptpubkey to add the UTXO to watch list (for Neutrino)
    2. Use blockheight as a hint for efficient rescan
    3. Verify the UTXO exists with matching scriptpubkey
    4. Return the UTXO value and confirmations

    Default implementation falls back to get_utxo() for full node backends.

    Args:
        txid: Transaction ID
        vout: Output index
        scriptpubkey: Expected scriptPubKey (hex)
        blockheight: Block height where UTXO was confirmed

    Returns:
        UTXOVerificationResult with verification status and UTXO data
    """
    # Default implementation for full node backends
    # Just uses get_utxo() directly since we can query any UTXO
    utxo = await self.get_utxo(txid, vout)

    if utxo is None:
        return UTXOVerificationResult(
            valid=False,
            error="UTXO not found or spent",
        )

    # Verify scriptpubkey matches
    scriptpubkey_matches = utxo.scriptpubkey.lower() == scriptpubkey.lower()

    if not scriptpubkey_matches:
        return UTXOVerificationResult(
            valid=False,
            value=utxo.value,
            confirmations=utxo.confirmations,
            error="ScriptPubKey mismatch",
            scriptpubkey_matches=False,
        )

    return UTXOVerificationResult(
        valid=True,
        value=utxo.value,
        confirmations=utxo.confirmations,
        scriptpubkey_matches=True,
    )

`BondVerificationRequest`

Request to verify a single fidelity bond UTXO.

All fields are derived from the bond proof data. The address and scriptpubkey are pre-computed by the caller using derive_bond_address(utxo_pub, locktime).

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class BondVerificationRequest:
    """Request to verify a single fidelity bond UTXO.

    All fields are derived from the bond proof data. The address and scriptpubkey
    are pre-computed by the caller using ``derive_bond_address(utxo_pub, locktime)``.
    """

    txid: str
    """Transaction ID (hex, big-endian)"""
    vout: int
    """Output index"""
    utxo_pub: bytes
    """33-byte compressed public key from bond proof"""
    locktime: int
    """Locktime from bond proof (Unix timestamp)"""
    address: str
    """Derived P2WSH bech32 address"""
    scriptpubkey: str
    """Derived P2WSH scriptPubKey (hex)"""

Attributes

`address: str` `instance-attribute`

Derived P2WSH bech32 address

`locktime: int` `instance-attribute`

Locktime from bond proof (Unix timestamp)

`scriptpubkey: str` `instance-attribute`

Derived P2WSH scriptPubKey (hex)

`txid: str` `instance-attribute`

Transaction ID (hex, big-endian)

`utxo_pub: bytes` `instance-attribute`

33-byte compressed public key from bond proof

`vout: int` `instance-attribute`

Output index

`BondVerificationResult`

Result of verifying a single fidelity bond UTXO.

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class BondVerificationResult:
    """Result of verifying a single fidelity bond UTXO."""

    txid: str
    """Transaction ID"""
    vout: int
    """Output index"""
    value: int
    """UTXO value in satoshis (0 if verification failed)"""
    confirmations: int
    """Number of confirmations (0 if unconfirmed or failed)"""
    block_time: int
    """Confirmation timestamp (0 if unconfirmed or failed)"""
    valid: bool
    """Whether the bond UTXO exists, is unspent, and has positive confirmations"""
    error: str | None = None
    """Error description if verification failed"""

Attributes

`block_time: int` `instance-attribute`

Confirmation timestamp (0 if unconfirmed or failed)

`confirmations: int` `instance-attribute`

Number of confirmations (0 if unconfirmed or failed)

`error: str | None = None` `class-attribute` `instance-attribute`

Error description if verification failed

`txid: str` `instance-attribute`

Transaction ID

`valid: bool` `instance-attribute`

Whether the bond UTXO exists, is unspent, and has positive confirmations

`value: int` `instance-attribute`

UTXO value in satoshis (0 if verification failed)

`vout: int` `instance-attribute`

Output index

`Transaction`

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class Transaction:
    txid: str
    raw: str
    confirmations: int
    block_height: int | None = None
    block_time: int | None = None

Attributes

`block_height: int | None = None` `class-attribute` `instance-attribute`

`block_time: int | None = None` `class-attribute` `instance-attribute`

`confirmations: int` `instance-attribute`

`raw: str` `instance-attribute`

`txid: str` `instance-attribute`

`UTXO`

Backend-layer UTXO model returned by blockchain backends.

This type intentionally only contains chain-derived fields. Wallet metadata such as freeze state, labels, derivation path, fidelity-bond flags, and locktime are attached later by the wallet service when converting backend UTXOs into wallet-layer UTXOInfo entries.

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class UTXO:
    """Backend-layer UTXO model returned by blockchain backends.

    This type intentionally only contains chain-derived fields. Wallet metadata
    such as freeze state, labels, derivation path, fidelity-bond flags, and
    locktime are attached later by the wallet service when converting backend
    UTXOs into wallet-layer ``UTXOInfo`` entries.
    """

    txid: str
    vout: int
    value: int
    address: str
    confirmations: int
    scriptpubkey: str
    height: int | None = None

Attributes

`address: str` `instance-attribute`

`confirmations: int` `instance-attribute`

`height: int | None = None` `class-attribute` `instance-attribute`

`scriptpubkey: str` `instance-attribute`

`txid: str` `instance-attribute`

`value: int` `instance-attribute`

`vout: int` `instance-attribute`

`UTXOVerificationResult`

Result of UTXO verification with metadata.

Used by neutrino_compat feature for Neutrino-compatible verification.

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class UTXOVerificationResult:
    """
    Result of UTXO verification with metadata.

    Used by neutrino_compat feature for Neutrino-compatible verification.
    """

    valid: bool
    value: int = 0
    confirmations: int = 0
    error: str | None = None
    scriptpubkey_matches: bool = False

jmwallet.backends.base