jmwalletd.routers.coinjoin

jmwalletd.routers.coinjoin

Maker and taker (coinjoin) endpoints.

Attributes

router = APIRouter() module-attribute

Classes

Functions:

build_coinjoin_taker_config(*, body: Any, mnemonic: Any, jm_settings: Any, taker_config_cls: Any, config_overrides: dict[str, dict[str, str]] | None = None, passphrase: str = '') -> Any

Build a TakerConfig for a one-shot do_coinjoin request.

Delegates to :func:taker.config_builder.build_taker_config_kwargs (the same mapping the CLI taker and tumbler use) so a CoinJoin started through the daemon honors every [taker] policy setting (passed via config or TAKER__* env). This endpoint used to hand-maintain a mirror of that mapping, which silently drifted twice (issue #530, then the adaptive orderbook-wait knobs).

In particular minimum_makers is capped against the requested counterparties: a request for fewer makers than the policy minimum_makers (default 4) would otherwise select a valid N-maker CoinJoin and then reject it with Not enough makers selected: N.

config_overrides is the daemon's in-memory configset store; the fee policy JAM writes there ([POLICY] tx_fees etc.) is applied on top of the settings so a sat/vB rate chosen in the UI is honored (issue #566). A txfee on the request itself (the fee picked on JAM's Send page) takes precedence over tx_fees for this CoinJoin only (issue #636).

Source code in jmwalletd/src/jmwalletd/routers/coinjoin.py

def build_coinjoin_taker_config(
    *,
    body: Any,
    mnemonic: Any,
    jm_settings: Any,
    taker_config_cls: Any,
    config_overrides: dict[str, dict[str, str]] | None = None,
    passphrase: str = "",
) -> Any:
    """Build a ``TakerConfig`` for a one-shot ``do_coinjoin`` request.

    Delegates to :func:`taker.config_builder.build_taker_config_kwargs` (the
    same mapping the CLI taker and tumbler use) so a CoinJoin started through
    the daemon honors every ``[taker]`` policy setting (passed via config or
    ``TAKER__*`` env). This endpoint used to hand-maintain a mirror of that
    mapping, which silently drifted twice (issue #530, then the adaptive
    orderbook-wait knobs).

    In particular ``minimum_makers`` is capped against the requested
    ``counterparties``: a request for fewer makers than the policy
    ``minimum_makers`` (default 4) would otherwise select a valid N-maker
    CoinJoin and then reject it with ``Not enough makers selected: N``.

    ``config_overrides`` is the daemon's in-memory ``configset`` store; the
    fee policy JAM writes there (``[POLICY] tx_fees`` etc.) is applied on top
    of the settings so a sat/vB rate chosen in the UI is honored (issue #566).
    A ``txfee`` on the request itself (the fee picked on JAM's Send page)
    takes precedence over ``tx_fees`` for this CoinJoin only (issue #636).
    """
    from jmwalletd.fee_policy import resolve_policy_fee_overrides
    from taker.config_builder import build_taker_config_kwargs

    fee_overrides = resolve_policy_fee_overrides(config_overrides, request_tx_fee=body.txfee)
    kwargs = build_taker_config_kwargs(
        jm_settings,
        mnemonic,
        passphrase,
        amount=body.amount_sats,
        destination=body.destination,
        mixdepth=body.mixdepth,
        counterparties=int(body.counterparties),
        max_abs_fee=fee_overrides.max_cj_fee_abs,
        max_rel_fee=fee_overrides.max_cj_fee_rel,
        max_sweep_fee_change=fee_overrides.max_sweep_fee_change,
        fee_rate=fee_overrides.fee_rate,
        block_target=fee_overrides.block_target,
        tx_fee_factor=fee_overrides.tx_fee_factor,
    )
    return taker_config_cls(**kwargs)

direct_send(walletname: str, body: DirectSendRequest, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> DirectSendResponse async

Send bitcoin directly (without coinjoin).

Source code in jmwalletd/src/jmwalletd/routers/coinjoin.py

@router.post("/wallet/{walletname}/taker/direct-send", operation_id="directsend")
async def direct_send(
    walletname: str,
    body: DirectSendRequest,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> DirectSendResponse:
    """Send bitcoin directly (without coinjoin)."""
    if state.taker_running:
        raise ActionNotAllowed("A coinjoin is already in progress.")

    ws = state.wallet_service

    try:
        from jmwalletd.fee_policy import resolve_policy_fee_overrides
        from jmwalletd.send import do_direct_send

        # Honor the fee policy JAM stores via configset ([POLICY] tx_fees):
        # a manual sat/vB rate or block target set in the UI applies to
        # direct sends too (issue #566). A txfee on the request (the fee
        # picked on JAM's Send page) takes precedence for this send only
        # (issue #636).
        fee_overrides = resolve_policy_fee_overrides(
            state.config_overrides, request_tx_fee=body.txfee
        )
        settings = get_settings()
        fee_target_blocks = fee_overrides.block_target
        if fee_overrides.fee_rate is None and fee_target_blocks is None:
            fee_target_blocks = settings.wallet.default_fee_block_target
        tx_result = await do_direct_send(
            wallet_service=ws,
            mixdepth=body.mixdepth,
            amount_sats=body.amount_sats,
            destination=body.destination,
            fee_rate=fee_overrides.fee_rate,
            fee_target_blocks=fee_target_blocks,
            tx_fee_factor=(
                fee_overrides.tx_fee_factor
                if fee_overrides.tx_fee_factor is not None
                else settings.taker.tx_fee_factor
            ),
            max_fee_rate_sat_vb=settings.wallet.max_fee_rate_sat_vb,
            input_utxos=body.input_utxos,
            rbf=body.rbf,
        )
    except ValueError as exc:
        raise InvalidRequestFormat(str(exc)) from exc
    except Exception as exc:
        logger.error("Direct send failed")
        logger.bind(sensitive=True).exception("Direct send failed")
        raise TransactionFailed(str(exc)) from exc

    # Build the txinfo response.
    txinfo = _build_txinfo(tx_result)

    # Notify WebSocket clients about the transaction immediately, and mark it
    # so the background transaction monitor does not emit a duplicate
    # first-seen notification for the same txid (it still reports confirmation).
    state.mark_tx_broadcast(txinfo.txid)
    state.broadcast_ws({"txid": txinfo.txid, "txdetails": txinfo.model_dump()})

    return DirectSendResponse(txinfo=txinfo)

do_coinjoin(walletname: str, body: DoCoinjoinRequest, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> JSONResponse async

Initiate a coinjoin transaction (asynchronous).

Source code in jmwalletd/src/jmwalletd/routers/coinjoin.py

@router.post("/wallet/{walletname}/taker/coinjoin", status_code=202, operation_id="docoinjoin")
async def do_coinjoin(
    walletname: str,
    body: DoCoinjoinRequest,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> JSONResponse:
    """Initiate a coinjoin transaction (asynchronous)."""
    if state.coinjoin_state != CoinjoinState.NOT_RUNNING:
        raise ServiceAlreadyStarted("A coinjoin or maker service is already running.")
    if not state.wallet_mnemonic:
        raise NoWalletFound("Wallet mnemonic not available in daemon state.")

    try:
        from jmwalletd._backend import get_backend
        from taker.config import TakerConfig
        from taker.models import TakerState
        from taker.taker import Taker

        state.activate_coinjoin_state(CoinjoinState.TAKER_RUNNING)
        state.last_broadcast_policy = None
        state.last_broadcast_method = None
        state.last_broadcast_fallback_reason = None
        state.last_taker_status = None
        state.last_taker_txid = None
        state.last_taker_error = None

        async def _run_coinjoin() -> None:
            taker: Any | None = None
            run_exception: Exception | None = None
            cancelled = False
            try:
                jm_settings = get_settings()
                bitcoin_network = (
                    jm_settings.network_config.bitcoin_network or jm_settings.network_config.network
                )
                backend = await get_backend(
                    state.data_dir,
                    force_new=True,
                    mnemonic=state.wallet_mnemonic,
                    network=bitcoin_network.value,
                    passphrase=state.wallet_bip39_passphrase,
                )
                config = build_coinjoin_taker_config(
                    body=body,
                    mnemonic=state.wallet_mnemonic,
                    jm_settings=jm_settings,
                    taker_config_cls=TakerConfig,
                    config_overrides=state.config_overrides,
                    passphrase=state.wallet_bip39_passphrase,
                )
                taker = Taker(
                    wallet=ws,
                    backend=backend,
                    config=config,
                )
                state._taker_ref = taker
                await taker.start()
                await taker.do_coinjoin(
                    amount=body.amount_sats,
                    destination=body.destination,
                    mixdepth=body.mixdepth,
                    counterparty_count=body.counterparties,
                    input_utxos=body.input_utxos,
                )
            except asyncio.CancelledError:
                # /taker/stop cancels this task rather than raising into it, so
                # this only fires for cancellation, never a plain failure.
                # Record it and re-raise so the awaiting stop endpoint still
                # observes the cancellation as normal.
                cancelled = True
                raise
            except Exception as exc:
                run_exception = exc
                logger.error("Coinjoin failed")
                logger.bind(sensitive=True).exception("Coinjoin failed")
            finally:
                # Always tear down the taker so its directory-client and
                # background tasks do not leak. Keep the shared wallet open
                # for any subsequent operation on the daemon.
                if taker is not None:
                    state.last_broadcast_policy = taker.last_broadcast_policy or None
                    state.last_broadcast_method = taker.last_broadcast_method or None
                    state.last_broadcast_fallback_reason = (
                        taker.last_broadcast_fallback_reason or None
                    )
                    status = taker.state.value if taker.state else None
                    # A raw TakerState like "fetching_orderbook" is only a real
                    # terminal outcome when the taker reached it on its own
                    # (COMPLETE/FAILED). If a stop cancelled the run or an
                    # unhandled exception unwound it mid-phase, that leftover
                    # in-progress phase is not the final word -- normalize it,
                    # unless a broadcast already succeeded (COMPLETE), which
                    # cancellation/an exception during teardown must not erase.
                    if status != TakerState.COMPLETE.value:
                        if cancelled:
                            status = TakerState.CANCELLED.value
                        elif run_exception is not None:
                            status = TakerState.FAILED.value
                    state.last_taker_status = status
                    state.last_taker_txid = taker.txid or None
                    # ``last_failure_reason`` covers the specific failure paths the
                    # taker itself recognizes (e.g. a declined confirmation); fall
                    # back to the exception that unwound this task for anything
                    # else, so a caller polling /taker/status is never left with
                    # a "failed" status and no reason.
                    state.last_taker_error = taker.last_failure_reason or (
                        str(run_exception) if run_exception is not None else None
                    )
                    try:
                        await taker.stop(close_wallet=False)
                    except Exception:
                        logger.error("Taker teardown failed")
                        logger.bind(sensitive=True).exception("Taker teardown failed")
                elif run_exception is not None:
                    # The Taker never got constructed (e.g. config/backend setup
                    # failed) -- still record something explaining why.
                    state.last_taker_status = TakerState.FAILED.value
                    state.last_taker_error = str(run_exception)
                elif cancelled:
                    state.last_taker_status = TakerState.CANCELLED.value
                state.activate_coinjoin_state(CoinjoinState.NOT_RUNNING)
                state._taker_ref = None

        ws = state.wallet_service
        state._taker_task = asyncio.create_task(_run_coinjoin())

    except ImportError:
        state.activate_coinjoin_state(CoinjoinState.NOT_RUNNING)
        raise BackendNotReady("Taker module not available.") from None
    except Exception as exc:
        state.activate_coinjoin_state(CoinjoinState.NOT_RUNNING)
        raise BackendNotReady(str(exc)) from exc

    return JSONResponse(content={}, status_code=202)

start_maker(walletname: str, body: StartMakerRequest, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> JSONResponse async

Start the yield generator (maker) service.

Source code in jmwalletd/src/jmwalletd/routers/coinjoin.py

@router.post("/wallet/{walletname}/maker/start", status_code=202, operation_id="startmaker")
async def start_maker(
    walletname: str,
    body: StartMakerRequest,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> JSONResponse:
    """Start the yield generator (maker) service."""
    if state.coinjoin_state != CoinjoinState.NOT_RUNNING:
        raise ServiceAlreadyStarted("A coinjoin or maker service is already running.")
    if not state.wallet_mnemonic:
        raise NoWalletFound("Wallet mnemonic not available in daemon state.")

    # Parse maker parameters.
    try:
        txfee = int(body.txfee)
        cjfee_a = int(body.cjfee_a)
        cjfee_r = str(body.cjfee_r)
        minsize = int(body.minsize)
    except ValueError as exc:
        raise InvalidRequestFormat(f"Invalid maker parameter: {exc}") from exc

    try:
        from jmwalletd._backend import get_backend
        from jmwalletd.maker_config import build_daemon_maker_config
        from maker.bot import MakerBot

        state.activate_coinjoin_state(CoinjoinState.MAKER_RUNNING)

        async def _run_maker() -> None:
            maker: MakerBot | None = None
            backend: Any | None = None
            try:
                ws = state.wallet_service
                backend = await get_backend(
                    state.data_dir,
                    force_new=True,
                    wallet_service=ws,
                )
                jm_settings = get_settings()
                config = build_daemon_maker_config(
                    jm_settings,
                    state.wallet_mnemonic,
                    state.data_dir,
                    offer_type=body.ordertype,
                    min_size=minsize,
                    cj_fee_relative=cjfee_r,
                    cj_fee_absolute=cjfee_a,
                    tx_fee_contribution=txfee,
                    passphrase=state.wallet_bip39_passphrase,
                )

                def _publish_maker_nick(_old_nick: str, new_nick: str) -> None:
                    state.nickname = new_nick
                    write_nick_state(state.data_dir, "maker", new_nick)

                maker = MakerBot(
                    wallet=ws,
                    backend=backend,
                    config=config,
                    nick_change_callback=_publish_maker_nick,
                )
                state._maker_ref = maker
                state.nickname = maker.nick
                write_nick_state(state.data_dir, "maker", maker.nick)

                await maker.start()
                # NOTE: maker.start() blocks until shutdown (it awaits
                # asyncio.gather on listen tasks).  The session endpoint
                # now reads current_offers directly from the maker ref,
                # so there is nothing to do here.
            except Exception:
                logger.error("Maker failed")
                logger.bind(sensitive=True).exception("Maker failed")
            finally:
                if maker is not None:
                    try:
                        await maker.stop()
                    except Exception:
                        logger.error("Error stopping maker after its run ended")
                        logger.bind(sensitive=True).exception(
                            "Error stopping maker after its run ended"
                        )
                if backend is not None:
                    try:
                        await backend.close()
                    except Exception:
                        logger.error("Error closing maker blockchain backend")
                        logger.bind(sensitive=True).exception(
                            "Error closing maker blockchain backend"
                        )
                state.activate_coinjoin_state(CoinjoinState.NOT_RUNNING)
                state.offer_list = None
                state.nickname = None
                state._maker_ref = None
                state._maker_task = None
                remove_nick_state(state.data_dir, "maker")

        state._maker_task = asyncio.create_task(_run_maker())
    except ImportError:
        state.activate_coinjoin_state(CoinjoinState.NOT_RUNNING)
        raise BackendNotReady("Maker module not available.") from None
    except Exception as exc:
        state.activate_coinjoin_state(CoinjoinState.NOT_RUNNING)
        raise BackendNotReady(str(exc)) from exc

    return JSONResponse(content={}, status_code=202)

stop_coinjoin(walletname: str, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> JSONResponse async

Stop a running coinjoin/tumbler.

Source code in jmwalletd/src/jmwalletd/routers/coinjoin.py

@router.get("/wallet/{walletname}/taker/stop", status_code=202, operation_id="stopcoinjoin")
async def stop_coinjoin(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> JSONResponse:
    """Stop a running coinjoin/tumbler."""
    if not state.taker_running:
        raise ServiceNotStarted()

    # Signal the taker to stop if a reference is held.
    if state._taker_ref is not None:
        try:
            await state._taker_ref.stop()
        except Exception:
            logger.error("Error stopping taker")
            logger.bind(sensitive=True).exception("Error stopping taker")

    if state._taker_task is not None and not state._taker_task.done():
        state._taker_task.cancel()
        with contextlib.suppress(asyncio.CancelledError, Exception):
            await state._taker_task

    state.activate_coinjoin_state(CoinjoinState.NOT_RUNNING)
    state._taker_ref = None
    state._taker_task = None
    return JSONResponse(content={}, status_code=202)

stop_maker(walletname: str, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> JSONResponse async

Stop the yield generator (maker) service.

Source code in jmwalletd/src/jmwalletd/routers/coinjoin.py

@router.get("/wallet/{walletname}/maker/stop", status_code=202, operation_id="stopmaker")
async def stop_maker(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> JSONResponse:
    """Stop the yield generator (maker) service."""
    if not state.maker_running:
        raise ServiceNotStarted()

    # Signal the maker to stop if a reference is held.
    if state._maker_ref is not None:
        try:
            await state._maker_ref.stop()
        except Exception:
            logger.error("Error stopping maker")
            logger.bind(sensitive=True).exception("Error stopping maker")

    if state._maker_task is not None and not state._maker_task.done():
        state._maker_task.cancel()
        with contextlib.suppress(asyncio.CancelledError, Exception):
            await state._maker_task

    state.activate_coinjoin_state(CoinjoinState.NOT_RUNNING)
    state.offer_list = None
    state.nickname = None
    state._maker_ref = None
    state._maker_task = None
    return JSONResponse(content={}, status_code=202)

taker_status(walletname: str, _auth: dict[str, Any] = Depends(require_auth), _wallet: None = Depends(require_wallet_match), state: DaemonState = Depends(get_daemon_state)) -> TakerStatusResponse async

Report the outcome of the most recent single-shot taker/coinjoin call.

While a run is in progress, reads live off the Taker instance so callers can poll for phase changes; once it tears down, falls back to the snapshot _run_coinjoin took right before doing so. Returns an "empty" response (status and txid both None) if no taker run has happened yet this session -- that is not itself evidence of anything.

running reflects the single-shot taker/coinjoin state specifically (not state.taker_running, which is also true while a tumbler plan is driving its own takers internally): a tumble in progress must not be reported as "running" here while showing a stale snapshot from a previous single-shot call underneath it.

Source code in jmwalletd/src/jmwalletd/routers/coinjoin.py

@router.get("/wallet/{walletname}/taker/status", operation_id="takerstatus")
async def taker_status(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> TakerStatusResponse:
    """Report the outcome of the most recent single-shot taker/coinjoin call.

    While a run is in progress, reads live off the ``Taker`` instance so
    callers can poll for phase changes; once it tears down, falls back to the
    snapshot ``_run_coinjoin`` took right before doing so. Returns an "empty"
    response (``status`` and ``txid`` both ``None``) if no taker run has
    happened yet this session -- that is not itself evidence of anything.

    ``running`` reflects the single-shot ``taker/coinjoin`` state specifically
    (not ``state.taker_running``, which is also true while a tumbler plan is
    driving its own takers internally): a tumble in progress must not be
    reported as "running" here while showing a stale snapshot from a previous
    single-shot call underneath it.
    """
    running = state.coinjoin_state == CoinjoinState.TAKER_RUNNING
    taker = state._taker_ref
    if taker is not None:
        return TakerStatusResponse(
            running=running,
            status=taker.state.value if taker.state else None,
            txid=taker.txid or None,
            error=taker.last_failure_reason or None,
            broadcast_method=taker.last_broadcast_method or None,
        )

    return TakerStatusResponse(
        running=running,
        status=state.last_taker_status,
        txid=state.last_taker_txid,
        error=state.last_taker_error,
        broadcast_method=state.last_broadcast_method,
    )