tumbler.builder

tumbler.builder

Plan builder.

Constructs a :class:~tumbler.plan.Plan from a list of destination addresses and the current wallet state (per-mixdepth balances). Callers are expected to hold wallet locks while reading balances; the builder itself is pure and side-effect free so it is trivially testable.

Shape of the produced plan

The plan is a sequence of phases grouped into two conceptual stages:

• Stage 1 - origin cleavage. For every non-empty mixdepth, schedule one :class:~tumbler.plan.TakerCoinjoinPhase sweep to an internal address of the next mixdepth. This severs the pre-tumble coin graph before any funds reach a destination.
• Stage 2 - destination mixing. For each destination, schedule a small number of fractional taker CoinJoins, optionally interleaved with :class:~tumbler.plan.MakerSessionPhase sessions, and a final sweep to the destination address.

The role mixing mitigates, but does not eliminate, the subset-sum signature problem tracked in upstream issue #114. See the design doc at docs/technical/tumbler-redesign.md for rationale.

Attributes

INTERNAL_DESTINATION = 'INTERNAL' module-attribute

_ = PhaseKind module-attribute

Classes

PlanBuilder dataclass

Turns :class:TumbleParameters into a :class:~tumbler.plan.Plan.

Source code in tumbler/src/tumbler/builder.py

