jmwalletd.routers.tumbler

jmwalletd.routers.tumbler

Tumbler endpoints backed by :mod:tumbler.

The router exposes a small, stateless-ish HTTP surface over the persistent YAML plan managed by :mod:tumbler.persistence and the in-memory runner owned by :class:jmwalletd.state.DaemonState:

• POST /tumbler/plan -- build a new plan and persist it as PENDING.
• POST /tumbler/start -- run the pending plan; the runner updates the plan in place and the daemon keeps a handle on the task.
• GET /tumbler/status -- fetch the current plan (in-memory if running, otherwise from disk). Flags a stale plan whose on-disk status is RUNNING but no runner is live (crash recovery marker).
• POST /tumbler/stop -- cooperatively request the runner to stop; the task transitions the plan to CANCELLED and tears down its taker / maker.
• DELETE /tumbler/plan -- remove a terminal or pending plan. Refuses when the runner is live -- stop first.

See docs/technical/tumbler-redesign.md for the state matrix and subset-sum rationale. The router itself is intentionally thin so all plan semantics stay in :mod:tumbler.

Attributes

_ = BackendNotReady module-attribute

router = APIRouter() module-attribute

Classes

Functions:

build_tumbler_taker_config(*, phase: Any, mnemonic: Any, jm_settings: Any, taker_config_cls: Any) -> Any

Build a TakerConfig for a tumbler taker phase.

Delegates to :func:taker.config_builder.build_taker_config_kwargs (the same mapping the CLI taker and standalone tumbler use) so daemon-run tumbler phases honor every [taker] policy setting. This factory used to set only the network/Tor/directory fields, so fee limits, timeouts, and the orderbook-wait knobs silently fell back to TakerConfig defaults for tumbles started through the API.

minimum_makers is capped at the phase's counterparty_count so a sweep that legitimately selects N makers is not rejected against a higher policy threshold (default 4), which failed phases with Not enough makers for sweep: N.

destination is resolved inside the runner (INTERNAL sentinel), so an empty placeholder is passed here; the Taker reads it only when do_coinjoin is not given one.

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

def build_tumbler_taker_config(
    *,
    phase: Any,
    mnemonic: Any,
    jm_settings: Any,
    taker_config_cls: Any,
) -> Any:
    """Build a ``TakerConfig`` for a tumbler taker phase.

    Delegates to :func:`taker.config_builder.build_taker_config_kwargs` (the
    same mapping the CLI taker and standalone tumbler use) so daemon-run
    tumbler phases honor every ``[taker]`` policy setting. This factory used
    to set only the network/Tor/directory fields, so fee limits, timeouts,
    and the orderbook-wait knobs silently fell back to ``TakerConfig``
    defaults for tumbles started through the API.

    ``minimum_makers`` is capped at the phase's ``counterparty_count`` so a
    sweep that legitimately selects N makers is not rejected against a
    higher policy threshold (default 4), which failed phases with
    ``Not enough makers for sweep: N``.

    ``destination`` is resolved inside the runner (INTERNAL sentinel), so an
    empty placeholder is passed here; the Taker reads it only when
    ``do_coinjoin`` is not given one.
    """
    from taker.config_builder import build_taker_config_kwargs

    kwargs = build_taker_config_kwargs(
        jm_settings,
        mnemonic,
        "",
        amount=getattr(phase, "amount", 0) or 0,
        destination="",
        mixdepth=getattr(phase, "mixdepth", 0),
        counterparties=int(getattr(phase, "counterparty_count", 1) or 1),
    )
    return taker_config_cls(**kwargs)

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

Build and persist a fresh tumble plan for the active wallet.

An already-running plan for the wallet is always protected: callers must POST /tumbler/stop first. A plan in any other state (pending, completed, failed, cancelled) is overwritten unconditionally -- passing force=true is only required for a pending plan, to make the destructive intent explicit.

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

