maker.bot

maker.bot

Main maker bot implementation.

Coordinates all maker components: - Wallet synchronization - Directory server connections - Offer creation and announcement - CoinJoin protocol handling

Attributes

MAX_LOG_RATE_LIMIT_ENTRIES = 200000 module-attribute

Classes

MakerBot

Bases: BackgroundTasksMixin, ProtocolHandlersMixin, DirectConnectionMixin

Main maker bot coordinating all components.

Source code in maker/src/maker/bot.py

class MakerBot(BackgroundTasksMixin, ProtocolHandlersMixin, DirectConnectionMixin):
    """
    Main maker bot coordinating all components.
    """

    def __init__(
        self,
        wallet: WalletService,
        backend: BlockchainBackend,
        config: MakerConfig,
    ):
        self.wallet = wallet
        self.backend = backend
        self.config = config

        # Create nick identity for signing messages
        self.nick_identity = NickIdentity(JM_VERSION)
        self.nick = self.nick_identity.nick

        self.offer_manager = OfferManager(self.wallet, config, self.nick)

        self.directory_clients: dict[str, DirectoryClient] = {}
        # Shared connection plumbing (parsing, SOCKS isolation creds,
        # DirectoryClient construction, retry loop). The pool stores its
        # connected clients in self.directory_clients via the shared-dict
        # binding below, so tests and existing call sites that mutate
        # self.directory_clients directly continue to work unchanged.
        self._directory_pool = MakerDirectoryPool(
            config=config,
            nick_identity=self.nick_identity,
            neutrino_compat=backend.can_provide_neutrino_metadata(),
        )
        self._directory_pool.clients = self.directory_clients
        self.active_sessions: dict[str, MakerSession] = {}
        self._reserved_commitments: set[str] = set()
        self.current_offers: list[Offer] = []
        self.fidelity_bond: FidelityBondInfo | None = None
        self.current_block_height: int = 0  # Cached block height for bond proof generation

        self.running = False
        self.listen_tasks: list[asyncio.Task[None]] = []
        self._session_cleanup_task: asyncio.Task[None] | None = None
        self._session_handler_task_count = 0
        self._detached_handler_tasks: set[asyncio.Task[None]] = set()
        self._pending_signed_rounds: dict[tuple[str, str], PendingSignedRound] = {}
        self._pending_signed_rounds_lock = asyncio.Lock()
        self._fatal_error: Exception | None = None

        # Session locks now live on each `MakerSession` (one asyncio.Lock per
        # taker_nick) so we no longer keep a parallel dict on the bot.

        # Hidden service listener for direct peer connections
        self.hidden_service_listener: HiddenServiceListener | None = None
        self.direct_connections: dict[str, TCPConnection] = {}
        self._direct_connection_states: dict[TCPConnection, DirectConnectionState] = {}

        # Tor control for dynamic hidden service creation
        self._tor_control: TorControlClient | None = None
        self._ephemeral_hidden_service: EphemeralHiddenService | None = None

        # Generic per-peer rate limiter (token bucket algorithm)
        # Generous burst (100 msgs) but low sustained rate (10 msg/s)
        self._message_rate_limiter = RateLimiter(
            rate_limit=config.message_rate_limit,
            burst_limit=config.message_burst_limit,
        )

        # Fidelity bond addresses loaded at startup, kept for periodic rescans so
        # newly funded bonds are detected without requiring a restart.
        self._fidelity_bond_addresses: list[tuple[str, int, int]] = []

        # Rate limiter for orderbook requests to prevent spam attacks
        self._orderbook_rate_limiter = OrderbookRateLimiter(
            rate_limit=config.orderbook_rate_limit,
            interval=config.orderbook_rate_interval,
            violation_ban_threshold=config.orderbook_violation_ban_threshold,
            violation_warning_threshold=config.orderbook_violation_warning_threshold,
            violation_severe_threshold=config.orderbook_violation_severe_threshold,
            ban_duration=config.orderbook_ban_duration,
        )

        # Rate limiter specifically for direct hidden service connections
        # This tracks by connection address (not nick) to prevent nick rotation attacks
        # where attackers use a different nick per request
        self._direct_connection_rate_limiter = DirectConnectionRateLimiter(
            message_rate_per_sec=5.0,  # Stricter than directory (5 msg/s vs 10)
            message_burst=20,  # Smaller burst
            orderbook_interval=30.0,  # Longer interval (30s vs 10s)
            orderbook_ban_threshold=10,  # Faster ban (10 violations vs 100)
            ban_duration=config.orderbook_ban_duration,
        )

        # Message deduplicator to handle receiving same message from multiple directories
        # This prevents processing duplicates and avoids false rate limit violations
        self._message_deduplicator = MessageDeduplicator(window_seconds=30.0)

        self._hp2_own_broadcast_semaphore = asyncio.Semaphore(1)
        self._hp2_relay_broadcast_semaphore = asyncio.Semaphore(1)

        # Track failed directory reconnection attempts
        # Key: node_id (host:port), Value: number of reconnection attempts
        self._directory_reconnect_attempts: dict[str, int] = {}

        # Track whether all directories were previously disconnected, so we can
        # send a recovery notification when at least one reconnects
        self._all_directories_disconnected: bool = False

        # Track CoinJoin txids we have already sent a mempool notification for,
        # so we do not re-notify on every pending-confirmation poll cycle.
        self._mempool_notified_txids: set[str] = set()

        # Track last log time for rate-limited logging
        # Key: log_key, Value: timestamp of last log
        self._rate_limited_log_times: dict[str, float] = {}

        # Own wallet nicks to exclude from CoinJoin sessions (self-CoinJoin protection)
        # Read the taker nick from state file if running both components from same wallet
        self._own_wallet_nicks: set[str] = set()
        taker_nick = read_nick_state(config.data_dir, "taker")
        if taker_nick:
            self._own_wallet_nicks.add(taker_nick)
            logger.info(f"Self-CoinJoin protection: excluding taker nick {taker_nick}")

    async def _setup_tor_hidden_service(self) -> str | None:
        """
        Create an ephemeral hidden service via Tor control port.

        Also configures Tor-level DoS defenses (intro point rate limiting, PoW)
        based on the hidden_service_dos configuration.

        Returns:
            The .onion address if successful, None otherwise
        """
        if not self.config.tor_control.enabled:
            logger.debug("Tor control port integration disabled")
            return None

        # Retry on transient auth failures (e.g. cookie file not yet fully written by Tor)
        max_auth_retries = 5
        auth_retry_delay = 3.0
        last_auth_error: TorAuthenticationError | None = None
        for attempt in range(1, max_auth_retries + 1):
            try:
                return await self._try_setup_tor_hidden_service()
            except TorAuthenticationError as e:
                last_auth_error = e
                logger.warning(
                    f"Tor authentication failed (attempt {attempt}/{max_auth_retries}): {e} "
                    f"— retrying in {auth_retry_delay}s..."
                )
                await asyncio.sleep(auth_retry_delay)
            except TorControlError as e:
                # Non-auth errors are not retried — log and fall back gracefully
                logger.warning(
                    f"Could not create ephemeral hidden service via Tor control port: {e}\n"
                    f"  Tor control configured: "
                    f"{self.config.tor_control.host}:{self.config.tor_control.port}\n"
                    f"  Cookie path: {self.config.tor_control.cookie_path}\n"
                    f"  → Maker will advertise 'NOT-SERVING-ONION' and rely on directory routing."
                )
                return None

        # All retries exhausted — log warning and fall back to NOT-SERVING-ONION
        logger.warning(
            f"Could not authenticate to Tor control port after {max_auth_retries} attempts: "
            f"{last_auth_error}\n"
            f"  Tor control configured: "
            f"{self.config.tor_control.host}:{self.config.tor_control.port}\n"
            f"  Cookie path: {self.config.tor_control.cookie_path}\n"
            f"  → Maker will advertise 'NOT-SERVING-ONION' and rely on directory routing.\n"
            f"  → Ensure the Tor cookie file is readable and Tor has fully started."
        )
        return None

    async def _try_setup_tor_hidden_service(self) -> str | None:
        """
        Single attempt to create an ephemeral hidden service via Tor control port.
        Raises TorControlError (including TorAuthenticationError) on failure.
        """
        try:
            logger.info(
                f"Connecting to Tor control port at "
                f"{self.config.tor_control.host}:{self.config.tor_control.port}..."
            )

            self._tor_control = TorControlClient(
                control_host=self.config.tor_control.host,
                control_port=self.config.tor_control.port,
                cookie_path=self.config.tor_control.cookie_path,
                password=self.config.tor_control.password.get_secret_value()
                if self.config.tor_control.password
                else None,
            )

            await self._tor_control.connect()
            await self._tor_control.authenticate()

            # Get Tor version and capabilities for logging and DoS defense setup
            try:
                tor_version = await self._tor_control.get_version()
                logger.info(f"Connected to Tor {tor_version}")
                caps = await self._tor_control.get_capabilities()
            except TorControlError:
                logger.debug("Could not get Tor version (non-critical)")
                caps = None

            # Create ephemeral hidden service
            # Maps external port (advertised) to our local serving port
            dos_config = self.config.hidden_service_dos
            logger.info(
                f"Creating ephemeral hidden service on port {self.config.onion_serving_port} -> "
                f"{self.config.tor_target_host}:{self.config.onion_serving_port}..."
            )

            self._ephemeral_hidden_service = (
                await self._tor_control.create_ephemeral_hidden_service(
                    ports=[
                        (
                            self.config.onion_serving_port,
                            f"{self.config.tor_target_host}:{self.config.onion_serving_port}",
                        )
                    ],
                    # Don't discard private key in case we want to log it for debugging
                    discard_pk=True,
                    # Don't detach - we want the service to be removed when we disconnect
                    detach=False,
                    # Apply max_streams limit if configured (DoS protection)
                    max_streams=dos_config.max_streams,
                    # Apply Tor-level DoS defenses (intro point rate limiting, PoW)
                    # These must be set at creation time for ephemeral hidden services
                    dos_config=dos_config,
                )
            )

            logger.info(
                f"Created ephemeral hidden service: {self._ephemeral_hidden_service.onion_address}"
            )

            # Log summary of active defenses (only those actually applied to ephemeral HS)
            defenses = []
            # Note: intro_dos is NOT supported for ephemeral HS, don't list it as active
            # Note: PoW via ADD_ONION requires Tor 0.4.9.2+
            if dos_config.pow_enabled and caps and caps.has_add_onion_pow:
                defenses.append("PoW=enabled")
            if dos_config.max_streams:
                defenses.append(f"max_streams={dos_config.max_streams}")
            if defenses:
                logger.info(f"Tor DoS defenses active: {', '.join(defenses)}")
            else:
                logger.info(
                    "No Tor-level DoS defenses active for ephemeral HS "
                    "(requires Tor 0.4.9.2+ for PoW, or use persistent HS in torrc)"
                )

            return self._ephemeral_hidden_service.onion_address

        except TorControlError:
            # Clean up partial connection before re-raising
            if self._tor_control:
                await self._tor_control.close()
                self._tor_control = None
            raise

    async def _cleanup_tor_hidden_service(self) -> None:
        """Remove the ephemeral hidden service and close the Tor control connection."""
        if self._ephemeral_hidden_service and self._tor_control:
            try:
                await self._tor_control.delete_ephemeral_hidden_service(
                    self._ephemeral_hidden_service.service_id
                )
                logger.info("Removed ephemeral Tor hidden service")
            except TorControlError as e:
                logger.warning(f"Failed to remove ephemeral hidden service: {e}")
        if self._tor_control:
            try:
                await self._tor_control.close()
            except Exception as e:
                logger.warning(f"Error closing Tor control connection: {e}")
            self._tor_control = None
            self._ephemeral_hidden_service = None

    async def _regenerate_nick(self) -> None:
        """
        Regenerate nick identity for privacy (currently disabled).

        Nick regeneration is disabled because:
        1. Reference implementation doesn't regenerate nicks after CoinJoin
        2. Fidelity bond makers need stable identity for reputation
        3. Causes timing issues with !push (taker waits ~60s to collect signatures)
        4. Privacy is maintained through Tor hidden services

        Future consideration: Could be re-enabled as opt-in feature with grace period.
        """
        pass

    async def start(self) -> None:
        """
        Start the maker bot.

        Flow:
        1. Initialize commitment blacklist
        2. Sync wallet with blockchain
        3. Create ephemeral hidden service if tor_control enabled
        4. Connect to directory servers
        5. Create and announce offers
        6. Listen for taker requests
        """
        try:
            logger.info(f"Starting maker bot (nick: {self.nick})")

            # Log wallet name if using descriptor wallet backend
            from jmwallet.backends.descriptor_wallet import DescriptorWalletBackend

            if isinstance(self.backend, DescriptorWalletBackend):
                logger.info(f"Using wallet: {self.backend.wallet_name}")

            # Initialize commitment blacklist with configured data directory
            set_blacklist_path(data_dir=self.config.data_dir)

            # Load fidelity bond addresses for optimized scanning
            # We scan wallet + fidelity bonds in a single pass to avoid extra
            # backend round-trips during sync.
            from jmcore.paths import get_default_data_dir
            from jmwallet.wallet.bond_registry import load_registry

            resolved_data_dir = (
                self.config.data_dir if self.config.data_dir else get_default_data_dir()
            )
            fidelity_bond_addresses: list[tuple[str, int, int]] = []

            # Fidelity bonds are explicitly disabled
            if self.config.no_fidelity_bond:
                logger.info(
                    "Fidelity bonds disabled (--no-fidelity-bond). Running without bond proof."
                )
            # Option 1: Manual specification via fidelity_bond_index + locktimes (bypasses registry)
            # This is useful when running in Docker or when you don't have a registry yet
            elif (
                self.config.fidelity_bond_index is not None and self.config.fidelity_bond_locktimes
            ):
                logger.info(
                    f"Using manual fidelity bond specification: "
                    f"locktimes={self.config.fidelity_bond_locktimes}"
                )
                for locktime in self.config.fidelity_bond_locktimes:
                    from jmcore.timenumber import timestamp_to_timenumber

                    timenumber = timestamp_to_timenumber(locktime)
                    address = self.wallet.get_fidelity_bond_address(timenumber, locktime)
                    fidelity_bond_addresses.append((address, locktime, timenumber))
                    logger.info(
                        f"Generated fidelity bond address for locktime {locktime}: {address}"
                    )
            # Option 2: Load from registry (default)
            else:
                bond_registry = load_registry(
                    resolved_data_dir,
                    self.wallet.wallet_fingerprint,
                    allow_legacy_fallback=False,
                )
                network_bonds = [
                    bond for bond in bond_registry.bonds if bond.network == self.config.network
                ]
                if network_bonds:
                    # Extract (address, locktime, index) tuples from registry
                    fidelity_bond_addresses = [
                        (bond.address, bond.locktime, bond.index) for bond in network_bonds
                    ]
                    logger.info(
                        f"Loaded {len(fidelity_bond_addresses)} "
                        f"fidelity bond address(es) from registry"
                    )

            logger.info("Syncing wallet and fidelity bonds...")

            # Store bond addresses on the instance so periodic rescans can use them
            # to detect newly funded bonds without requiring a restart.
            self._fidelity_bond_addresses = fidelity_bond_addresses

            # Setup descriptor wallet if needed (one-time operation)
            if isinstance(self.backend, DescriptorWalletBackend):
                # Check if base wallet is set up (without counting bonds)
                base_wallet_ready = await self.wallet.is_descriptor_wallet_ready(
                    fidelity_bond_count=0
                )
                # Check if wallet with bonds is set up
                full_wallet_ready = await self.wallet.is_descriptor_wallet_ready(
                    fidelity_bond_count=len(fidelity_bond_addresses)
                )

                if not base_wallet_ready:
                    # First time setup - import everything including bonds
                    logger.info("Descriptor wallet not set up. Importing descriptors...")
                    await self.wallet.setup_descriptor_wallet(
                        rescan=True,
                        fidelity_bond_addresses=fidelity_bond_addresses,
                    )
                    logger.info("Descriptor wallet setup complete")
                elif not full_wallet_ready and fidelity_bond_addresses:
                    # Base wallet exists but bonds are missing - import just the bonds
                    logger.info(
                        "Descriptor wallet exists but fidelity bond addresses not imported. "
                        "Importing bond addresses..."
                    )
                    await self.wallet.import_fidelity_bond_addresses(
                        fidelity_bond_addresses, rescan=True
                    )

                # Use fast descriptor wallet sync
                await self.wallet.sync_with_descriptor_wallet(fidelity_bond_addresses)
            else:
                # Use standard sync (BIP157/158 for neutrino, mempool API, etc.)
                await self.wallet.sync_all(fidelity_bond_addresses)
            await self.wallet.reconstruct_imported_state_safe()

            # Update bond registry with UTXO info from the scan (only if using registry)
            if self.config.fidelity_bond_index is None and fidelity_bond_addresses:
                from jmwallet.wallet.bond_registry import save_registry

                bond_registry = load_registry(
                    resolved_data_dir,
                    self.wallet.wallet_fingerprint,
                    allow_legacy_fallback=False,
                )
                for bond in bond_registry.bonds:
                    # Find the UTXO for this bond address in mixdepth 0
                    bond_utxo = next(
                        (
                            utxo
                            for utxo in self.wallet.utxo_cache.get(0, [])
                            if utxo.address == bond.address
                        ),
                        None,
                    )
                    if bond_utxo:
                        # Update the bond registry with UTXO info
                        bond.txid = bond_utxo.txid
                        bond.vout = bond_utxo.vout
                        bond.value = bond_utxo.value
                        bond.confirmations = bond_utxo.confirmations
                        logger.debug(
                            f"Updated bond {bond.address[:20]}... with UTXO "
                            f"{bond_utxo.txid[:16]}...:{bond_utxo.vout}, value={bond_utxo.value}"
                        )

                # Save updated registry
                save_registry(bond_registry, resolved_data_dir, self.wallet.wallet_fingerprint)

            # Get current block height for bond proof generation
            self.current_block_height = await self.backend.get_block_height()
            logger.debug(f"Current block height: {self.current_block_height}")

            total_balance = await self.wallet.get_total_balance()
            logger.info(f"Wallet synced. Total balance: {total_balance:,} sats")

            # Find fidelity bond for proof generation
            # If a specific bond is selected in config, use it; otherwise use the best one
            if self.config.no_fidelity_bond:
                self.fidelity_bond = None
                logger.warning("Fidelity bond disabled (offers will have no bond proof)")
            elif self.config.selected_fidelity_bond:
                # User specified a specific bond
                sel_txid, sel_vout = self.config.selected_fidelity_bond
                bonds = await find_fidelity_bonds(self.wallet)
                self.fidelity_bond = next(
                    (b for b in bonds if b.txid == sel_txid and b.vout == sel_vout), None
                )
                if self.fidelity_bond:
                    logger.info(
                        f"Using selected fidelity bond: {sel_txid[:16]}...:{sel_vout}, "
                        f"value={self.fidelity_bond.value:,} sats, "
                        f"bond_value={self.fidelity_bond.bond_value:,}"
                    )
                else:
                    logger.warning(
                        f"Selected fidelity bond {sel_txid[:16]}...:{sel_vout} not found, "
                        "falling back to best available"
                    )
                    self.fidelity_bond = await get_best_fidelity_bond(
                        self.wallet, current_block_height=self.current_block_height
                    )
            else:
                # Auto-select the best (largest bond value) fidelity bond
                self.fidelity_bond = await get_best_fidelity_bond(
                    self.wallet, current_block_height=self.current_block_height
                )
            if self.fidelity_bond:
                ensure_fidelity_bond_certificate_valid(
                    self.fidelity_bond,
                    self.current_block_height,
                )
                logger.info(
                    f"Fidelity bond found: {self.fidelity_bond.txid[:16]}..., "
                    f"value={self.fidelity_bond.value:,} sats, "
                    f"bond_value={self.fidelity_bond.bond_value:,}"
                )
                md0_utxos = _get_fidelity_bond_linkable_utxos(
                    self.wallet,
                    resolved_data_dir,
                )
                if md0_utxos:
                    total_md0 = sum(u.value for u in md0_utxos)
                    logger.warning(
                        f"PRIVACY RISK: You have a fidelity bond AND "
                        f"{len(md0_utxos)} regular UTXO(s) ({total_md0:,} sats) "
                        f"in mixdepth 0.\n"
                        f"Using md0 UTXOs in coinjoins can link your identity "
                        f"to your fidelity bond.\n"
                        f"Recommendation: sweep md0 funds to mixdepth 1 as a "
                        f"taker coinjoin, then freeze or spend the md0 UTXOs."
                    )
            else:
                logger.warning("No fidelity bond found (offers will have no bond proof)")

            logger.info("Creating offers...")
            self.current_offers = await self.offer_manager.create_offers()

            # If no offers due to insufficient balance, wait and retry
            retry_count = 0
            max_retries = 30  # 5 minutes max wait (30 * 10s)
            while not self.current_offers and retry_count < max_retries:
                retry_count += 1
                logger.warning(
                    f"No offers created (insufficient balance?). "
                    f"Waiting 10s and retrying... (attempt {retry_count}/{max_retries})"
                )
                await asyncio.sleep(10)

                # Re-sync wallet to check for new funds
                from jmwallet.backends.descriptor_wallet import DescriptorWalletBackend

                if isinstance(self.backend, DescriptorWalletBackend):
                    await self.wallet.sync_with_descriptor_wallet()
                else:
                    await self.wallet.sync_all()
                await self.wallet.reconstruct_imported_state_safe()
                total_balance = await self.wallet.get_total_balance()
                logger.info(f"Wallet re-synced. Total balance: {total_balance:,} sats")

                self.current_offers = await self.offer_manager.create_offers()

            if not self.current_offers:
                logger.error(
                    f"No offers created after {max_retries} retries. "
                    "Please fund the wallet and restart."
                )
                return

            # Log summary of created offers
            logger.info(f"Created {len(self.current_offers)} offer(s) to announce:")
            for offer in self.current_offers:
                fee_display = (
                    f"{float(offer.cjfee) * 100:.4f}%"
                    if offer.ordertype.value.endswith("reloffer")
                    else f"{offer.cjfee} sats"
                )
                logger.info(
                    f"  oid={offer.oid}: {offer.ordertype.value}, "
                    f"size={offer.minsize:,}-{offer.maxsize:,} sats, fee={fee_display}"
                )

            # Set up ephemeral hidden service via Tor control port if enabled
            # This must happen before connecting to directory servers so we can
            # advertise the onion address
            ephemeral_onion = await self._setup_tor_hidden_service()
            if ephemeral_onion:
                # Override onion_host with the dynamically created one
                object.__setattr__(self.config, "onion_host", ephemeral_onion)
                logger.info(f"Using ephemeral onion address: {ephemeral_onion}")

            # Determine the onion address to advertise
            onion_host = self.config.onion_host

            logger.info("Connecting to directory servers...")
            await self._connect_to_directories_with_retry()

            # Start hidden service listener if we have an onion address (static or ephemeral)
            if onion_host:
                logger.info(
                    f"Starting hidden service listener on "
                    f"{self.config.onion_serving_host}:{self.config.onion_serving_port}..."
                )
                self.hidden_service_listener = HiddenServiceListener(
                    host=self.config.onion_serving_host,
                    port=self.config.onion_serving_port,
                    on_connection=self._on_direct_connection,
                )
                await self.hidden_service_listener.start()
                logger.info(f"Hidden service listener started (onion: {onion_host})")

            logger.info("Announcing offers...")
            await self._announce_offers()

            logger.info("Maker bot started. Listening for takers...")
            self.running = True

            self._start_session_cleanup_task()

            # Start listening on all directory clients
            for node_id, client in self.directory_clients.items():
                task = asyncio.create_task(self._listen_client(node_id, client))
                self.listen_tasks.append(task)

            # If hidden service listener is running, start serve_forever task
            if self.hidden_service_listener:
                task = asyncio.create_task(self.hidden_service_listener.serve_forever())
                self.listen_tasks.append(task)

            # Start background task to monitor pending transactions
            monitor_task = asyncio.create_task(self._monitor_pending_transactions())
            self.listen_tasks.append(monitor_task)

            # Start periodic wallet rescan task
            rescan_task = asyncio.create_task(self._periodic_rescan())
            self.listen_tasks.append(rescan_task)

            # Start periodic rate limit status logging task
            status_task = asyncio.create_task(self._periodic_rate_limit_status())
            self.listen_tasks.append(status_task)

            # Start periodic directory connection status logging task
            conn_status_task = asyncio.create_task(self._periodic_directory_connection_status())
            self.listen_tasks.append(conn_status_task)

            # Start periodic directory reconnection task
            reconnect_task = asyncio.create_task(self._periodic_directory_reconnect())
            self.listen_tasks.append(reconnect_task)

            # Start periodic summary notification task (if enabled)
            notifier = get_notifier()
            if notifier.config.notify_summary:
                summary_task = asyncio.create_task(self._periodic_summary())
                self.listen_tasks.append(summary_task)
            else:
                logger.info("Periodic summary notifications disabled (notify_summary=false)")

            # Wait for all listening tasks to complete
            await asyncio.gather(*self.listen_tasks, return_exceptions=True)
            if self._fatal_error is not None:
                raise self._fatal_error

        except ExpiredFidelityBondCertificateError:
            raise
        except Exception as e:
            logger.error(f"Failed to start maker bot: {e}")
            raise

    async def stop(self) -> None:
        """Stop the maker bot"""
        logger.info("Stopping maker bot...")
        self.running = False

        # Stop the dedicated reaper, then independently expire every exact
        # session object. Handler cancellation and cleanup are bounded inside
        # _expire_timed_out_session, including cancellation-suppressing tasks.
        cleanup_task = self._session_cleanup_task
        if cleanup_task is not None:
            cleanup_task.cancel()
            await asyncio.wait({cleanup_task}, timeout=1.0)

        expiration_tasks = [
            asyncio.create_task(self._expire_timed_out_session(nick, session))
            for nick, session in list(self.active_sessions.items())
        ]
        if expiration_tasks:
            _, pending_expirations = await asyncio.wait(expiration_tasks, timeout=1.5)
            for expiration_task in pending_expirations:
                expiration_task.cancel()

        await self._drain_pending_signed_rounds()

        # Cancellation-suppressing handlers remain strongly referenced until
        # they actually finish. Shutdown sends another cancellation request and
        # waits only for a bounded grace period.
        detached_tasks = set(self._detached_handler_tasks)
        for detached_task in detached_tasks:
            detached_task.cancel()
        if detached_tasks:
            _, pending_detached = await asyncio.wait(
                detached_tasks, timeout=_DETACHED_SHUTDOWN_GRACE_SEC
            )
            if pending_detached:
                logger.warning(
                    f"{len(pending_detached)} detached maker handler task(s) "
                    "still suppress cancellation after shutdown grace period"
                )

        # Cancel all listening tasks after detached handlers have released
        # their directory dispatchers.
        for listener_task in self.listen_tasks:
            listener_task.cancel()

        if self.listen_tasks:
            _, pending_listeners = await asyncio.wait(set(self.listen_tasks), timeout=2.0)
            for listener_task in pending_listeners:
                listener_task.cancel()
        self.listen_tasks.clear()
        self._session_cleanup_task = None

        # Stop hidden service listener
        if self.hidden_service_listener:
            await self.hidden_service_listener.stop()

        # Clean up Tor control connection (ephemeral hidden service auto-removed)
        await self._cleanup_tor_hidden_service()

        # Close every accepted direct socket, including sockets that never
        # completed a handshake or authenticated a sender.
        direct_sockets = set(self._direct_connection_states)
        direct_sockets.update(self.direct_connections.values())
        for conn in direct_sockets:
            try:
                await conn.close()
            except Exception:
                pass
        self.direct_connections.clear()
        self._direct_connection_states.clear()

        # Close all directory clients
        for client in self.directory_clients.values():
            try:
                await client.close()
            except Exception:
                pass

        # Do not close the wallet here as it might be shared (e.g. in jmwalletd)
        # The caller is responsible for managing the wallet lifecycle.
        # await self.wallet.close()
        logger.info("Maker bot stopped")

    def _start_session_cleanup_task(self) -> None:
        """Start the bot-wide session reaper exactly once."""
        if self._session_cleanup_task is not None and not self._session_cleanup_task.done():
            return
        self._session_cleanup_task = asyncio.create_task(
            self._periodic_session_cleanup(), name="maker-session-cleanup"
        )
        self.listen_tasks.append(self._session_cleanup_task)

    def _abort_for_fatal_error(self, error: Exception) -> None:
        """Stop listener tasks so ``start`` can propagate a fatal background error."""
        if self._fatal_error is None:
            self._fatal_error = error
        self.running = False

        current_task = asyncio.current_task()
        for listener_task in self.listen_tasks:
            if listener_task is not current_task:
                listener_task.cancel()

    def _log_rate_limited(
        self, key: str, message: str, level: str = "warning", interval: float = 10.0
    ) -> None:
        """
        Logs a message with rate limiting to prevent log spam.
        """
        now = time.time()

        # Enforce maximum size to prevent memory leak
        if (
            key not in self._rate_limited_log_times
            and len(self._rate_limited_log_times) >= MAX_LOG_RATE_LIMIT_ENTRIES
        ):
            # Remove the oldest entry (dictionaries preserve insertion order in Python 3.7+)
            try:
                oldest_key = next(iter(self._rate_limited_log_times))
                del self._rate_limited_log_times[oldest_key]
            except (StopIteration, KeyError):
                pass

        last_time = self._rate_limited_log_times.get(key, 0.0)
        if now - last_time >= interval:
            # Use getattr to call the correct logger method (e.g., logger.warning, logger.info)
            log_method = getattr(logger, level, logger.warning)
            log_method(message)
            self._rate_limited_log_times[key] = now

    async def _cleanup_timed_out_sessions(self) -> None:
        """Remove timed-out sessions from active_sessions and clean up rate limiter."""
        for sid, session in list(self.active_sessions.items()):
            if session.is_timed_out():
                await self._expire_timed_out_session(sid, session)

        await self._prune_pending_signed_rounds()

        # Ensure rate limiters are cleaned up periodically
        self._direct_connection_rate_limiter.cleanup_old_entries()

    def _release_commitment_reservation(self, commitment: str) -> None:
        """Release a commitment reserved by an in-flight local session."""
        self._reserved_commitments.discard(commitment.lower())

    async def _resync_wallet_and_update_offers(self) -> None:
        """Re-sync wallet and update offers if balance changed.

        This is the core rescan logic used by both post-CoinJoin resync
        and periodic rescan. It:
        1. Saves the current max balance
        2. Re-syncs the wallet
        3. If max balance changed, recreates and re-announces offers
        """
        # Get current max balance available for offers before resync (excludes fidelity bonds)
        old_max_balance = 0
        locked_outpoints = self.wallet.get_locked_input_outpoints()
        for mixdepth in range(self.wallet.mixdepth_count):
            balance = await self.wallet.get_balance_for_offers(
                mixdepth,
                min_confirmations=self.config.min_confirmations,
                restrict_md0=not self.config.allow_mixdepth_zero_merge,
                exclude=locked_outpoints,
            )
            old_max_balance = max(old_max_balance, balance)

        # Sync wallet (use descriptor wallet if available for fast sync)
        from jmwallet.backends.descriptor_wallet import DescriptorWalletBackend

        if isinstance(self.backend, DescriptorWalletBackend):
            await self.wallet.sync_with_descriptor_wallet(self._fidelity_bond_addresses)
        else:
            await self.wallet.sync_all(self._fidelity_bond_addresses)
        await self.wallet.reconstruct_imported_state_safe()

        # Update current block height
        self.current_block_height = await self.backend.get_block_height()
        logger.debug(f"Updated block height: {self.current_block_height}")
        if self.fidelity_bond is not None:
            ensure_fidelity_bond_certificate_valid(
                self.fidelity_bond,
                self.current_block_height,
            )

        # Update pending history immediately after sync (in case of restart)
        await self._update_pending_history()

        # Get new max balance for offers after resync (excludes fidelity bonds)
        new_max_balance = 0
        locked_outpoints = self.wallet.get_locked_input_outpoints()
        for mixdepth in range(self.wallet.mixdepth_count):
            balance = await self.wallet.get_balance_for_offers(
                mixdepth,
                min_confirmations=self.config.min_confirmations,
                restrict_md0=not self.config.allow_mixdepth_zero_merge,
                exclude=locked_outpoints,
            )
            new_max_balance = max(new_max_balance, balance)

        total_balance = await self.wallet.get_total_balance()

        # If max balance changed, update offers and log at INFO so operators
        # see balance/offer churn. When unchanged this rescan is a routine
        # no-op (every `rescan_interval_sec`, default 10 min) and is logged
        # at DEBUG to avoid flooding logs of long-running makers.
        if old_max_balance != new_max_balance:
            logger.info(f"Wallet re-synced. Total balance: {total_balance:,} sats")
            logger.info(
                f"Max balance changed: {old_max_balance:,} -> {new_max_balance:,} sats. "
                "Updating offers..."
            )
            await self._update_offers()
        else:
            logger.debug(
                f"Wallet re-synced (no change). Total balance: {total_balance:,} sats, "
                f"max offer balance: {new_max_balance:,} sats"
            )

    async def _update_offers(self) -> None:
        """Recreate and re-announce offers based on current wallet state.

        Called when wallet balance changes (after CoinJoin, external transaction,
        or deposit). This allows the maker to adapt to changing balances without
        requiring a restart.
        """
        try:
            new_offers = await self.offer_manager.create_offers()

            if not new_offers:
                logger.warning(
                    "No offers could be created (insufficient balance?). "
                    "Keeping existing offers active."
                )
                return

            # Check if offers actually changed (compare maxsize for each offer by ID)
            offers_changed = False
            if self.current_offers and new_offers:
                # Build lookup by offer ID for comparison
                old_offers_by_id = {o.oid: o for o in self.current_offers}
                new_offers_by_id = {o.oid: o for o in new_offers}

                # Check if offer count changed
                if set(old_offers_by_id.keys()) != set(new_offers_by_id.keys()):
                    offers_changed = True
                else:
                    # Check if any offer's maxsize changed
                    for oid, new_offer in new_offers_by_id.items():
                        old_offer = old_offers_by_id.get(oid)
                        if old_offer is None or old_offer.maxsize != new_offer.maxsize:
                            offers_changed = True
                            break

                if not offers_changed:
                    logger.debug("Offer maxsizes unchanged, skipping re-announcement")
                    return
            else:
                offers_changed = True  # First time or recovering from no offers

            # Regenerate nick when offers change for additional privacy
            # This makes it harder for observers to track maker activity over time
            await self._regenerate_nick()

            # Update offers with new nick (OfferManager.maker_nick was updated by _regenerate_nick)
            for offer in new_offers:
                offer.counterparty = self.nick

            self.current_offers = new_offers

            delay_max = self.config.offer_reannounce_delay_max
            if delay_max > 0:
                delay = secure_random.uniform(0, delay_max)
                logger.info(
                    f"Delaying offer re-announcement by {delay:.0f}s (max {delay_max}s) for privacy"
                )
                await asyncio.sleep(delay)

            await self._announce_offers()
            offer_summary = ", ".join(f"oid={o.oid}:{o.maxsize:,}" for o in new_offers)
            logger.info(f"Updated and re-announced {len(new_offers)} offer(s): {offer_summary}")
        except Exception as e:
            logger.error(f"Failed to update offers: {e}")

    async def _announce_offers(self) -> None:
        """Announce offers to all connected directory servers (public broadcast, NO bonds)"""
        for offer in self.current_offers:
            offer_msg = self._format_offer_announcement(offer, include_bond=False)

            for client in self.directory_clients.values():
                try:
                    await client.send_public_message(offer_msg)
                    logger.debug("Announced offer to directory")
                except Exception as e:
                    logger.error(f"Failed to announce offer: {e}")

    def _format_offer_announcement(self, offer: Offer, include_bond: bool = False) -> str:
        """Format offer for announcement.

        Format: <ordertype> <oid> <minsize> <maxsize> <txfee> <cjfee>[!tbond <proof>]

        Args:
            offer: The offer to format
            include_bond: If True, append fidelity bond proof (for PRIVMSG only)

        Note:
            According to the JoinMarket protocol:
            - Public broadcasts: NO fidelity bond proof
            - Private responses to !orderbook: Include !tbond <proof>
        """

        order_type_str = offer.ordertype.value

        # NOTE: Don't include nick!PUBLIC! prefix here - send_public_message() adds it
        msg = (
            f"{order_type_str} "
            f"{offer.oid} {offer.minsize} {offer.maxsize} "
            f"{offer.txfee} {offer.cjfee}"
        )

        # Append fidelity bond proof ONLY for private responses
        if include_bond and self.fidelity_bond is not None:
            # For private response, we use the requesting taker's nick
            # The ownership signature proves we control the UTXO
            bond_proof = create_fidelity_bond_proof(
                bond=self.fidelity_bond,
                maker_nick=self.nick,
                taker_nick=self.nick,  # Will be updated when sending to specific taker
                current_block_height=self.current_block_height,
            )
            if bond_proof:
                msg += f"!tbond {bond_proof}"
                logger.debug(
                    f"Added fidelity bond proof to offer (proof length: {len(bond_proof)})"
                )

        return msg