@dataclass
class PlanBuilder:
    """Turns :class:`TumbleParameters` into a :class:`~tumbler.plan.Plan`."""

    wallet_name: str
    params: TumbleParameters
    _phase_counter: int = field(default=0, init=False)

    # ------------------------------------------------------------------ public

    def build(self) -> Plan:
        if self.params.num_destinations < 1:
            raise ValueError("at least one destination address is required")
        if self.params.num_destinations > len(self._stage2_mixdepth_chain()):
            raise ValueError("tumble requires at least as many usable mixdepths as destinations")

        rng = self.params.rng
        phases: list[Phase] = []
        phases.extend(self._stage1_cleavage(rng))
        phases.extend(self._stage2_destinations(rng))

        plan_params = PlanParameters(
            maker_count_min=self.params.maker_count_min,
            maker_count_max=self.params.maker_count_max,
            time_lambda_seconds=self.params.time_lambda_seconds,
            include_maker_sessions=self.params.include_maker_sessions,
            mincjamount_sats=self.params.mincjamount_sats,
            max_phase_retries=self.params.max_phase_retries,
            seed=self.params.seed,
        )
        return Plan(
            wallet_name=self.wallet_name,
            destinations=list(self.params.destinations),
            parameters=plan_params,
            phases=phases,
        )

    # ------------------------------------------------------------------ stages

    def _stage1_cleavage(self, rng: random.Random) -> list[Phase]:
        """Sweep every non-empty mixdepth into the next mixdepth's internal address."""
        phases: list[Phase] = []
        # Match the reference tumbler: sweep higher mixdepths first so coins
        # forwarded from a lower mixdepth are not immediately swept again.
        for mixdepth in sorted(self.params.mixdepth_balances, reverse=True):
            balance = self.params.mixdepth_balances[mixdepth]
            if balance <= 0:
                continue
            phases.append(
                self._new_phase(
                    TakerCoinjoinPhase,
                    mixdepth=mixdepth,
                    amount=0,
                    counterparty_count=self._sample_counterparty_count(rng),
                    destination=INTERNAL_DESTINATION,
                    wait_seconds=self._sample_wait(rng, stage1=True),
                )
            )
        return phases

    def _stage2_destinations(self, rng: random.Random) -> list[Phase]:
        """
        Traverse the post-stage-1 mixdepth chain, emitting one role-mixed
        stage-2 block per mixdepth.

        This mirrors the reference tumbler's structure: every stage-2
        mixdepth gets maker and fractional activity, intermediate final
        sweeps advance to ``INTERNAL``, and only the last N mixdepths in the
        chain sweep to the user-supplied external destinations.
        """
        phases: list[Phase] = []
        chain = self._stage2_mixdepth_chain()
        destination_map = dict(
            zip(chain[-self.params.num_destinations :], self.params.destinations)
        )
        for chain_index, mixdepth in enumerate(chain):
            if self.params.include_maker_sessions:
                phases.append(
                    self._new_phase(
                        MakerSessionPhase,
                        duration_seconds=self.params.maker_session_seconds,
                        target_cj_count=None,
                        idle_timeout_seconds=self.params.maker_session_idle_timeout_seconds,
                        wait_seconds=self._sample_wait(rng),
                    )
                )
            # Fractional destination CJs (at least mintxcount-1 before the sweep).
            fractions = self._destination_fractions(self.params.mintxcount, rng)
            for fraction in fractions:
                phases.append(
                    self._new_phase(
                        TakerCoinjoinPhase,
                        mixdepth=mixdepth,
                        amount_fraction=fraction,
                        counterparty_count=self._sample_counterparty_count(rng),
                        destination=INTERNAL_DESTINATION,
                        wait_seconds=self._sample_wait(rng),
                        rounding_sigfigs=self._sample_rounding_sigfigs(rng),
                    )
                )
            # Final sweep of the mixdepth: advance internally until the last
            # destination-bearing mixdepths, then sweep externally.
            is_last = chain_index == len(chain) - 1
            phases.append(
                self._new_phase(
                    TakerCoinjoinPhase,
                    mixdepth=mixdepth,
                    amount=0,
                    counterparty_count=self._sample_counterparty_count(rng),
                    destination=destination_map.get(mixdepth, INTERNAL_DESTINATION),
                    # No trailing wait on the very last phase.
                    wait_seconds=0.0 if is_last else self._sample_wait(rng),
                )
            )
        return phases

    # ------------------------------------------------------------------ helpers

    def _new_phase(self, cls: type[Phase], **kwargs: Any) -> Phase:
        phase = cls(index=self._phase_counter, **kwargs)  # type: ignore[call-arg]
        self._phase_counter += 1
        return phase

    def _nonempty_mixdepths(self) -> list[int]:
        return sorted(m for m, bal in self.params.mixdepth_balances.items() if bal > 0)

    def _nonempty_mixdepth_count(self) -> int:
        return len(self._nonempty_mixdepths())

    def _stage2_mixdepth_chain(self) -> list[int]:
        """Return the stage-2 source mixdepth chain.

        The reference tumbler mixes through ``wallet_mixdepth_count - 1``
        successive mixdepths starting from the lowest mixdepth left non-empty
        after the descending stage-1 sweeps.
        """
        wallet_mixdepth_count = max(self.params.mixdepth_balances) + 1
        occupied = set(self._nonempty_mixdepths())
        for origin in sorted(occupied, reverse=True):
            occupied.discard(origin)
            occupied.add((origin + 1) % wallet_mixdepth_count)
        if not occupied:
            return []
        lowest = min(occupied)
        chain_length = max(wallet_mixdepth_count - 1, 0)
        return [((lowest + offset) % wallet_mixdepth_count) for offset in range(chain_length)]

    def _sample_counterparty_count(self, rng: random.Random) -> int:
        lo = self.params.maker_count_min
        hi = self.params.maker_count_max
        # Light normal distribution bounded into [lo, hi].
        mu = (lo + hi) / 2.0
        sigma = max((hi - lo) / 4.0, 0.5)
        value = int(round(rng.gauss(mu, sigma)))
        return max(lo, min(hi, value))

    def _sample_wait(self, rng: random.Random, stage1: bool = False) -> float:
        lam = self.params.time_lambda_seconds * (
            self.params.stage1_wait_multiplier if stage1 else 1.0
        )
        # Exponential with mean ``lam``. Clamp to avoid pathological pauses.
        u = rng.random()
        # rng.random() can return 0; guard against log(0).
        u = max(u, 1e-9)
        return min(-math.log(u) * lam, lam * 10.0)

    def _sample_rounding_sigfigs(self, rng: random.Random) -> int | None:
        """Sample a sigfig count or ``None`` to skip rounding for this phase.

        Mirrors the reference's per-tx ``do_round = random() < rounding_chance``
        followed by a weighted choice over ``rounding_sigfig_weights``.
        """
        chance = self.params.rounding_chance
        if chance <= 0.0 or rng.random() >= chance:
            return None
        weights = list(self.params.rounding_sigfig_weights)
        # Weighted choice over [1, 2, 3, 4, 5] sigfigs.
        return int(rng.choices([1, 2, 3, 4, 5], weights=weights, k=1)[0])

    def _destination_fractions(self, mintxcount: int, rng: random.Random) -> list[float]:
        """
        Return ``mintxcount - 1`` fractions in ``(0, 1)`` summing to < 1,
        with each at least 0.05. The remainder is swept by the trailing
        sweep phase. Uses the reference 'sorted knives' uniform scheme.
        """
        count = max(mintxcount - 1, 1)
        if count == 1:
            return [round(rng.uniform(0.3, 0.6), 4)]
        knives = sorted(rng.uniform(0.0, 1.0) for _ in range(count))
        fractions: list[float] = []
        prev = 0.0
        for knife in knives:
            fractions.append(max(knife - prev, 0.05))
            prev = knife
        # Normalize so the sum does not reach 1.0 (leave >= 5% for the sweep).
        total = sum(fractions) + 0.05
        if total >= 1.0:
            scale = 0.9 / total
            fractions = [max(round(f * scale, 4), 0.05) for f in fractions]
        return fractions