@router.post("/wallet/{walletname}/tumbler/plan", status_code=201, operation_id="tumblerplan")
async def create_plan(
    walletname: str,
    body: TumblerPlanRequest,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> TumblerPlanResponse:
    """Build and persist a fresh tumble plan for the active wallet.

    An already-running plan for the wallet is always protected: callers must
    ``POST /tumbler/stop`` first. A plan in any other state (pending,
    completed, failed, cancelled) is overwritten unconditionally -- passing
    ``force=true`` is only required for a pending plan, to make the
    destructive intent explicit.
    """
    if _runner_alive_for(state, state.wallet_name):
        raise ServiceAlreadyStarted("A tumbler is already running; stop it first.")

    existing = _reconcile_on_request(state, state.wallet_name)
    if existing is not None and existing.status == PlanStatus.PENDING and not body.force:
        raise ActionNotAllowed("A pending plan already exists; pass force=true to overwrite it.")

    ws = state.wallet_service
    if ws is None:
        raise NoWalletFound()

    balances = await _mixdepth_balances(ws, num_mixdepths=getattr(ws, "mixdepth_count", 5))
    if not any(v > 0 for v in balances.values()):
        raise ActionNotAllowed("Wallet has no confirmed coins to tumble.")

    extra = _normalize_legacy_tumbler_parameters(body.parameters)
    try:
        params = TumbleParameters(
            destinations=list(body.destinations),
            mixdepth_balances=balances,
            **extra,  # type: ignore[arg-type]
        )
    except (TypeError, ValueError) as exc:
        raise InvalidRequestFormat(f"Invalid tumbler parameters: {exc}") from exc

    try:
        plan = PlanBuilder(wallet_name=state.wallet_name, params=params).build()
    except ValueError as exc:
        raise InvalidRequestFormat(str(exc)) from exc

    save_plan(plan, state.data_dir)
    state.broadcast_ws({"tumbler": {"event": "plan_created", "wallet_name": plan.wallet_name}})
    logger.info(
        "tumbler plan created: wallet={} phases={} destinations={}",
        plan.wallet_name,
        len(plan.phases),
        len(plan.destinations),
    )
    return _plan_to_response(plan)

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

Remove a non-running plan from disk.

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

@router.delete(
    "/wallet/{walletname}/tumbler/plan", status_code=204, operation_id="tumblerplandelete"
)
async def delete_plan_endpoint(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> JSONResponse:
    """Remove a non-running plan from disk."""
    if _runner_alive_for(state, state.wallet_name):
        raise ActionNotAllowed("A tumbler is running; stop it before deleting the plan.")

    # Reconcile first so a stale ``RUNNING`` plan on disk is flipped to FAILED
    # before deletion; keeps the observable event stream consistent.
    _reconcile_on_request(state, state.wallet_name)

    removed = delete_plan(state.wallet_name, state.data_dir)
    if not removed:
        raise NoWalletFound("No tumbler plan exists for this wallet.")
    state.broadcast_ws({"tumbler": {"event": "plan_deleted", "wallet_name": state.wallet_name}})
    return JSONResponse(content=None, status_code=204)

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

Return the live plan if the runner is active, otherwise the on-disk plan.

When the on-disk plan is RUNNING but no runner is live, the response's stale flag is set so the UI can prompt the user to acknowledge the failure and delete the plan.

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

@router.get("/wallet/{walletname}/tumbler/status", operation_id="tumblerstatus")
async def get_status(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> TumblerPlanResponse:
    """Return the live plan if the runner is active, otherwise the on-disk plan.

    When the on-disk plan is ``RUNNING`` but no runner is live, the response's
    ``stale`` flag is set so the UI can prompt the user to acknowledge the
    failure and delete the plan.
    """
    if _runner_alive_for(state, state.wallet_name):
        # ``tumble_runner`` is the authoritative state while running.
        return _plan_to_response(state.tumble_runner.plan)

    try:
        plan = load_plan(state.wallet_name, state.data_dir)
    except PlanNotFoundError as exc:
        raise NoWalletFound("No tumbler plan exists for this wallet.") from exc
    except PlanCorruptError as exc:
        raise ActionNotAllowed(f"Tumbler plan is corrupt: {exc}") from exc

    stale = plan.status == PlanStatus.RUNNING
    if stale:
        # Best-effort reconcile so successive calls do not keep flagging.
        plan.status = PlanStatus.FAILED
        plan.error = plan.error or "daemon restarted mid-run"
        save_plan(plan, state.data_dir)
    return _plan_to_response(plan, stale=stale)

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

Load the pending plan and run it in the background.

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

@router.post("/wallet/{walletname}/tumbler/start", status_code=202, operation_id="tumblerstart")
async def start_plan(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> JSONResponse:
    """Load the pending plan and run it in the background."""
    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.")

    plan = _reconcile_on_request(state, state.wallet_name)
    if plan is None:
        raise NoWalletFound("No tumbler plan exists for this wallet; create one first.")
    if plan.status in (PlanStatus.COMPLETED, PlanStatus.FAILED, PlanStatus.CANCELLED):
        raise ActionNotAllowed(f"Plan is in terminal state {plan.status.value}; create a new plan.")
    if plan.status == PlanStatus.RUNNING:
        raise ServiceAlreadyStarted("Plan is already running.")

    ws = state.wallet_service
    if ws is None:
        raise NoWalletFound()

    # Factories are closed over the current wallet/settings at start time.
    jm_settings = get_settings()

    from jmwalletd._backend import get_backend
    from maker.bot import MakerBot
    from maker.config import MakerConfig
    from taker.config import TakerConfig
    from taker.taker import Taker

    async def _taker_factory(phase: Any) -> Any:
        backend = await get_backend(
            state.data_dir,
            force_new=True,
            wallet_service=ws,
        )
        config = build_tumbler_taker_config(
            phase=phase,
            mnemonic=state.wallet_mnemonic,
            jm_settings=jm_settings,
            taker_config_cls=TakerConfig,
        )
        return Taker(wallet=ws, backend=backend, config=config)

    async def _maker_factory(_phase: Any) -> Any:
        backend = await get_backend(
            state.data_dir,
            force_new=True,
            wallet_service=ws,
        )
        config = MakerConfig(
            mnemonic=state.wallet_mnemonic,
            network=jm_settings.network_config.network,
            directory_servers=jm_settings.get_directory_servers(),
            socks_host=jm_settings.tor.socks_host,
            socks_port=jm_settings.tor.socks_port,
            stream_isolation=jm_settings.tor.stream_isolation,
            # Log maker history into the daemon's data dir (#531).
            data_dir=state.data_dir,
        )
        # Tumbler maker sessions must run as 0-fee sw0absoffer with no
        # fidelity bond. See ``tumbler.maker_policy`` for the rationale.
        from tumbler.maker_policy import apply_tumbler_maker_policy

        apply_tumbler_maker_policy(config)
        return MakerBot(wallet=ws, backend=backend, config=config)

    def _on_state_changed(p: Plan) -> None:
        state.broadcast_ws(
            {
                "tumbler": {
                    "event": "plan_updated",
                    "wallet_name": p.wallet_name,
                    "status": str(p.status),
                    "current_phase": p.current_phase,
                    "total_phases": len(p.phases),
                }
            }
        )

    async def _get_confirmations(txid: str) -> int | None:
        """Return confirmation count for ``txid`` via the shared backend.

        Two-stage lookup (see :func:`tumbler.confirmations.resolve_confirmations`):

        1. ``backend.get_transaction(txid)`` (full nodes / mempool.space).
        2. Watched-address fallback via the CoinJoin history file
           (works with neutrino, which cannot fetch arbitrary txids by
           id but *can* match watched addresses via BIP158).
        """
        from tumbler.confirmations import resolve_confirmations

        try:
            backend = await get_backend(state.data_dir, wallet_service=ws)
        except Exception:
            logger.exception("get_confirmations(%s) backend resolution failed", txid)
            return None
        return await resolve_confirmations(txid, backend, state.data_dir)

    ctx = RunnerContext(
        wallet_service=ws,
        wallet_name=state.wallet_name,
        data_dir=state.data_dir,
        taker_factory=_taker_factory,
        maker_factory=_maker_factory,
        on_state_changed=_on_state_changed,
        get_confirmations=_get_confirmations,
        min_confirmations_between_phases=jm_settings.tumbler.min_confirmations_between_phases,
        confirmation_poll_interval=jm_settings.tumbler.confirmation_poll_interval,
        retry_delay_seconds=jm_settings.tumbler.retry_delay_seconds,
    )
    runner = TumbleRunner(plan, ctx)

    state.tumble_runner = runner
    state.tumble_plan_wallet = state.wallet_name
    state.activate_coinjoin_state(CoinjoinState.TUMBLER_RUNNING)

    async def _run() -> Plan:
        try:
            return await runner.run()
        except Exception:
            logger.exception("tumbler runner crashed")
            raise
        finally:
            state.activate_coinjoin_state(CoinjoinState.NOT_RUNNING)
            state.tumble_runner = None
            state.tumble_plan_wallet = None
            state.tumble_task = None

    state.tumble_task = asyncio.create_task(_run())
    return JSONResponse(content={}, status_code=202)

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

Cooperatively stop the running plan; transition it to CANCELLED.

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

@router.post("/wallet/{walletname}/tumbler/stop", status_code=202, operation_id="tumblerstop")
async def stop_plan(
    walletname: str,
    _auth: dict[str, Any] = Depends(require_auth),
    _wallet: None = Depends(require_wallet_match),
    state: DaemonState = Depends(get_daemon_state),
) -> JSONResponse:
    """Cooperatively stop the running plan; transition it to ``CANCELLED``."""
    if not _runner_alive_for(state, state.wallet_name):
        raise ServiceNotStarted("No tumbler is running for this wallet.")

    runner = state.tumble_runner
    task = state.tumble_task
    assert runner is not None and task is not None  # noqa: S101  -- invariant
    # Keep stop responsive: signal cancellation and let the runner finish in
    # the background. Waiting inline here can exceed client/read timeouts when
    # the active phase is in network I/O.
    runner.request_stop()

    async def _finish_stop() -> None:
        try:
            await runner.stop_and_wait(task)
        except Exception:
            logger.exception("error while stopping tumbler runner")
            if not task.done():
                task.cancel()
                with contextlib.suppress(asyncio.CancelledError, Exception):
                    await task

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