jmcore.directory_pool

jmcore.directory_pool

Shared directory-client pool primitives for maker and taker.

This module provides :class:DirectoryClientPool, a base class that owns the boilerplate around connecting to one or more directory servers over Tor:

• Parsing a configured "host:port" server string into a canonical node_id (used as dict key in both maker and taker).
• Wiring up SOCKS stream isolation credentials when enabled.
• Building, connecting, and tearing down :class:DirectoryClient instances.
• A bounded connect_all_with_retry loop suitable for startup, where Tor may still be bootstrapping circuits, plus a reusable reconnect_disconnected cycle for periodic background reconnects.

Component-specific behavior (post-connect offer announcement for makers, direct-peer connections and nick tracking for takers) is layered on by subclassing and overriding hooks:

• :meth:_build_client_kwargs -- supply per-component constructor kwargs for DirectoryClient (e.g. location, neutrino_compat).
• :meth:_on_directory_connected -- run after each successful connect (e.g. start a listener task, announce offers).
• :meth:_on_directory_disconnected -- cleanup hook.

The pool deliberately stays unaware of higher-level concerns like message dispatch, response correlation, or direct peer connections; those continue to live in the subclasses that own them.

Attributes

DEFAULT_SETTLE_GRACE_SEC = 5.0 module-attribute

Classes

DirectoryClientPool

Manages a pool of :class:DirectoryClient connections over Tor.

Subclasses customize per-connection construction by overriding :meth:_build_client_kwargs, and react to lifecycle transitions by overriding :meth:_on_directory_connected / :meth:_on_directory_disconnected.

Attributes: directory_servers: Configured "host:port" addresses to connect to. network: Bitcoin network ("mainnet", "testnet", "signet", "regtest"). nick_identity: Long-lived identity used for directory handshake. socks_host, socks_port: Tor SOCKS proxy. connection_timeout: Per-connection handshake timeout, in seconds. stream_isolation: When True, use distinct SOCKS credentials per isolation category (directory, peer) so Tor opens fresh circuits and the directories cannot link our flows. clients: Connected clients keyed by canonical "host:port" node_id. Subclasses may iterate this dict but should not mutate it directly; use :meth:connect_all_with_retry, :meth:reconnect_disconnected, or :meth:close_all.

Source code in jmcore/src/jmcore/directory_pool.py