Attributes

params: TumbleParameters instance-attribute

wallet_name: str instance-attribute

Methods:

build() -> Plan

Source code in tumbler/src/tumbler/builder.py

def build(self) -> Plan:
    if self.params.num_destinations < 1:
        raise ValueError("at least one destination address is required")
    if self.params.num_destinations > len(self._stage2_mixdepth_chain()):
        raise ValueError("tumble requires at least as many usable mixdepths as destinations")

    rng = self.params.rng
    phases: list[Phase] = []
    phases.extend(self._stage1_cleavage(rng))
    phases.extend(self._stage2_destinations(rng))

    plan_params = PlanParameters(
        maker_count_min=self.params.maker_count_min,
        maker_count_max=self.params.maker_count_max,
        time_lambda_seconds=self.params.time_lambda_seconds,
        include_maker_sessions=self.params.include_maker_sessions,
        mincjamount_sats=self.params.mincjamount_sats,
        max_phase_retries=self.params.max_phase_retries,
        seed=self.params.seed,
    )
    return Plan(
        wallet_name=self.wallet_name,
        destinations=list(self.params.destinations),
        parameters=plan_params,
        phases=phases,
    )

TumbleParameters dataclass

High-level knobs passed to :meth:PlanBuilder.build.

These are copied into :class:~tumbler.plan.PlanParameters on the plan so the choices can be inspected later and the plan can be rebuilt for debugging. They do not drive the runner directly.

Source code in tumbler/src/tumbler/builder.py