Attributes

active_sessions: dict[str, MakerSession] = {} instance-attribute

backend = backend instance-attribute

config = config instance-attribute

current_block_height: int = 0 instance-attribute

current_offers: list[Offer] = [] instance-attribute

direct_connections: dict[str, TCPConnection] = {} instance-attribute

directory_clients: dict[str, DirectoryClient] = {} instance-attribute

fidelity_bond: FidelityBondInfo | None = None instance-attribute

hidden_service_listener: HiddenServiceListener | None = None instance-attribute

listen_tasks: list[asyncio.Task[None]] = [] instance-attribute

nick = self.nick_identity.nick instance-attribute

nick_identity = NickIdentity(JM_VERSION) instance-attribute

offer_manager = OfferManager(self.wallet, config, self.nick) instance-attribute

running = False instance-attribute

wallet = wallet instance-attribute

Methods:

__init__(wallet: WalletService, backend: BlockchainBackend, config: MakerConfig)

Source code in maker/src/maker/bot.py

def __init__(
    self,
    wallet: WalletService,
    backend: BlockchainBackend,
    config: MakerConfig,
):
    self.wallet = wallet
    self.backend = backend
    self.config = config

    # Create nick identity for signing messages
    self.nick_identity = NickIdentity(JM_VERSION)
    self.nick = self.nick_identity.nick

    self.offer_manager = OfferManager(self.wallet, config, self.nick)

    self.directory_clients: dict[str, DirectoryClient] = {}
    # Shared connection plumbing (parsing, SOCKS isolation creds,
    # DirectoryClient construction, retry loop). The pool stores its
    # connected clients in self.directory_clients via the shared-dict
    # binding below, so tests and existing call sites that mutate
    # self.directory_clients directly continue to work unchanged.
    self._directory_pool = MakerDirectoryPool(
        config=config,
        nick_identity=self.nick_identity,
        neutrino_compat=backend.can_provide_neutrino_metadata(),
    )
    self._directory_pool.clients = self.directory_clients
    self.active_sessions: dict[str, MakerSession] = {}
    self._reserved_commitments: set[str] = set()
    self.current_offers: list[Offer] = []
    self.fidelity_bond: FidelityBondInfo | None = None
    self.current_block_height: int = 0  # Cached block height for bond proof generation

    self.running = False
    self.listen_tasks: list[asyncio.Task[None]] = []
    self._session_cleanup_task: asyncio.Task[None] | None = None
    self._session_handler_task_count = 0
    self._detached_handler_tasks: set[asyncio.Task[None]] = set()
    self._pending_signed_rounds: dict[tuple[str, str], PendingSignedRound] = {}
    self._pending_signed_rounds_lock = asyncio.Lock()
    self._fatal_error: Exception | None = None

    # Session locks now live on each `MakerSession` (one asyncio.Lock per
    # taker_nick) so we no longer keep a parallel dict on the bot.

    # Hidden service listener for direct peer connections
    self.hidden_service_listener: HiddenServiceListener | None = None
    self.direct_connections: dict[str, TCPConnection] = {}
    self._direct_connection_states: dict[TCPConnection, DirectConnectionState] = {}

    # Tor control for dynamic hidden service creation
    self._tor_control: TorControlClient | None = None
    self._ephemeral_hidden_service: EphemeralHiddenService | None = None

    # Generic per-peer rate limiter (token bucket algorithm)
    # Generous burst (100 msgs) but low sustained rate (10 msg/s)
    self._message_rate_limiter = RateLimiter(
        rate_limit=config.message_rate_limit,
        burst_limit=config.message_burst_limit,
    )

    # Fidelity bond addresses loaded at startup, kept for periodic rescans so
    # newly funded bonds are detected without requiring a restart.
    self._fidelity_bond_addresses: list[tuple[str, int, int]] = []

    # Rate limiter for orderbook requests to prevent spam attacks
    self._orderbook_rate_limiter = OrderbookRateLimiter(
        rate_limit=config.orderbook_rate_limit,
        interval=config.orderbook_rate_interval,
        violation_ban_threshold=config.orderbook_violation_ban_threshold,
        violation_warning_threshold=config.orderbook_violation_warning_threshold,
        violation_severe_threshold=config.orderbook_violation_severe_threshold,
        ban_duration=config.orderbook_ban_duration,
    )

    # Rate limiter specifically for direct hidden service connections
    # This tracks by connection address (not nick) to prevent nick rotation attacks
    # where attackers use a different nick per request
    self._direct_connection_rate_limiter = DirectConnectionRateLimiter(
        message_rate_per_sec=5.0,  # Stricter than directory (5 msg/s vs 10)
        message_burst=20,  # Smaller burst
        orderbook_interval=30.0,  # Longer interval (30s vs 10s)
        orderbook_ban_threshold=10,  # Faster ban (10 violations vs 100)
        ban_duration=config.orderbook_ban_duration,
    )

    # Message deduplicator to handle receiving same message from multiple directories
    # This prevents processing duplicates and avoids false rate limit violations
    self._message_deduplicator = MessageDeduplicator(window_seconds=30.0)

    self._hp2_own_broadcast_semaphore = asyncio.Semaphore(1)
    self._hp2_relay_broadcast_semaphore = asyncio.Semaphore(1)

    # Track failed directory reconnection attempts
    # Key: node_id (host:port), Value: number of reconnection attempts
    self._directory_reconnect_attempts: dict[str, int] = {}

    # Track whether all directories were previously disconnected, so we can
    # send a recovery notification when at least one reconnects
    self._all_directories_disconnected: bool = False

    # Track CoinJoin txids we have already sent a mempool notification for,
    # so we do not re-notify on every pending-confirmation poll cycle.
    self._mempool_notified_txids: set[str] = set()

    # Track last log time for rate-limited logging
    # Key: log_key, Value: timestamp of last log
    self._rate_limited_log_times: dict[str, float] = {}

    # Own wallet nicks to exclude from CoinJoin sessions (self-CoinJoin protection)
    # Read the taker nick from state file if running both components from same wallet
    self._own_wallet_nicks: set[str] = set()
    taker_nick = read_nick_state(config.data_dir, "taker")
    if taker_nick:
        self._own_wallet_nicks.add(taker_nick)
        logger.info(f"Self-CoinJoin protection: excluding taker nick {taker_nick}")

start() -> None async

Start the maker bot.

Flow: 1. Initialize commitment blacklist 2. Sync wallet with blockchain 3. Create ephemeral hidden service if tor_control enabled 4. Connect to directory servers 5. Create and announce offers 6. Listen for taker requests

Source code in maker/src/maker/bot.py

async def start(self) -> None:
    """
    Start the maker bot.

    Flow:
    1. Initialize commitment blacklist
    2. Sync wallet with blockchain
    3. Create ephemeral hidden service if tor_control enabled
    4. Connect to directory servers
    5. Create and announce offers
    6. Listen for taker requests
    """
    try:
        logger.info(f"Starting maker bot (nick: {self.nick})")

        # Log wallet name if using descriptor wallet backend
        from jmwallet.backends.descriptor_wallet import DescriptorWalletBackend

        if isinstance(self.backend, DescriptorWalletBackend):
            logger.info(f"Using wallet: {self.backend.wallet_name}")

        # Initialize commitment blacklist with configured data directory
        set_blacklist_path(data_dir=self.config.data_dir)

        # Load fidelity bond addresses for optimized scanning
        # We scan wallet + fidelity bonds in a single pass to avoid extra
        # backend round-trips during sync.
        from jmcore.paths import get_default_data_dir
        from jmwallet.wallet.bond_registry import load_registry

        resolved_data_dir = (
            self.config.data_dir if self.config.data_dir else get_default_data_dir()
        )
        fidelity_bond_addresses: list[tuple[str, int, int]] = []

        # Fidelity bonds are explicitly disabled
        if self.config.no_fidelity_bond:
            logger.info(
                "Fidelity bonds disabled (--no-fidelity-bond). Running without bond proof."
            )
        # Option 1: Manual specification via fidelity_bond_index + locktimes (bypasses registry)
        # This is useful when running in Docker or when you don't have a registry yet
        elif (
            self.config.fidelity_bond_index is not None and self.config.fidelity_bond_locktimes
        ):
            logger.info(
                f"Using manual fidelity bond specification: "
                f"locktimes={self.config.fidelity_bond_locktimes}"
            )
            for locktime in self.config.fidelity_bond_locktimes:
                from jmcore.timenumber import timestamp_to_timenumber

                timenumber = timestamp_to_timenumber(locktime)
                address = self.wallet.get_fidelity_bond_address(timenumber, locktime)
                fidelity_bond_addresses.append((address, locktime, timenumber))
                logger.info(
                    f"Generated fidelity bond address for locktime {locktime}: {address}"
                )
        # Option 2: Load from registry (default)
        else:
            bond_registry = load_registry(
                resolved_data_dir,
                self.wallet.wallet_fingerprint,
                allow_legacy_fallback=False,
            )
            network_bonds = [
                bond for bond in bond_registry.bonds if bond.network == self.config.network
            ]
            if network_bonds:
                # Extract (address, locktime, index) tuples from registry
                fidelity_bond_addresses = [
                    (bond.address, bond.locktime, bond.index) for bond in network_bonds
                ]
                logger.info(
                    f"Loaded {len(fidelity_bond_addresses)} "
                    f"fidelity bond address(es) from registry"
                )

        logger.info("Syncing wallet and fidelity bonds...")

        # Store bond addresses on the instance so periodic rescans can use them
        # to detect newly funded bonds without requiring a restart.
        self._fidelity_bond_addresses = fidelity_bond_addresses

        # Setup descriptor wallet if needed (one-time operation)
        if isinstance(self.backend, DescriptorWalletBackend):
            # Check if base wallet is set up (without counting bonds)
            base_wallet_ready = await self.wallet.is_descriptor_wallet_ready(
                fidelity_bond_count=0
            )
            # Check if wallet with bonds is set up
            full_wallet_ready = await self.wallet.is_descriptor_wallet_ready(
                fidelity_bond_count=len(fidelity_bond_addresses)
            )

            if not base_wallet_ready:
                # First time setup - import everything including bonds
                logger.info("Descriptor wallet not set up. Importing descriptors...")
                await self.wallet.setup_descriptor_wallet(
                    rescan=True,
                    fidelity_bond_addresses=fidelity_bond_addresses,
                )
                logger.info("Descriptor wallet setup complete")
            elif not full_wallet_ready and fidelity_bond_addresses:
                # Base wallet exists but bonds are missing - import just the bonds
                logger.info(
                    "Descriptor wallet exists but fidelity bond addresses not imported. "
                    "Importing bond addresses..."
                )
                await self.wallet.import_fidelity_bond_addresses(
                    fidelity_bond_addresses, rescan=True
                )

            # Use fast descriptor wallet sync
            await self.wallet.sync_with_descriptor_wallet(fidelity_bond_addresses)
        else:
            # Use standard sync (BIP157/158 for neutrino, mempool API, etc.)
            await self.wallet.sync_all(fidelity_bond_addresses)
        await self.wallet.reconstruct_imported_state_safe()

        # Update bond registry with UTXO info from the scan (only if using registry)
        if self.config.fidelity_bond_index is None and fidelity_bond_addresses:
            from jmwallet.wallet.bond_registry import save_registry

            bond_registry = load_registry(
                resolved_data_dir,
                self.wallet.wallet_fingerprint,
                allow_legacy_fallback=False,
            )
            for bond in bond_registry.bonds:
                # Find the UTXO for this bond address in mixdepth 0
                bond_utxo = next(
                    (
                        utxo
                        for utxo in self.wallet.utxo_cache.get(0, [])
                        if utxo.address == bond.address
                    ),
                    None,
                )
                if bond_utxo:
                    # Update the bond registry with UTXO info
                    bond.txid = bond_utxo.txid
                    bond.vout = bond_utxo.vout
                    bond.value = bond_utxo.value
                    bond.confirmations = bond_utxo.confirmations
                    logger.debug(
                        f"Updated bond {bond.address[:20]}... with UTXO "
                        f"{bond_utxo.txid[:16]}...:{bond_utxo.vout}, value={bond_utxo.value}"
                    )

            # Save updated registry
            save_registry(bond_registry, resolved_data_dir, self.wallet.wallet_fingerprint)

        # Get current block height for bond proof generation
        self.current_block_height = await self.backend.get_block_height()
        logger.debug(f"Current block height: {self.current_block_height}")

        total_balance = await self.wallet.get_total_balance()
        logger.info(f"Wallet synced. Total balance: {total_balance:,} sats")

        # Find fidelity bond for proof generation
        # If a specific bond is selected in config, use it; otherwise use the best one
        if self.config.no_fidelity_bond:
            self.fidelity_bond = None
            logger.warning("Fidelity bond disabled (offers will have no bond proof)")
        elif self.config.selected_fidelity_bond:
            # User specified a specific bond
            sel_txid, sel_vout = self.config.selected_fidelity_bond
            bonds = await find_fidelity_bonds(self.wallet)
            self.fidelity_bond = next(
                (b for b in bonds if b.txid == sel_txid and b.vout == sel_vout), None
            )
            if self.fidelity_bond:
                logger.info(
                    f"Using selected fidelity bond: {sel_txid[:16]}...:{sel_vout}, "
                    f"value={self.fidelity_bond.value:,} sats, "
                    f"bond_value={self.fidelity_bond.bond_value:,}"
                )
            else:
                logger.warning(
                    f"Selected fidelity bond {sel_txid[:16]}...:{sel_vout} not found, "
                    "falling back to best available"
                )
                self.fidelity_bond = await get_best_fidelity_bond(
                    self.wallet, current_block_height=self.current_block_height
                )
        else:
            # Auto-select the best (largest bond value) fidelity bond
            self.fidelity_bond = await get_best_fidelity_bond(
                self.wallet, current_block_height=self.current_block_height
            )
        if self.fidelity_bond:
            ensure_fidelity_bond_certificate_valid(
                self.fidelity_bond,
                self.current_block_height,
            )
            logger.info(
                f"Fidelity bond found: {self.fidelity_bond.txid[:16]}..., "
                f"value={self.fidelity_bond.value:,} sats, "
                f"bond_value={self.fidelity_bond.bond_value:,}"
            )
            md0_utxos = _get_fidelity_bond_linkable_utxos(
                self.wallet,
                resolved_data_dir,
            )
            if md0_utxos:
                total_md0 = sum(u.value for u in md0_utxos)
                logger.warning(
                    f"PRIVACY RISK: You have a fidelity bond AND "
                    f"{len(md0_utxos)} regular UTXO(s) ({total_md0:,} sats) "
                    f"in mixdepth 0.\n"
                    f"Using md0 UTXOs in coinjoins can link your identity "
                    f"to your fidelity bond.\n"
                    f"Recommendation: sweep md0 funds to mixdepth 1 as a "
                    f"taker coinjoin, then freeze or spend the md0 UTXOs."
                )
        else:
            logger.warning("No fidelity bond found (offers will have no bond proof)")

        logger.info("Creating offers...")
        self.current_offers = await self.offer_manager.create_offers()

        # If no offers due to insufficient balance, wait and retry
        retry_count = 0
        max_retries = 30  # 5 minutes max wait (30 * 10s)
        while not self.current_offers and retry_count < max_retries:
            retry_count += 1
            logger.warning(
                f"No offers created (insufficient balance?). "
                f"Waiting 10s and retrying... (attempt {retry_count}/{max_retries})"
            )
            await asyncio.sleep(10)

            # Re-sync wallet to check for new funds
            from jmwallet.backends.descriptor_wallet import DescriptorWalletBackend

            if isinstance(self.backend, DescriptorWalletBackend):
                await self.wallet.sync_with_descriptor_wallet()
            else:
                await self.wallet.sync_all()
            await self.wallet.reconstruct_imported_state_safe()
            total_balance = await self.wallet.get_total_balance()
            logger.info(f"Wallet re-synced. Total balance: {total_balance:,} sats")

            self.current_offers = await self.offer_manager.create_offers()

        if not self.current_offers:
            logger.error(
                f"No offers created after {max_retries} retries. "
                "Please fund the wallet and restart."
            )
            return

        # Log summary of created offers
        logger.info(f"Created {len(self.current_offers)} offer(s) to announce:")
        for offer in self.current_offers:
            fee_display = (
                f"{float(offer.cjfee) * 100:.4f}%"
                if offer.ordertype.value.endswith("reloffer")
                else f"{offer.cjfee} sats"
            )
            logger.info(
                f"  oid={offer.oid}: {offer.ordertype.value}, "
                f"size={offer.minsize:,}-{offer.maxsize:,} sats, fee={fee_display}"
            )

        # Set up ephemeral hidden service via Tor control port if enabled
        # This must happen before connecting to directory servers so we can
        # advertise the onion address
        ephemeral_onion = await self._setup_tor_hidden_service()
        if ephemeral_onion:
            # Override onion_host with the dynamically created one
            object.__setattr__(self.config, "onion_host", ephemeral_onion)
            logger.info(f"Using ephemeral onion address: {ephemeral_onion}")

        # Determine the onion address to advertise
        onion_host = self.config.onion_host

        logger.info("Connecting to directory servers...")
        await self._connect_to_directories_with_retry()

        # Start hidden service listener if we have an onion address (static or ephemeral)
        if onion_host:
            logger.info(
                f"Starting hidden service listener on "
                f"{self.config.onion_serving_host}:{self.config.onion_serving_port}..."
            )
            self.hidden_service_listener = HiddenServiceListener(
                host=self.config.onion_serving_host,
                port=self.config.onion_serving_port,
                on_connection=self._on_direct_connection,
            )
            await self.hidden_service_listener.start()
            logger.info(f"Hidden service listener started (onion: {onion_host})")

        logger.info("Announcing offers...")
        await self._announce_offers()

        logger.info("Maker bot started. Listening for takers...")
        self.running = True

        self._start_session_cleanup_task()

        # Start listening on all directory clients
        for node_id, client in self.directory_clients.items():
            task = asyncio.create_task(self._listen_client(node_id, client))
            self.listen_tasks.append(task)

        # If hidden service listener is running, start serve_forever task
        if self.hidden_service_listener:
            task = asyncio.create_task(self.hidden_service_listener.serve_forever())
            self.listen_tasks.append(task)

        # Start background task to monitor pending transactions
        monitor_task = asyncio.create_task(self._monitor_pending_transactions())
        self.listen_tasks.append(monitor_task)

        # Start periodic wallet rescan task
        rescan_task = asyncio.create_task(self._periodic_rescan())
        self.listen_tasks.append(rescan_task)

        # Start periodic rate limit status logging task
        status_task = asyncio.create_task(self._periodic_rate_limit_status())
        self.listen_tasks.append(status_task)

        # Start periodic directory connection status logging task
        conn_status_task = asyncio.create_task(self._periodic_directory_connection_status())
        self.listen_tasks.append(conn_status_task)

        # Start periodic directory reconnection task
        reconnect_task = asyncio.create_task(self._periodic_directory_reconnect())
        self.listen_tasks.append(reconnect_task)

        # Start periodic summary notification task (if enabled)
        notifier = get_notifier()
        if notifier.config.notify_summary:
            summary_task = asyncio.create_task(self._periodic_summary())
            self.listen_tasks.append(summary_task)
        else:
            logger.info("Periodic summary notifications disabled (notify_summary=false)")

        # Wait for all listening tasks to complete
        await asyncio.gather(*self.listen_tasks, return_exceptions=True)
        if self._fatal_error is not None:
            raise self._fatal_error

    except ExpiredFidelityBondCertificateError:
        raise
    except Exception as e:
        logger.error(f"Failed to start maker bot: {e}")
        raise

stop() -> None async

Stop the maker bot

Source code in maker/src/maker/bot.py

async def stop(self) -> None:
    """Stop the maker bot"""
    logger.info("Stopping maker bot...")
    self.running = False

    # Stop the dedicated reaper, then independently expire every exact
    # session object. Handler cancellation and cleanup are bounded inside
    # _expire_timed_out_session, including cancellation-suppressing tasks.
    cleanup_task = self._session_cleanup_task
    if cleanup_task is not None:
        cleanup_task.cancel()
        await asyncio.wait({cleanup_task}, timeout=1.0)

    expiration_tasks = [
        asyncio.create_task(self._expire_timed_out_session(nick, session))
        for nick, session in list(self.active_sessions.items())
    ]
    if expiration_tasks:
        _, pending_expirations = await asyncio.wait(expiration_tasks, timeout=1.5)
        for expiration_task in pending_expirations:
            expiration_task.cancel()

    await self._drain_pending_signed_rounds()

    # Cancellation-suppressing handlers remain strongly referenced until
    # they actually finish. Shutdown sends another cancellation request and
    # waits only for a bounded grace period.
    detached_tasks = set(self._detached_handler_tasks)
    for detached_task in detached_tasks:
        detached_task.cancel()
    if detached_tasks:
        _, pending_detached = await asyncio.wait(
            detached_tasks, timeout=_DETACHED_SHUTDOWN_GRACE_SEC
        )
        if pending_detached:
            logger.warning(
                f"{len(pending_detached)} detached maker handler task(s) "
                "still suppress cancellation after shutdown grace period"
            )

    # Cancel all listening tasks after detached handlers have released
    # their directory dispatchers.
    for listener_task in self.listen_tasks:
        listener_task.cancel()

    if self.listen_tasks:
        _, pending_listeners = await asyncio.wait(set(self.listen_tasks), timeout=2.0)
        for listener_task in pending_listeners:
            listener_task.cancel()
    self.listen_tasks.clear()
    self._session_cleanup_task = None

    # Stop hidden service listener
    if self.hidden_service_listener:
        await self.hidden_service_listener.stop()

    # Clean up Tor control connection (ephemeral hidden service auto-removed)
    await self._cleanup_tor_hidden_service()

    # Close every accepted direct socket, including sockets that never
    # completed a handshake or authenticated a sender.
    direct_sockets = set(self._direct_connection_states)
    direct_sockets.update(self.direct_connections.values())
    for conn in direct_sockets:
        try:
            await conn.close()
        except Exception:
            pass
    self.direct_connections.clear()
    self._direct_connection_states.clear()

    # Close all directory clients
    for client in self.directory_clients.values():
        try:
            await client.close()
        except Exception:
            pass

    # Do not close the wallet here as it might be shared (e.g. in jmwalletd)
    # The caller is responsible for managing the wallet lifecycle.
    # await self.wallet.close()
    logger.info("Maker bot stopped")

Functions: