tractor/tractor/discovery/_addr.py

# tractor: structured concurrent "actors".
# Copyright 2018-eternity Tyler Goodlet.

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
from __future__ import annotations
from uuid import uuid4
from typing import (
    Any,
    Protocol,
    ClassVar,
    Literal,
    Type,
    TypeAlias,
    TYPE_CHECKING,
)

from trio import (
    SocketListener,
)

from ..log import get_logger
from ..runtime._state import (
    _def_tpt_proto,
)
from ..ipc._tcp import TCPAddress
from ..ipc._uds import (
    UDSAddress,
    HAS_UDS,
)

if TYPE_CHECKING:
    # ONLY type-annots, the eager import costs ~4.5ms
    # of `import tractor` wall-time (gh #470).
    from ._tunnel import (
        TunnelledAddress,
    )
    from ..runtime._runtime import Actor
else:
    Actor = Any
    TunnelledAddress = Any

log = get_logger()

# TODO, maybe breakout the netns key to a struct?
# class NetNs(Struct)[str, int]:
#     ...

# TODO, can't we just use a type alias
# for this? namely just some `tuple[str, int, str, str]`?
#
# -[ ] would also just be simpler to keep this as SockAddr[tuple]
#     or something, implying it's just a simple pair of values which can
#     presumably be mapped to all transports?
# -[ ] `pydoc socket.socket.getsockname()` delivers a 4-tuple for
#     ipv6 `(hostaddr, port, flowinfo, scope_id)`.. so how should we
#     handle that?
# -[ ] as a further alternative to this wrap()/unwrap() approach we
#     could just implement `enc/dec_hook()`s for the `Address`-types
#     and just deal with our internal objs directly and always and
#     leave it to the codec layer to figure out marshalling?
#    |_ would mean only one spot to do the `.unwrap()` (which we may
#       end up needing to call from the hook()s anyway?)
# -[x] rename to `UnwrappedAddress[Descriptor]` ??
#    seems like the right name as per,
#    https://www.geeksforgeeks.org/introduction-to-address-descriptor/
#
TaggedTCPAddress: TypeAlias = tuple[
    Literal['tcp'],
    str,
    int,
]
TaggedUnixAddress: TypeAlias = tuple[
    Literal['unix'],
    str,
]
TaggedUDSAlias: TypeAlias = tuple[
    Literal['uds'],
    str,
]
TaggedAddress: TypeAlias = (
    TaggedTCPAddress
    |TaggedUnixAddress
)

# Input-only compatibility forms retained for older callers and
# serialized payloads.
LegacyTCPAddress: TypeAlias = tuple[str, int]
LegacyUDSAddress: TypeAlias = tuple[str, str]
LegacyUnwrappedAddress: TypeAlias = (
    LegacyTCPAddress
    |LegacyUDSAddress
)
UnwrappedAddress = TaggedAddress
# ?TODO? should we also include another 2 fields from our `Aid` msg
# such that we include the runtime `Actor.uid` of `.name` and `.uuid`?
# - would ensure uniqueness across entire net?
# - allows for easier runtime-level filtering of "actors by service
#   name"


# TODO, maybe rename to `SocketAddress`?
class Address(Protocol):
    proto_key: ClassVar[str]
    unwrapped_type: ClassVar[type]

    # TODO, i feel like an `.is_bound()` is a better thing to
    # support?
    # Lke, what use does this have besides a noop and if it's not
    # valid why aren't we erroring on creation/use?
    @property
    def is_valid(self) -> bool:
        ...

    # TODO, maybe `.netns` is a better name?
    @property
    def namespace(self) -> tuple[str, str|int]|None:
        '''
        The if-available, OS-specific "network namespace" key.

        '''
        ...

    @property
    def bindspace(self) -> str:
        '''
        Deliver the socket address' "bindable space" from
        a `socket.socket.bind()` and thus from the perspective of
        specific transport protocol domain.

        I.e. for most (layer-4) network-socket protocols this is
        normally the ipv4/6 address, for UDS this is normally
        a filesystem (sub-directory).

        For (distributed) network protocols this is normally the routing
        layer's domain/(ip-)address, though it might also include a "network namespace"
        key different then the default.

        For local-host-only transports this is either an explicit
        namespace (with types defined by the OS: netns, Cgroup, IPC,
        pid, etc. on linux) or failing that the sub-directory in the
        filesys in which socket/shm files are located *under*.

        '''
        ...

    @classmethod
    def from_addr(cls, addr: UnwrappedAddress) -> Address:
        ...

    def unwrap(self) -> UnwrappedAddress:
        '''
        Deliver the underying minimum field set in
        a primitive python data type-structure.
        '''
        ...

    @classmethod
    def get_random(
        cls,
        current_actor: Actor,
        bindspace: str|None = None,
    ) -> Address:
        ...

    # TODO, this should be something like a `.get_def_registar_addr()`
    # or similar since,
    # - it should be a **host singleton** (not root/tree singleton)
    # - we **only need this value** when one isn't provided to the
    #   runtime at boot and we want to implicitly provide a host-wide
    #   registrar.
    # - each rooted-actor-tree should likely have its own
    #   micro-registry (likely the root being it), also see
    @classmethod
    def get_root(cls) -> Address:
        ...

    def __repr__(self) -> str:
        ...

    def __eq__(self, other) -> bool:
        ...

    async def open_listener(
        self,
        **kwargs,
    ) -> SocketListener:
        ...

    async def close_listener(self):
        ...


# the address types available on this host: TCP always, UDS only
# where usable (`HAS_UDS`). Both registries derive from this single
# list via each type's `proto_key`.
_address_protos: list[Type[Address]] = [TCPAddress]
if HAS_UDS:
    _address_protos.append(UDSAddress)

_address_types: dict[str, Type[Address]] = {
    cls.proto_key: cls
    for cls in _address_protos
}


# TODO! really these are discovery sys default addrs ONLY useful for
# when none is provided to a root actor on first boot.
_default_lo_addrs: dict[str, UnwrappedAddress] = {
    cls.proto_key: cls.get_root().unwrap()
    for cls in _address_protos
}


def get_address_cls(name: str) -> Type[Address]:
    try:
        return _address_types[name]
    except KeyError:
        raise NotImplementedError(
            f'No IPC transport backend for {name!r} on this '
            f'platform!\n'
            f'(available: {list(_address_types)})\n'
        )


def is_wrapped_addr(addr: any) -> bool:
    # XXX NOTE, a `TunnelledAddress` is genuinely "wrapped" but is
    # deliberately NOT in `_address_types`: it has no
    # `MsgTransport` of its own (a tunnel is transparent to
    # `socket(2)`), so it gets no proto-key entry. See `._tunnel`.
    from ._tunnel import TunnelledAddress
    return (
        type(addr) in _address_types.values()
        or
        isinstance(addr, TunnelledAddress)
    )


def mk_uuid() -> str:
    '''
    Encapsulate creation of a uuid4 as `str` as used
    for creating `Actor.uid: tuple[str, str]` and/or
    `.msg.types.Aid`.

    '''
    return str(uuid4())


def wrap_address(
    addr: (
        TaggedAddress
        |TaggedUDSAlias
        |LegacyUnwrappedAddress
        |list[str|int]
        |str
        |Address
        |TunnelledAddress
    ),
) -> Address|TunnelledAddress:
    '''
    Wrap an `UnwrappedAddress` as an `Address`-type based
    on matching builtin python data-structures which we adhoc
    use for each.

    XXX NOTE, careful care must be placed to ensure
    `UnwrappedAddress` cases are **definitely unique** otherwise the
    wrong transport backend may be loaded and will break many
    low-level things in our runtime in a not-fun-to-debug way!

    XD

    '''
    if is_wrapped_addr(addr):
        return addr

    cls: Type|None = None
    # if 'sock' in addr[0]:
    #     import pdbp; pdbp.set_trace()
    match addr:

        case (
            ('tcp', str(), int())
            |
            ['tcp', str(), int()]
        ):
            return TCPAddress.from_addr(addr)

        case (
            (('unix' | 'uds'), str())
            |
            [('unix' | 'uds'), str()]
        ):
            return UDSAddress.from_addr(addr)

        # classic network socket-address as tuple/list
        case (
            (str(), int())
            |
            [str(), int()]
        ):
            cls = TCPAddress

        case (
            # (str()|Path(), str()|Path()),
            # ^TODO? uhh why doesn't this work!?

            (_, filename)
        ) if type(filename) is str:
            cls = UDSAddress

        # likely an unset UDS or TCP reg address as defaulted in
        # `_state._runtime_vars['_root_mailbox']`
        #
        # TODO? figure out when/if we even need this?
        case (
            None
            |
            [None, None]
        ):
            cls: Type[Address] = get_address_cls(_def_tpt_proto)
            addr: UnwrappedAddress = cls.get_root().unwrap()

        # multiaddr-format string, e.g.
        # '/ip4/127.0.0.1/tcp/1616'
        case str() if addr.startswith('/'):
            from tractor.discovery._multiaddr import (
                parse_maddr,
            )
            return parse_maddr(addr)

        case _:
            # import pdbp; pdbp.set_trace()
            # from tractor.devx import mk_pdb
            # mk_pdb().set_trace()
            raise TypeError(
                f'Can not wrap unwrapped-address ??\n'
                f'type(addr): {type(addr)!r}\n'
                f'addr: {addr!r}\n'
            )

    return cls.from_addr(addr)


def default_lo_addrs(
    transports: list[str],
) -> list[Type[Address]]:
    '''
    Return the default, host-singleton, registry address
    for an input transport key set.

    '''
    lo_addrs: list[UnwrappedAddress] = []
    for transport in transports:
        try:
            lo_addrs.append(_default_lo_addrs[transport])
        except KeyError:
            raise NotImplementedError(
                f'No default loopback addr for transport '
                f'{transport!r} on this platform!\n'
                f'(available: {list(_default_lo_addrs)})\n'
            )
    return lo_addrs