class DirectoryClientPool:
    """
    Manages a pool of :class:`DirectoryClient` connections over Tor.

    Subclasses customize per-connection construction by overriding
    :meth:`_build_client_kwargs`, and react to lifecycle transitions by
    overriding :meth:`_on_directory_connected` /
    :meth:`_on_directory_disconnected`.

    Attributes:
        directory_servers: Configured "host:port" addresses to connect to.
        network: Bitcoin network ("mainnet", "testnet", "signet", "regtest").
        nick_identity: Long-lived identity used for directory handshake.
        socks_host, socks_port: Tor SOCKS proxy.
        connection_timeout: Per-connection handshake timeout, in seconds.
        stream_isolation: When True, use distinct SOCKS credentials per
            isolation category (directory, peer) so Tor opens fresh
            circuits and the directories cannot link our flows.
        clients: Connected clients keyed by canonical ``"host:port"``
            ``node_id``. Subclasses may iterate this dict but should not
            mutate it directly; use :meth:`connect_all_with_retry`,
            :meth:`reconnect_disconnected`, or :meth:`close_all`.
    """

    def __init__(
        self,
        *,
        directory_servers: list[str],
        network: str,
        nick_identity: NickIdentity,
        socks_host: str = "127.0.0.1",
        socks_port: int = 9050,
        connection_timeout: float = 120.0,
        stream_isolation: bool = False,
        nick_auth_mode: NickAuthMode = NickAuthMode.PREFER_VERIFIED,
        nick_auth_directory_ids: dict[str, str] | None = None,
    ):
        self.directory_servers = directory_servers
        self.network = network
        self.nick_identity = nick_identity
        self.socks_host = socks_host
        self.socks_port = socks_port
        self.connection_timeout = connection_timeout
        self.stream_isolation = stream_isolation
        self.nick_auth_mode = nick_auth_mode
        self.nick_auth_directory_ids = nick_auth_directory_ids or {}

        # Pre-compute isolation credentials (None when disabled). The peer
        # credentials are exposed for subclasses that establish direct
        # onion-to-onion connections (taker), even though the pool itself
        # only uses _dir_creds.
        self._dir_creds: tuple[str | None, str | None] = (None, None)
        self._peer_creds: tuple[str | None, str | None] = (None, None)
        if stream_isolation:
            dir_c = get_isolation_credentials(IsolationCategory.DIRECTORY)
            self._dir_creds = (dir_c.username, dir_c.password)
            peer_c = get_isolation_credentials(IsolationCategory.PEER)
            self._peer_creds = (peer_c.username, peer_c.password)

        self.clients: dict[str, DirectoryClient] = {}
        self.last_connection_result = DirectoryConnectionResult(
            connected=0, total=len(directory_servers)
        )

    # -- Hooks ----------------------------------------------------------

    def _build_client_kwargs(self, host: str, port: int) -> dict[str, Any]:
        """
        Return ``DirectoryClient`` constructor kwargs for one server.

        Subclasses may override to inject component-specific arguments
        such as ``location`` (maker's onion host:port) or
        ``neutrino_compat``. The base implementation supplies only the
        universally required arguments.
        """
        return {
            "host": host,
            "port": port,
            "network": self.network,
            "nick_identity": self.nick_identity,
            "socks_host": self.socks_host,
            "socks_port": self.socks_port,
            "timeout": self.connection_timeout,
            "nick_auth_mode": self.nick_auth_mode,
            "nick_auth_directory_id": self.nick_auth_directory_ids.get(f"{host}:{port}"),
            "socks_username": self._dir_creds[0],
            "socks_password": self._dir_creds[1],
        }

    async def _on_directory_connected(self, node_id: str, client: DirectoryClient) -> None:
        """
        Called after a successful connect for ``node_id``.

        Subclasses may override to start a listener task, announce
        offers, or emit notifications. The default is a no-op.
        """

    async def _on_directory_disconnected(self, node_id: str) -> None:
        """Called when a directory has been removed from the pool. Default no-op."""

    # -- Single-server connect ------------------------------------------

    async def connect_to_directory(self, dir_server: str) -> tuple[str, DirectoryClient] | None:
        """
        Connect to a single directory server.

        Args:
            dir_server: Address as ``"host"`` or ``"host:port"``.

        Returns:
            ``(node_id, client)`` on success, or ``None`` if the address
            could not be parsed or the connect/handshake failed. The
            returned client is not yet stored in ``self.clients``; the
            caller (or :meth:`connect_all_with_retry` /
            :meth:`reconnect_disconnected`) handles registration.
        """
        try:
            host, port = parse_directory_address(dir_server)
        except Exception as e:
            logger.bind(sensitive=True).debug(f"Cannot parse directory address {dir_server!r}: {e}")
            return None

        node_id = f"{host}:{port}"

        client: DirectoryClient | None = None
        try:
            kwargs = self._build_client_kwargs(host, port)
            client = DirectoryClient(**kwargs)
            await client.connect()
            return (node_id, client)
        except asyncio.CancelledError:
            # Startup may finish while this client's handshake is still pending.
            # DirectoryClient.connect() only cleans up ordinary exceptions.
            if client is not None:
                with contextlib.suppress(Exception):
                    await client.close()
            raise
        except Exception as e:
            logger.bind(sensitive=True).debug(f"Failed to connect to {dir_server}: {e}")
            return None

    # -- Bulk connect / reconnect ---------------------------------------

    async def connect_all_parallel(self) -> int:
        """
        Connect to every configured server in parallel (single pass).

        On success, the client is added to ``self.clients`` and the
        ``_on_directory_connected`` hook fires. Suitable for callers
        that do not want a retry loop (e.g. taker, which only runs once
        per CoinJoin).

        Returns the number of successful connections. The aggregate result,
        including the configured total and failed count, is retained in
        :attr:`last_connection_result`.
        """
        tasks = [self.connect_to_directory(server) for server in self.directory_servers]
        results = await asyncio.gather(*tasks)

        connected = 0
        for result in results:
            if result is None:
                continue
            node_id, client = result
            self.clients[node_id] = client
            await self._on_directory_connected(node_id, client)
            connected += 1
        self.last_connection_result = DirectoryConnectionResult(
            connected=connected,
            total=len(self.directory_servers),
        )
        return connected

    async def connect_all_with_retry(
        self,
        *,
        timeout: float,
        initial_delay: float = 5.0,
        max_delay: float = 30.0,
        backoff: float = 1.5,
        settle_grace: float = DEFAULT_SETTLE_GRACE_SEC,
    ) -> int:
        """
        Connect to configured servers concurrently with bounded retry.

        Tor may still be bootstrapping when the caller starts; this loop
        keeps retrying failed servers until either at least one client
        connects or ``timeout`` elapses, whichever comes first. After
        timeout the method returns gracefully so callers can rely on a
        separate periodic reconnect task. Retain completed successes, keep
        waiting for the remaining attempts for at most ``settle_grace``
        seconds after the first success (still bounded by ``timeout``), then
        cancel what is left and let periodic reconnection fill in the
        remaining directories. The grace covers attempts whose Tor circuit is
        already up and are mid-handshake; a slow or dead endpoint must not
        prevent callers from listening on a ready connection.

        Args:
            timeout: Hard deadline in seconds across all attempts.
            initial_delay: Sleep between full retry passes when no
                connection has succeeded yet.
            max_delay: Cap on the exponentially-backed-off delay.
            backoff: Multiplier applied to ``initial_delay`` per pass.
            settle_grace: Extra seconds to wait for in-flight attempts after
                the first success.

        Returns:
            Number of clients currently in ``self.clients`` when the
            method returns.
        """
        loop = asyncio.get_running_loop()
        deadline = loop.time() + timeout
        delay = initial_delay
        attempt = 0

        while loop.time() < deadline:
            attempt += 1
            # Aliases such as host and host:5222 must not race to replace one
            # another, leaving an authenticated but unowned connection behind.
            missing = {node_id: server for server, node_id in self.list_disconnected()}
            tasks = {
                asyncio.create_task(self.connect_to_directory(server)): server
                for server in missing.values()
            }
            pending = set(tasks)
            adopted: set[asyncio.Task[tuple[str, DirectoryClient] | None]] = set()
            settle_deadline: float | None = None
            try:
                while pending:
                    wait_until = deadline if settle_deadline is None else settle_deadline
                    done, pending = await asyncio.wait(
                        pending,
                        timeout=max(0.0, wait_until - loop.time()),
                        return_when=asyncio.FIRST_COMPLETED,
                    )
                    # Preserve configured ordering among simultaneous results.
                    for task, server in tasks.items():
                        if task not in done:
                            continue
                        result = task.result()
                        if result is None:
                            logger.bind(sensitive=True).warning(
                                f"Could not connect to {server} (attempt {attempt})"
                            )
                            continue
                        node_id, client = result
                        self.clients[node_id] = client
                        try:
                            await self._on_directory_connected(node_id, client)
                        except BaseException:
                            self.clients.pop(node_id, None)
                            raise
                        adopted.add(task)
                        logger.bind(sensitive=True).info(f"Connected to directory: {server}")
                    if self.clients and settle_deadline is None:
                        settle_deadline = min(deadline, loop.time() + settle_grace)
                    if loop.time() >= wait_until:
                        break
            finally:
                for task in tasks:
                    if not task.done():
                        task.cancel()
                results = await asyncio.gather(*tasks, return_exceptions=True)
                # A connect can complete just as the caller cancels or a hook
                # yields. Close every successful result we did not adopt.
                for task, outcome in zip(tasks, results, strict=True):
                    if task not in adopted and isinstance(outcome, tuple):
                        with contextlib.suppress(Exception):
                            await outcome[1].close()

            if self.clients:
                return len(self.clients)

            remaining = deadline - loop.time()
            if remaining <= 0:
                break

            wait = min(delay, remaining)
            logger.info(f"Retrying directory connections in {wait:.0f}s...")
            await asyncio.sleep(wait)
            delay = min(delay * backoff, max_delay)

        if not self.clients:
            logger.error(
                f"Failed to connect to any directory server after {timeout}s. "
                "Caller should rely on the periodic reconnect task."
            )
        return len(self.clients)

    def list_disconnected(self) -> list[tuple[str, str]]:
        """
        Return ``(dir_server, node_id)`` pairs for configured servers
        not currently in ``self.clients``.

        Servers whose addresses fail to parse are skipped (with a debug
        log) rather than reported as disconnected, since the user almost
        certainly mistyped them and re-trying won't help.
        """
        connected = set(self.clients.keys())
        out: list[tuple[str, str]] = []
        for server in self.directory_servers:
            try:
                host, port = parse_directory_address(server)
            except Exception as e:
                logger.bind(sensitive=True).debug(f"Skipping unparseable directory {server!r}: {e}")
                continue
            node_id = f"{host}:{port}"
            if node_id not in connected:
                out.append((server, node_id))
        return out

    async def reconnect_disconnected(self) -> list[tuple[str, DirectoryClient]]:
        """
        Attempt to reconnect to every configured-but-disconnected server.

        Newly connected clients are added to ``self.clients`` and
        ``_on_directory_connected`` fires for each. The return value is
        the list of ``(node_id, client)`` for callers that need to do
        additional per-reconnect work (e.g. start a listener task).
        """
        newly_connected: list[tuple[str, DirectoryClient]] = []
        for dir_server, _expected_node_id in self.list_disconnected():
            result = await self.connect_to_directory(dir_server)
            if result is None:
                continue
            node_id, client = result
            self.clients[node_id] = client
            logger.bind(sensitive=True).info(f"Reconnected to directory: {dir_server}")
            await self._on_directory_connected(node_id, client)
            newly_connected.append((node_id, client))
        return newly_connected

    # -- Teardown -------------------------------------------------------

    async def close_all(self) -> None:
        """
        Close every directory client and fire the disconnected hook for
        each.

        Errors during individual closes are logged at warning level but
        do not abort the shutdown.
        """
        for node_id, client in list(self.clients.items()):
            try:
                await client.close()
            except Exception as e:
                logger.warning("Error closing directory connection")
                logger.bind(sensitive=True).warning(f"Error closing connection to {node_id}: {e}")
            try:
                await self._on_directory_disconnected(node_id)
            except Exception as e:
                logger.warning("Directory disconnect hook failed")
                logger.bind(sensitive=True).warning(f"Error in disconnect hook for {node_id}: {e}")
        self.clients.clear()

