distribute

-module(distribute@registry).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/registry.gleam").
-export([typed_name/3, named/2, typed_name_to_string/1, typed_name_encoder/1, typed_name_decoder/1, pool_member/2, unregister_error_to_string/1, register_error_to_string/1, register/2, register_typed/2, register_global/2, register_global_with_resolver/3, unregister/1, unregister_typed/1, whereis/1, lookup/1, is_registered/1, lookup_error_to_string/1, lookup_with_timeout/3, lookup_async/4]).
-export_type([typed_name/1, register_error/0, unregister_error/0, lookup_error/0]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

-opaque typed_name(GRC) :: {typed_name,
        binary(),
        fun((GRC) -> {ok, bitstring()} |
            {error, distribute@codec:encode_error()}),
        fun((bitstring()) -> {ok, GRC} |
            {error, distribute@codec:decode_error()})}.

-type register_error() :: already_exists |
    invalid_process |
    {invalid_argument, binary()} |
    {network_error, binary()} |
    {registration_failed, binary()}.

-type unregister_error() :: not_found | not_owned.

-type lookup_error() :: lookup_not_found |
    {lookup_invalid_timeout, integer()} |
    {lookup_invalid_poll_interval, integer()}.

-file("src/distribute/registry.gleam", 36).
?DOC(" Create a typed name from explicit encoder/decoder.\n").
-spec typed_name(
    binary(),
    fun((GRD) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, GRD} | {error, distribute@codec:decode_error()})
) -> typed_name(GRD).
typed_name(Name, Encoder, Decoder) ->
    {typed_name, Name, Encoder, Decoder}.

-file("src/distribute/registry.gleam", 77).
?DOC(
    " Create a typed name from a bundled `Codec`.\n"
    "\n"
    " ```gleam\n"
    " let counter = registry.named(\"counter\", codec.int())\n"
    " ```\n"
    "\n"
    " ## Caution: split-brain conflict resolution is brutal\n"
    "\n"
    " Names registered via `register_global` are stored in Erlang's\n"
    " `:global` table. After a network partition heals, `:global` may\n"
    " discover the same name registered to different PIDs on each\n"
    " side. Its default conflict resolver (`random_notify_name/3`)\n"
    " picks a winner and sends an **uncatchable**\n"
    " `exit(loser_pid, kill)` to the other PID. The library cannot\n"
    " trap this and there is no user hook to plug in: it is a\n"
    " property of `:global`, not of `distribute`.\n"
    "\n"
    " Operational consequences:\n"
    "\n"
    " - critical singletons can vanish without warning during a\n"
    "   partition-heal event;\n"
    " - any work in flight on the loser PID is lost;\n"
    " - Subjects obtained from a pre-split `lookup` point at the\n"
    "   loser and silently drop sends after the kill.\n"
    "\n"
    " Mitigation: subscribe to `cluster_monitor` and treat each\n"
    " `NodeUp` after a known partition as a \"re-validate critical\n"
    " state\" signal; re-`lookup` registered actors before each\n"
    " critical operation rather than caching Subjects across long\n"
    " timescales. For workloads where this is unacceptable, use `pg`\n"
    " (process groups, eventually-consistent, no kill-on-merge) or a\n"
    " consensus layer instead. See `docs/safety_and_limits.md` under\n"
    " \"BEAM and OS-level risks\" for the full discussion.\n"
).
-spec named(binary(), distribute@codec:codec(GRH)) -> typed_name(GRH).
named(Name, C) ->
    {typed_name, Name, erlang:element(2, C), erlang:element(3, C)}.

-file("src/distribute/registry.gleam", 82).
?DOC(" Get the name string.\n").
-spec typed_name_to_string(typed_name(any())) -> binary().
typed_name_to_string(Tn) ->
    erlang:element(2, Tn).

-file("src/distribute/registry.gleam", 87).
?DOC(" Get the encoder.\n").
-spec typed_name_encoder(typed_name(GRM)) -> fun((GRM) -> {ok, bitstring()} |
    {error, distribute@codec:encode_error()}).
typed_name_encoder(Tn) ->
    erlang:element(3, Tn).

-file("src/distribute/registry.gleam", 92).
?DOC(" Get the decoder.\n").
-spec typed_name_decoder(typed_name(GRP)) -> fun((bitstring()) -> {ok, GRP} |
    {error, distribute@codec:decode_error()}).