@dataclass
class TumbleParameters:
    """
    High-level knobs passed to :meth:`PlanBuilder.build`.

    These are copied into :class:`~tumbler.plan.PlanParameters` on the plan
    so the choices can be inspected later and the plan can be rebuilt for
    debugging. They do not drive the runner directly.
    """

    destinations: list[str]
    mixdepth_balances: Mapping[int, int]
    """Current confirmed balance per mixdepth, in satoshis."""
    maker_count_min: int = 5
    maker_count_max: int = 9
    time_lambda_seconds: float = 6.0 * 60.0 * 60.0
    """Average wait between phases (mean of an exponential distribution).

    Default is 6 hours, matching the reference implementation's
    ``--timelambda=360`` option. Anything shorter risks defeating the
    timing-correlation defence; values are clamped per-sample at 10x mean.
    """
    stage1_wait_multiplier: float = 3.0
    """Multiplier on ``time_lambda_seconds`` for stage-1 (cleavage) sweeps.

    Reference default is 3x: stage-1 cuts the link to the pre-tumble coin
    history, so longer waits there are particularly important.
    """
    include_maker_sessions: bool = True
    mincjamount_sats: int = 100_000
    maker_session_seconds: float = 12.0 * 60.0 * 60.0
    maker_session_idle_timeout_seconds: float | None = None
    """If set, maker phases exit successfully when no CoinJoin has been served
    within this many seconds. Useful as a safety fallback when the wallet is
    never selected as a counterparty."""
    mintxcount: int = 2
    """Minimum number of destination-bearing taker CJs per mixdepth (excluding sweep)."""
    max_phase_retries: int = 3
    """Maximum re-tries per failed taker CoinJoin phase before the plan fails."""
    rounding_chance: float = 0.25
    """Probability that any given non-sweep taker CJ amount is rounded to a
    random number of significant figures. Reference default is 0.25 (25%).
    Set to 0.0 to disable rounding entirely. Sweeps are never rounded.
    """
    rounding_sigfig_weights: tuple[float, float, float, float, float] = (55, 15, 25, 65, 40)
    """Weights for choosing the number of significant figures to round to
    when rounding fires. The five weights map to 1, 2, 3, 4, 5 sigfigs
    respectively. Reference defaults from
    ``jmclient/cli_options.py:rounding_sigfig_weights``.
    """
    seed: int | None = None

    @property
    def rng(self) -> random.Random:
        return random.Random(self.seed)

    @property
    def num_destinations(self) -> int:
        return len(self.destinations)

Attributes

destinations: list[str] instance-attribute

include_maker_sessions: bool = True class-attribute instance-attribute

maker_count_max: int = 9 class-attribute instance-attribute

maker_count_min: int = 5 class-attribute instance-attribute

maker_session_idle_timeout_seconds: float | None = None class-attribute instance-attribute

If set, maker phases exit successfully when no CoinJoin has been served within this many seconds. Useful as a safety fallback when the wallet is never selected as a counterparty.

maker_session_seconds: float = 12.0 * 60.0 * 60.0 class-attribute instance-attribute

max_phase_retries: int = 3 class-attribute instance-attribute

Maximum re-tries per failed taker CoinJoin phase before the plan fails.

mincjamount_sats: int = 100000 class-attribute instance-attribute

mintxcount: int = 2 class-attribute instance-attribute

Minimum number of destination-bearing taker CJs per mixdepth (excluding sweep).

mixdepth_balances: Mapping[int, int] instance-attribute

Current confirmed balance per mixdepth, in satoshis.

num_destinations: int property

rng: random.Random property

rounding_chance: float = 0.25 class-attribute instance-attribute

Probability that any given non-sweep taker CJ amount is rounded to a random number of significant figures. Reference default is 0.25 (25%). Set to 0.0 to disable rounding entirely. Sweeps are never rounded.

rounding_sigfig_weights: tuple[float, float, float, float, float] = (55, 15, 25, 65, 40) class-attribute instance-attribute

Weights for choosing the number of significant figures to round to when rounding fires. The five weights map to 1, 2, 3, 4, 5 sigfigs respectively. Reference defaults from jmclient/cli_options.py:rounding_sigfig_weights.

seed: int | None = None class-attribute instance-attribute

stage1_wait_multiplier: float = 3.0 class-attribute instance-attribute

Multiplier on time_lambda_seconds for stage-1 (cleavage) sweeps.

Reference default is 3x: stage-1 cuts the link to the pre-tumble coin history, so longer waits there are particularly important.

time_lambda_seconds: float = 6.0 * 60.0 * 60.0 class-attribute instance-attribute

Average wait between phases (mean of an exponential distribution).

Default is 6 hours, matching the reference implementation's --timelambda=360 option. Anything shorter risks defeating the timing-correlation defence; values are clamped per-sample at 10x mean.