Attributes

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

connection_timeout = connection_timeout instance-attribute

directory_servers = directory_servers instance-attribute

last_connection_result = DirectoryConnectionResult(connected=0, total=len(directory_servers)) instance-attribute

network = network instance-attribute

nick_auth_directory_ids = nick_auth_directory_ids or {} instance-attribute

nick_auth_mode = nick_auth_mode instance-attribute

nick_identity = nick_identity instance-attribute

socks_host = socks_host instance-attribute

socks_port = socks_port instance-attribute

stream_isolation = stream_isolation instance-attribute

Methods:

__init__(*, directory_servers: list[str], network: str, nick_identity: NickIdentity, socks_host: str = '127.0.0.1', socks_port: int = 9050, connection_timeout: float = 120.0, stream_isolation: bool = False, nick_auth_mode: NickAuthMode = NickAuthMode.PREFER_VERIFIED, nick_auth_directory_ids: dict[str, str] | None = None)

Source code in jmcore/src/jmcore/directory_pool.py

def __init__(
    self,
    *,
    directory_servers: list[str],
    network: str,
    nick_identity: NickIdentity,
    socks_host: str = "127.0.0.1",
    socks_port: int = 9050,
    connection_timeout: float = 120.0,
    stream_isolation: bool = False,
    nick_auth_mode: NickAuthMode = NickAuthMode.PREFER_VERIFIED,
    nick_auth_directory_ids: dict[str, str] | None = None,
):
    self.directory_servers = directory_servers
    self.network = network
    self.nick_identity = nick_identity
    self.socks_host = socks_host
    self.socks_port = socks_port
    self.connection_timeout = connection_timeout
    self.stream_isolation = stream_isolation
    self.nick_auth_mode = nick_auth_mode
    self.nick_auth_directory_ids = nick_auth_directory_ids or {}

    # Pre-compute isolation credentials (None when disabled). The peer
    # credentials are exposed for subclasses that establish direct
    # onion-to-onion connections (taker), even though the pool itself
    # only uses _dir_creds.
    self._dir_creds: tuple[str | None, str | None] = (None, None)
    self._peer_creds: tuple[str | None, str | None] = (None, None)
    if stream_isolation:
        dir_c = get_isolation_credentials(IsolationCategory.DIRECTORY)
        self._dir_creds = (dir_c.username, dir_c.password)
        peer_c = get_isolation_credentials(IsolationCategory.PEER)
        self._peer_creds = (peer_c.username, peer_c.password)

    self.clients: dict[str, DirectoryClient] = {}
    self.last_connection_result = DirectoryConnectionResult(
        connected=0, total=len(directory_servers)
    )