typed_name_decoder(Tn) ->
    erlang:element(4, Tn).

-file("src/distribute/registry.gleam", 100).
?DOC(
    " Derive a `TypedName` for a pool worker.\n"
    "\n"
    " `pool_member(base, 2)` returns a `TypedName` with name `\"<base>_2\"` and\n"
    " the same codecs.\n"
).
-spec pool_member(typed_name(GRS), integer()) -> typed_name(GRS).
pool_member(Base, Index) ->
    {typed_name,
        <<<<(erlang:element(2, Base))/binary, "_"/utf8>>/binary,
            (erlang:integer_to_binary(Index))/binary>>,
        erlang:element(3, Base),
        erlang:element(4, Base)}.

-file("src/distribute/registry.gleam", 136).
-spec unregister_error_to_string(unregister_error()) -> binary().
unregister_error_to_string(Err) ->
    case Err of
        not_found ->
            <<"Name not registered"/utf8>>;

        not_owned ->
            <<"Name owned by a remote node"/utf8>>
    end.

-file("src/distribute/registry.gleam", 143).
-spec register_error_to_string(register_error()) -> binary().
register_error_to_string(Err) ->
    case Err of
        already_exists ->
            <<"Name is already registered"/utf8>>;

        invalid_process ->
            <<"Invalid or dead process"/utf8>>;

        {invalid_argument, Reason} ->
            <<"Invalid argument: "/utf8, Reason/binary>>;

        {network_error, Reason@1} ->
            <<"Network error: "/utf8, Reason@1/binary>>;

        {registration_failed, Reason@2} ->
            <<"Registration failed: "/utf8, Reason@2/binary>>
    end.

-file("src/distribute/registry.gleam", 226).
-spec emit_register_outcome(binary(), {ok, nil} | {error, register_error()}) -> nil.
emit_register_outcome(Name, Outcome) ->
    case Outcome of
        {ok, nil} ->
            distribute@telemetry:emit({actor_registered, Name});

        {error, E} ->
            distribute@telemetry:emit(
                {actor_registration_failed, Name, register_error_to_string(E)}
            )
    end.

-file("src/distribute/registry.gleam", 725).
-spec validate_name(binary()) -> {ok, nil} | {error, register_error()}.
validate_name(Name) ->
    case distribute_ffi_utils:is_valid_registry_name(Name) of
        true ->
            {ok, nil};

        false ->
            {error,
                {invalid_argument,
                    <<"Name must be 1..255 bytes from charset [a-zA-Z0-9._-]"/utf8>>}}
    end.

-file("src/distribute/registry.gleam", 217).
?DOC(
    " Register a PID under a global name.\n"
    "\n"
    " Low-level escape hatch. Prefer `register_global/2` when you have a\n"
    " `GlobalSubject`, or `register_typed/2` when you already hold the exact\n"
    " distributed `Subject(BitArray)` that remote lookups will reconstruct.\n"
    "\n"
    " A raw PID carries no protocol/tag invariant by itself: if you register the\n"
    " owner of a subject whose mailbox tag does not equal `name`, remote lookups\n"
    " will still reconstruct `unsafe_from_name(name, pid)` and can silently send into a\n"
    " mailbox slot the actor never receives on.\n"
    "\n"
    " ## Caution: split-brain conflict resolution is brutal\n"
    "\n"
    " Backed by `:global`. After a network partition heals, conflict\n"
    " resolution sends an uncatchable `exit(loser_pid, kill)` to one\n"
    " side of any duplicate registration. See `named/2` and\n"
    " `docs/safety_and_limits.md` for the full operational picture.\n"
).
-spec register(binary(), gleam@erlang@process:pid_()) -> {ok, nil} |
    {error, register_error()}.
register(Name, Pid) ->
    Outcome = case validate_name(Name) of
        {error, E} ->
            {error, E};

        {ok, _} ->
            registry_ffi:register(Name, Pid)
    end,
    emit_register_outcome(Name, Outcome),
    Outcome.

-file("src/distribute/registry.gleam", 252).
?DOC(
    " Register a name-tagged distributed `Subject(BitArray)` under a global name.\n"
    "\n"
    " Low-level escape hatch: this is the raw-subject sibling of\n"
    " `register_global/2`. The supplied subject must already be the exact\n"
    " name-tagged distributed subject that remote lookups will reconstruct\n"
    " (`global.unsafe_from_name(name, pid, ...)` or the subject returned by\n"
    " `actor.start_*`).\n"
    "\n"
    " Subjects built with `process.new_subject()` (random Ref tag) or\n"
    " `global.from_pid()` (Nil tag) are rejected: a remote `lookup` would\n"
    " reconstruct `unsafe_from_name(name, pid)` and silently send into a mailbox slot\n"
    " that never matches the original selector.\n"
).
-spec register_typed(binary(), gleam@erlang@process:subject(bitstring())) -> {ok,
        nil} |
    {error, register_error()}.
register_typed(Name, Subject) ->
    Outcome = case validate_name(Name) of
        {error, E} ->
            {error, E};

        {ok, _} ->
            case distribute_ffi_utils:subject_tag_matches_name(Subject, Name) of
                false ->
                    {error,
                        {invalid_argument,
                            <<<<"Subject tag does not match registry name \""/utf8,
                                    Name/binary>>/binary,
                                "\". Build it with global.unsafe_from_name/4 or actor.start_*."/utf8>>}};

                true ->
                    case gleam@erlang@process:subject_owner(Subject) of
                        {ok, Pid} ->
                            registry_ffi:register(Name, Pid);

                        {error, nil} ->
                            {error, invalid_process}
                    end
            end
    end,
    emit_register_outcome(Name, Outcome),
    Outcome.

-file("src/distribute/registry.gleam", 313).
?DOC(
    " Register a `GlobalSubject` under a typed name.\n"
    "\n"
    " The `msg` type parameter on `TypedName(msg)` and `GlobalSubject(msg)`\n"
    " must match. The compiler enforces this.\n"
    "\n"
    " **Runtime invariant**: the supplied `gs` must have been built with\n"
    " `global.unsafe_from_name(name, ...)` (or via `actor.start_*`, which does the\n"
    " same internally) so that its tag equals the registry name. A remote\n"
    " `lookup` reconstructs the Subject as `unsafe_from_name(name, pid)` and any\n"
    " other tag would silently accumulate messages in the actor's mailbox\n"
    " without ever matching its selector.\n"
    "\n"
    " We enforce the invariant up front. Registering a Subject built with\n"
    " `global.new()` (random Ref tag) or `global.from_pid()` (Nil tag)\n"
    " returns `Error(InvalidArgument(...))` instead of silently breaking.\n"
    "\n"
    " ## Propagation is not atomic across the cluster\n"
    "\n"
    " `Ok(Nil)` means `:global` accepted the registration **on this\n"
    " node**. Cross-cluster visibility follows the standard `:global`\n"
    " gossip path and is not instantaneous: a `lookup` from a remote\n"
    " node a microsecond after this call returns can still see the\n"
    " previous state (or no state) until propagation completes. This\n"
    " is a property of `:global` itself, not a defect in\n"
    " `distribute`. Callers who need to register-then-immediately-\n"
    " lookup on another node have three options:\n"
    "\n"
    " - Use `lookup_with_timeout` with a small budget (200-500 ms is\n"
    "   usually plenty); the polling loop bridges the propagation\n"
    "   window without busy-waiting.\n"
    " - Send a synchronous `call` from the registering node back to\n"
    "   the actor before notifying remote callers; the round trip\n"
    "   forces propagation to complete.\n"
    " - For test setups, call `:global.sync()` directly via FFI.\n"
    "   Not recommended for production hot paths because it blocks\n"
    "   on the global_name_server queue.\n"
).
-spec register_global(typed_name(GSL), distribute@global:global_subject(GSL)) -> {ok,
        nil} |
    {error, register_error()}.
register_global(Tn, Gs) ->
    case distribute_ffi_utils:subject_tag_matches_name(
        distribute@global:subject(Gs),
        erlang:element(2, Tn)
    ) of
        false ->
            Err = {invalid_argument,
                <<<<"GlobalSubject tag does not match TypedName \""/utf8,
                        (erlang:element(2, Tn))/binary>>/binary,
                    "\". Build it with global.unsafe_from_name/4 or actor.start_*."/utf8>>},
            emit_register_outcome(erlang:element(2, Tn), {error, Err}),
            {error, Err};

        true ->
            case distribute@global:owner(Gs) of
                {ok, Pid} ->
                    register(erlang:element(2, Tn), Pid);

                {error, nil} ->
                    emit_register_outcome(
                        erlang:element(2, Tn),
                        {error, invalid_process}
                    ),
                    {error, invalid_process}
            end
    end.

-file("src/distribute/registry.gleam", 399).
?DOC(
    " Like `register_global/2`, but installs a custom split-brain\n"
    " conflict resolver alongside the registration.\n"
    "\n"
    " `:global` invokes the resolver when the same name is claimed by\n"
    " two different PIDs (typically after a network partition heals).\n"
    " The default resolver behind `register_global/2`\n"
    " (`:global.random_notify_name/3`) picks a winner at random and\n"
    " kills the loser. For workloads where one side should always win,\n"
    " a leader pinned to a specific node, a router whose state\n"
    " must survive on a primary, a singleton whose oldest instance is\n"
    " authoritative. Pass an explicit resolver from\n"
    " `distribute/conflict`.\n"
    "\n"
    " ```gleam\n"
    " import distribute/conflict\n"
    " import distribute/registry\n"
    "\n"
    " let assert Ok(Nil) =\n"
    "   registry.register_global_with_resolver(\n"
    "     counter_name,\n"
    "     gs,\n"
    "     conflict.node_priority([\"primary@host\", \"secondary@host\"]),\n"
    "   )\n"
    " ```\n"
    "\n"
    " ## Operational guard rails\n"
    "\n"
    " The resolver runs *inside* `:global`'s singleton worker\n"
    " process; while it executes, every `:global` operation cluster-\n"
    " wide is serialised behind it. To bound the worst-case stall,\n"
    " the FFI shim spawns the resolver in a short-lived worker with\n"
    " a hard deadline read from\n"
    " `config.get().conflict_resolver_timeout_ms` (default\n"
    " **1 000 ms**). On timeout or crash, the fallback\n"
    " (lowest term-ordered PID wins) fires and\n"
    " `telemetry.ConflictResolverFailed` is emitted so operators can\n"
    " see that the user fn misbehaved. Every resolution emits\n"
    " `telemetry.ConflictResolved` with the surviving PID (or `None`\n"
    " for `KillBoth`).\n"
    "\n"
    " ## Data-loss honesty box\n"
    "\n"
    " The fallback is **deterministic** but not **state-aware**: if\n"
    " your resolver was supposed to keep the side with the most\n"
    " recent state and it fails (timeout, panic, malformed return),\n"
    " the fallback picks lowest-PID and the cluster may lose\n"
    " whatever the loser was holding. For stateful actors where this\n"
    " would corrupt the system, use `conflict.kill_both()` as the\n"
    " resolver and watch `telemetry.ConflictResolved(_, None)` to\n"
    " trigger an application-level recovery path (re-elect, re-\n"
    " bootstrap from durable storage). See `distribute/conflict` for\n"
    " the full discussion.\n"
    "\n"
    " ## Tag invariant\n"
    "\n"
    " Same as `register_global/2`: the supplied `GlobalSubject` must\n"
    " carry the `TypedName` as its tag. A mismatch returns\n"
    " `Error(InvalidArgument(...))` before any `:global` call. Build\n"
    " the subject via `actor.start_*` or via `lookup`.\n"
).
-spec register_global_with_resolver(
    typed_name(GSQ),
    distribute@global:global_subject(GSQ),
    fun((binary(), gleam@erlang@process:pid_(), gleam@erlang@process:pid_()) -> distribute@conflict:conflict_outcome())
) -> {ok, nil} | {error, register_error()}.
register_global_with_resolver(Tn, Gs, Resolver) ->
    case distribute_ffi_utils:subject_tag_matches_name(
        distribute@global:subject(Gs),
        erlang:element(2, Tn)
    ) of
        false ->
            Err = {invalid_argument,
                <<<<"GlobalSubject tag does not match TypedName \""/utf8,
                        (erlang:element(2, Tn))/binary>>/binary,
                    "\". Build it with global.unsafe_from_name/4 or actor.start_*."/utf8>>},
            emit_register_outcome(erlang:element(2, Tn), {error, Err}),
            {error, Err};

        true ->
            case distribute@global:owner(Gs) of
                {ok, Pid} ->
                    Outcome = conflict_ffi:register_with_resolver(
                        erlang:element(2, Tn),
                        Pid,
                        Resolver
                    ),
                    emit_register_outcome(erlang:element(2, Tn), Outcome),
                    Outcome;

                {error, nil} ->
                    emit_register_outcome(
                        erlang:element(2, Tn),
                        {error, invalid_process}
                    ),
                    {error, invalid_process}
            end
    end.

-file("src/distribute/registry.gleam", 462).
-spec emit_unregister_outcome(binary(), {ok, nil} | {error, unregister_error()}) -> nil.
emit_unregister_outcome(Name, Outcome) ->
    Removed = case Outcome of
        {ok, nil} ->
            true;

        {error, _} ->
            false
    end,
    distribute@telemetry:emit({actor_unregistered, Name, Removed}).

-file("src/distribute/registry.gleam", 449).
?DOC(
    " Unregister a global name.\n"
    "\n"
    " See also: `register/2`, `register_global/2`, `lookup/1`.\n"
    "\n"
    " Returns `Ok(Nil)` when *this* VM owned the name and the entry has\n"
    " been removed from `:global`. Returns `Error(NotOwned)` when the\n"
    " owning PID runs on another node (the local-ownership ACL refuses\n"
    " the unregister), and `Error(NotFound)` when the name was not\n"
    " registered at all. Idempotent cleanup paths can `let _ =\n"
    " unregister(name)` to discard the outcome; observability paths can\n"
    " pattern-match for diagnostics.\n"
    "\n"
    " **Local-ownership ACL**: stateless `node(Pid) =:= node()` check via\n"
    " `:global.whereis_name/1`. No local mirror can go stale, no\n"
    " auto-cleanup is needed when processes die. This blocks the \"registry\n"
    " wipe\" vector where unvalidated input flows into `unregister` and\n"
    " tears down arbitrary cluster routing. Standard auto-cleanup\n"
    " (`:global` removes a name when the owning process dies) is\n"
    " unaffected.\n"
).
-spec unregister(binary()) -> {ok, nil} | {error, unregister_error()}.
unregister(Name) ->
    Outcome = registry_ffi:unregister(Name),
    emit_unregister_outcome(Name, Outcome),
    Outcome.

-file("src/distribute/registry.gleam", 458).
?DOC(
    " Unregister using a `TypedName` directly. Type-safe sibling of\n"
    " `unregister/1` that reuses the protocol the caller already holds,\n"
    " avoiding a stringly-typed extraction at the call site.\n"
).
-spec unregister_typed(typed_name(any())) -> {ok, nil} |
    {error, unregister_error()}.
unregister_typed(Tn) ->
    unregister(erlang:element(2, Tn)).

-file("src/distribute/registry.gleam", 474).
?DOC(" Look up a globally registered PID by name.\n").
-spec whereis(binary()) -> {ok, gleam@erlang@process:pid_()} | {error, nil}.
whereis(Name) ->
    registry_ffi:whereis(Name).

-file("src/distribute/registry.gleam", 506).
?DOC(
    " Look up a `GlobalSubject` by `TypedName`.\n"
    "\n"
    " Reconstructs the subject using the name-based tag, so messages\n"
    " sent via the returned subject reach the actor mailbox.\n"
    "\n"
    " ## Snapshot semantics\n"
    "\n"
    " `lookup` returns the Subject that resolves to the PID found at\n"
    " the exact moment the lookup ran. The PID may crash, deregister,\n"
    " or leave the cluster a microsecond later. Subsequent\n"
    " `global.send` calls on a stale Subject silently drop messages\n"
    " because that is the BEAM's contract for `erlang:send/2` to a\n"
    " dead PID. For long-lived references, either re-`lookup` before\n"
    " each critical send, or use `global.call` (which monitors the\n"
    " target and surfaces `Error(TargetDown)` immediately when the PID\n"
    " has died).\n"
    "\n"
    " ## Propagation is not atomic across the cluster\n"
    "\n"
    " A `lookup` from node B issued microseconds after a successful\n"
    " `register_global` on node A may still return `Error(Nil)`\n"
    " because `:global` propagates registrations via gossip rather\n"
    " than synchronous broadcast. This is the standard `:global`\n"
    " behaviour, not a defect in `distribute`. For the\n"
    " register-then-immediately-lookup pattern across nodes, prefer\n"
    " `lookup_with_timeout` with a 200-500 ms budget over a\n"
    " single-shot `lookup`: the polling loop bridges the propagation\n"
    " window without busy-waiting.\n"
).
-spec lookup(typed_name(GTF)) -> {ok, distribute@global:global_subject(GTF)} |
    {error, nil}.
lookup(Tn) ->
    case registry_ffi:whereis(erlang:element(2, Tn)) of
        {ok, Pid} ->
            {ok,
                distribute@global:unsafe_from_name(
                    erlang:element(2, Tn),
                    Pid,
                    erlang:element(3, Tn),
                    erlang:element(4, Tn)
                )};

        {error, nil} ->
            {error, nil}
    end.

-file("src/distribute/registry.gleam", 514).
?DOC(" Check whether a name is currently registered.\n").
-spec is_registered(binary()) -> boolean().
is_registered(Name) ->
    case registry_ffi:whereis(Name) of
        {ok, _} ->
            true;

        {error, _} ->
            false
    end.

-file("src/distribute/registry.gleam", 531).
-spec lookup_error_to_string(lookup_error()) -> binary().
lookup_error_to_string(Err) ->
    case Err of
        lookup_not_found ->
            <<"Name not found within timeout"/utf8>>;

        {lookup_invalid_timeout, V} ->
            <<<<"Invalid timeout: "/utf8, (erlang:integer_to_binary(V))/binary>>/binary,
                " (must be > 0)"/utf8>>;

        {lookup_invalid_poll_interval, V@1} ->
            <<<<"Invalid poll interval: "/utf8,
                    (erlang:integer_to_binary(V@1))/binary>>/binary,
                " (must be > 0)"/utf8>>
    end.

-file("src/distribute/registry.gleam", 690).
-spec do_lookup_with_timeout(typed_name(GUI), integer(), integer(), integer()) -> {ok,
        distribute@global:global_subject(GUI)} |
    {error, lookup_error()}.
do_lookup_with_timeout(Tn, Start, Timeout_ms, Poll_interval_ms) ->
    case lookup(Tn) of
        {ok, Gs} ->
            {ok, Gs};

        {error, _} ->
            Elapsed = distribute_ffi_utils:monotonic_ms() - Start,
            case Elapsed >= Timeout_ms of
                true ->
                    {error, lookup_not_found};

                false ->
                    Remaining = Timeout_ms - Elapsed,
                    Wait_ms = case Poll_interval_ms < Remaining of
                        true ->
                            Poll_interval_ms;

                        false ->
                            Remaining
                    end,
                    gleam_erlang_ffi:sleep(Wait_ms),
                    do_lookup_with_timeout(
                        Tn,
                        Start,
                        Timeout_ms,
                        Poll_interval_ms
                    )
            end
    end.

-file("src/distribute/registry.gleam", 551).
?DOC(
    " Poll until a `TypedName` is registered or `timeout_ms` elapses.\n"
    "\n"
    " Retries every `poll_interval_ms` milliseconds. Both `timeout_ms` and\n"
    " `poll_interval_ms` must be strictly positive. Passing zero or a\n"
    " negative value returns a typed `Error` instead of crashing the\n"
    " scheduler (a non-positive `poll_interval_ms` would tight-loop or\n"
    " raise `badarg` inside `process.sleep`).\n"
    "\n"
    " **Warning**: blocks the calling process. Do not call inside an OTP actor\n"
    " handler. Use `lookup_async` instead.\n"
).
-spec lookup_with_timeout(typed_name(GTK), integer(), integer()) -> {ok,
        distribute@global:global_subject(GTK)} |
    {error, lookup_error()}.
lookup_with_timeout(Tn, Timeout_ms, Poll_interval_ms) ->
    case Timeout_ms =< 0 of
        true ->
            {error, {lookup_invalid_timeout, Timeout_ms}};

        false ->
            case Poll_interval_ms =< 0 of
                true ->
                    {error, {lookup_invalid_poll_interval, Poll_interval_ms}};

                false ->
                    do_lookup_with_timeout(
                        Tn,
                        distribute_ffi_utils:monotonic_ms(),
                        Timeout_ms,
                        Poll_interval_ms
                    )
            end
    end.

-file("src/distribute/registry.gleam", 641).
-spec do_async_loop(
    typed_name(GUB),
    gleam@erlang@process:subject({ok, distribute@global:global_subject(GUB)} |
        {error, lookup_error()}),
    gleam@erlang@process:selector(nil),
    integer(),
    integer(),
    integer()
) -> nil.
do_async_loop(Tn, Reply_to, Caller_dead, Start, Timeout_ms, Poll_interval_ms) ->
    case lookup(Tn) of
        {ok, Gs} ->
            gleam@erlang@process:send(Reply_to, {ok, Gs}),
            nil;

        {error, _} ->
            Elapsed = distribute_ffi_utils:monotonic_ms() - Start,
            case Elapsed >= Timeout_ms of
                true ->
                    gleam@erlang@process:send(
                        Reply_to,
                        {error, lookup_not_found}
                    ),
                    nil;

                false ->
                    Remaining = Timeout_ms - Elapsed,
                    Wait_ms = case Poll_interval_ms < Remaining of
                        true ->
                            Poll_interval_ms;

                        false ->
                            Remaining
                    end,
                    case gleam_erlang_ffi:select(Caller_dead, Wait_ms) of
                        {ok, nil} ->
                            nil;

                        {error, nil} ->
                            do_async_loop(
                                Tn,
                                Reply_to,
                                Caller_dead,
                                Start,
                                Timeout_ms,
                                Poll_interval_ms
                            )
                    end
            end
    end.

-file("src/distribute/registry.gleam", 626).
?DOC(
    " Worker body for `lookup_async`. Monitor-driven liveness on the caller\n"
    " catches both crashes and normal exits, unlike a link.\n"
).
-spec do_async_lookup(
    typed_name(GTV),
    gleam@erlang@process:subject({ok, distribute@global:global_subject(GTV)} |
        {error, lookup_error()}),
    gleam@erlang@process:pid_(),
    integer(),
    integer(),
    integer()
) -> nil.
do_async_lookup(Tn, Reply_to, Caller, Start, Timeout_ms, Poll_interval_ms) ->
    Mon = gleam@erlang@process:monitor(Caller),
    Caller_dead = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        gleam@erlang@process:select_specific_monitor(
            _pipe,
            Mon,
            fun(_) -> nil end
        )
    end,
    do_async_loop(
        Tn,
        Reply_to,
        Caller_dead,
        Start,
        Timeout_ms,
        Poll_interval_ms
    ).

-file("src/distribute/registry.gleam", 590).
?DOC(
    " Non-blocking poll: spawns a polling worker that probes the registry\n"
    " until the name resolves, the timeout elapses, or the caller dies.\n"
    "\n"
    " The worker **monitors the caller**. If the caller process dies for\n"
    " any reason (crash *or* normal `exit(normal)`), the worker stops\n"
    " polling and terminates immediately. A bare `link` would only catch\n"
    " crashes: a short-lived web handler that completes with reason\n"
    " `normal` would leave the worker busy-polling for the full timeout\n"
    " while the result has nowhere to go.\n"
    "\n"
    " Validation errors (invalid timeout or poll interval) are forwarded to\n"
    " `reply_to` synchronously. The worker is not spawned for invalid params.\n"
    "\n"
    " ```gleam\n"
    " let reply = process.new_subject()\n"
    " registry.lookup_async(tn, reply, 5000, 100)\n"
    " let assert Ok(Ok(gs)) = process.receive(reply, 5100)\n"
    " ```\n"
).
-spec lookup_async(
    typed_name(GTP),
    gleam@erlang@process:subject({ok, distribute@global:global_subject(GTP)} |
        {error, lookup_error()}),
    integer(),
    integer()
) -> nil.
lookup_async(Tn, Reply_to, Timeout_ms, Poll_interval_ms) ->
    case Timeout_ms =< 0 of
        true ->
            gleam@erlang@process:send(
                Reply_to,
                {error, {lookup_invalid_timeout, Timeout_ms}}
            );

        false ->
            case Poll_interval_ms =< 0 of
                true ->
                    gleam@erlang@process:send(
                        Reply_to,
                        {error,
                            {lookup_invalid_poll_interval, Poll_interval_ms}}
                    );

                false ->
                    Caller = erlang:self(),
                    _ = proc_lib:spawn(
                        fun() ->
                            do_async_lookup(
                                Tn,
                                Reply_to,
                                Caller,
                                distribute_ffi_utils:monotonic_ms(),
                                Timeout_ms,
                                Poll_interval_ms
                            )
                        end
                    ),
                    nil
            end
    end.