close_all() -> None async

Close every directory client and fire the disconnected hook for each.

Errors during individual closes are logged at warning level but do not abort the shutdown.

Source code in jmcore/src/jmcore/directory_pool.py

async def close_all(self) -> None:
    """
    Close every directory client and fire the disconnected hook for
    each.

    Errors during individual closes are logged at warning level but
    do not abort the shutdown.
    """
    for node_id, client in list(self.clients.items()):
        try:
            await client.close()
        except Exception as e:
            logger.warning("Error closing directory connection")
            logger.bind(sensitive=True).warning(f"Error closing connection to {node_id}: {e}")
        try:
            await self._on_directory_disconnected(node_id)
        except Exception as e:
            logger.warning("Directory disconnect hook failed")
            logger.bind(sensitive=True).warning(f"Error in disconnect hook for {node_id}: {e}")
    self.clients.clear()

connect_all_parallel() -> int async

Connect to every configured server in parallel (single pass).

On success, the client is added to self.clients and the _on_directory_connected hook fires. Suitable for callers that do not want a retry loop (e.g. taker, which only runs once per CoinJoin).

Returns the number of successful connections. The aggregate result, including the configured total and failed count, is retained in :attr:last_connection_result.

Source code in jmcore/src/jmcore/directory_pool.py

async def connect_all_parallel(self) -> int:
    """
    Connect to every configured server in parallel (single pass).

    On success, the client is added to ``self.clients`` and the
    ``_on_directory_connected`` hook fires. Suitable for callers
    that do not want a retry loop (e.g. taker, which only runs once
    per CoinJoin).

    Returns the number of successful connections. The aggregate result,
    including the configured total and failed count, is retained in
    :attr:`last_connection_result`.
    """
    tasks = [self.connect_to_directory(server) for server in self.directory_servers]
    results = await asyncio.gather(*tasks)

    connected = 0
    for result in results:
        if result is None:
            continue
        node_id, client = result
        self.clients[node_id] = client
        await self._on_directory_connected(node_id, client)
        connected += 1
    self.last_connection_result = DirectoryConnectionResult(
        connected=connected,
        total=len(self.directory_servers),
    )
    return connected

connect_all_with_retry(*, timeout: float, initial_delay: float = 5.0, max_delay: float = 30.0, backoff: float = 1.5, settle_grace: float = DEFAULT_SETTLE_GRACE_SEC) -> int async

Connect to configured servers concurrently with bounded retry.

Tor may still be bootstrapping when the caller starts; this loop keeps retrying failed servers until either at least one client connects or timeout elapses, whichever comes first. After timeout the method returns gracefully so callers can rely on a separate periodic reconnect task. Retain completed successes, keep waiting for the remaining attempts for at most settle_grace seconds after the first success (still bounded by timeout), then cancel what is left and let periodic reconnection fill in the remaining directories. The grace covers attempts whose Tor circuit is already up and are mid-handshake; a slow or dead endpoint must not prevent callers from listening on a ready connection.

Args: timeout: Hard deadline in seconds across all attempts. initial_delay: Sleep between full retry passes when no connection has succeeded yet. max_delay: Cap on the exponentially-backed-off delay. backoff: Multiplier applied to initial_delay per pass. settle_grace: Extra seconds to wait for in-flight attempts after the first success.

Returns: Number of clients currently in self.clients when the method returns.

Source code in jmcore/src/jmcore/directory_pool.py

async def connect_all_with_retry(
    self,
    *,
    timeout: float,
    initial_delay: float = 5.0,
    max_delay: float = 30.0,
    backoff: float = 1.5,
    settle_grace: float = DEFAULT_SETTLE_GRACE_SEC,
) -> int:
    """
    Connect to configured servers concurrently with bounded retry.

    Tor may still be bootstrapping when the caller starts; this loop
    keeps retrying failed servers until either at least one client
    connects or ``timeout`` elapses, whichever comes first. After
    timeout the method returns gracefully so callers can rely on a
    separate periodic reconnect task. Retain completed successes, keep
    waiting for the remaining attempts for at most ``settle_grace``
    seconds after the first success (still bounded by ``timeout``), then
    cancel what is left and let periodic reconnection fill in the
    remaining directories. The grace covers attempts whose Tor circuit is
    already up and are mid-handshake; a slow or dead endpoint must not
    prevent callers from listening on a ready connection.

    Args:
        timeout: Hard deadline in seconds across all attempts.
        initial_delay: Sleep between full retry passes when no
            connection has succeeded yet.
        max_delay: Cap on the exponentially-backed-off delay.
        backoff: Multiplier applied to ``initial_delay`` per pass.
        settle_grace: Extra seconds to wait for in-flight attempts after
            the first success.

    Returns:
        Number of clients currently in ``self.clients`` when the
        method returns.
    """
    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout
    delay = initial_delay
    attempt = 0

    while loop.time() < deadline:
        attempt += 1
        # Aliases such as host and host:5222 must not race to replace one
        # another, leaving an authenticated but unowned connection behind.
        missing = {node_id: server for server, node_id in self.list_disconnected()}
        tasks = {
            asyncio.create_task(self.connect_to_directory(server)): server
            for server in missing.values()
        }
        pending = set(tasks)
        adopted: set[asyncio.Task[tuple[str, DirectoryClient] | None]] = set()
        settle_deadline: float | None = None
        try:
            while pending:
                wait_until = deadline if settle_deadline is None else settle_deadline
                done, pending = await asyncio.wait(
                    pending,
                    timeout=max(0.0, wait_until - loop.time()),
                    return_when=asyncio.FIRST_COMPLETED,
                )
                # Preserve configured ordering among simultaneous results.
                for task, server in tasks.items():
                    if task not in done:
                        continue
                    result = task.result()
                    if result is None:
                        logger.bind(sensitive=True).warning(
                            f"Could not connect to {server} (attempt {attempt})"
                        )
                        continue
                    node_id, client = result
                    self.clients[node_id] = client
                    try:
                        await self._on_directory_connected(node_id, client)
                    except BaseException:
                        self.clients.pop(node_id, None)
                        raise
                    adopted.add(task)
                    logger.bind(sensitive=True).info(f"Connected to directory: {server}")
                if self.clients and settle_deadline is None:
                    settle_deadline = min(deadline, loop.time() + settle_grace)
                if loop.time() >= wait_until:
                    break
        finally:
            for task in tasks:
                if not task.done():
                    task.cancel()
            results = await asyncio.gather(*tasks, return_exceptions=True)
            # A connect can complete just as the caller cancels or a hook
            # yields. Close every successful result we did not adopt.
            for task, outcome in zip(tasks, results, strict=True):
                if task not in adopted and isinstance(outcome, tuple):
                    with contextlib.suppress(Exception):
                        await outcome[1].close()

        if self.clients:
            return len(self.clients)

        remaining = deadline - loop.time()
        if remaining <= 0:
            break

        wait = min(delay, remaining)
        logger.info(f"Retrying directory connections in {wait:.0f}s...")
        await asyncio.sleep(wait)
        delay = min(delay * backoff, max_delay)

    if not self.clients:
        logger.error(
            f"Failed to connect to any directory server after {timeout}s. "
            "Caller should rely on the periodic reconnect task."
        )
    return len(self.clients)

connect_to_directory(dir_server: str) -> tuple[str, DirectoryClient] | None async

Connect to a single directory server.

Args: dir_server: Address as "host" or "host:port".

Returns: (node_id, client) on success, or None if the address could not be parsed or the connect/handshake failed. The returned client is not yet stored in self.clients; the caller (or :meth:connect_all_with_retry / :meth:reconnect_disconnected) handles registration.

Source code in jmcore/src/jmcore/directory_pool.py

async def connect_to_directory(self, dir_server: str) -> tuple[str, DirectoryClient] | None:
    """
    Connect to a single directory server.

    Args:
        dir_server: Address as ``"host"`` or ``"host:port"``.

    Returns:
        ``(node_id, client)`` on success, or ``None`` if the address
        could not be parsed or the connect/handshake failed. The
        returned client is not yet stored in ``self.clients``; the
        caller (or :meth:`connect_all_with_retry` /
        :meth:`reconnect_disconnected`) handles registration.
    """
    try:
        host, port = parse_directory_address(dir_server)
    except Exception as e:
        logger.bind(sensitive=True).debug(f"Cannot parse directory address {dir_server!r}: {e}")
        return None

    node_id = f"{host}:{port}"

    client: DirectoryClient | None = None
    try:
        kwargs = self._build_client_kwargs(host, port)
        client = DirectoryClient(**kwargs)
        await client.connect()
        return (node_id, client)
    except asyncio.CancelledError:
        # Startup may finish while this client's handshake is still pending.
        # DirectoryClient.connect() only cleans up ordinary exceptions.
        if client is not None:
            with contextlib.suppress(Exception):
                await client.close()
        raise
    except Exception as e:
        logger.bind(sensitive=True).debug(f"Failed to connect to {dir_server}: {e}")
        return None

list_disconnected() -> list[tuple[str, str]]

Return (dir_server, node_id) pairs for configured servers not currently in self.clients.

Servers whose addresses fail to parse are skipped (with a debug log) rather than reported as disconnected, since the user almost certainly mistyped them and re-trying won't help.

Source code in jmcore/src/jmcore/directory_pool.py

def list_disconnected(self) -> list[tuple[str, str]]:
    """
    Return ``(dir_server, node_id)`` pairs for configured servers
    not currently in ``self.clients``.

    Servers whose addresses fail to parse are skipped (with a debug
    log) rather than reported as disconnected, since the user almost
    certainly mistyped them and re-trying won't help.
    """
    connected = set(self.clients.keys())
    out: list[tuple[str, str]] = []
    for server in self.directory_servers:
        try:
            host, port = parse_directory_address(server)
        except Exception as e:
            logger.bind(sensitive=True).debug(f"Skipping unparseable directory {server!r}: {e}")
            continue
        node_id = f"{host}:{port}"
        if node_id not in connected:
            out.append((server, node_id))
    return out

reconnect_disconnected() -> list[tuple[str, DirectoryClient]] async

Attempt to reconnect to every configured-but-disconnected server.

Newly connected clients are added to self.clients and _on_directory_connected fires for each. The return value is the list of (node_id, client) for callers that need to do additional per-reconnect work (e.g. start a listener task).

Source code in jmcore/src/jmcore/directory_pool.py

async def reconnect_disconnected(self) -> list[tuple[str, DirectoryClient]]:
    """
    Attempt to reconnect to every configured-but-disconnected server.

    Newly connected clients are added to ``self.clients`` and
    ``_on_directory_connected`` fires for each. The return value is
    the list of ``(node_id, client)`` for callers that need to do
    additional per-reconnect work (e.g. start a listener task).
    """
    newly_connected: list[tuple[str, DirectoryClient]] = []
    for dir_server, _expected_node_id in self.list_disconnected():
        result = await self.connect_to_directory(dir_server)
        if result is None:
            continue
        node_id, client = result
        self.clients[node_id] = client
        logger.bind(sensitive=True).info(f"Reconnected to directory: {dir_server}")
        await self._on_directory_connected(node_id, client)
        newly_connected.append((node_id, client))
    return newly_connected

DirectoryConnectionResult dataclass

Aggregate outcome of one parallel directory connection pass.

Source code in jmcore/src/jmcore/directory_pool.py

@dataclass(frozen=True)
class DirectoryConnectionResult:
    """Aggregate outcome of one parallel directory connection pass."""

    connected: int
    total: int

    @property
    def failed(self) -> int:
        """Return the number of configured directories that did not connect."""
        return self.total - self.connected

Attributes

connected: int instance-attribute

failed: int property

Return the number of configured directories that did not connect.

total: int instance-attribute

Functions: