distribute

-module(distribute@global).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/global.gleam").
-export([new/2, from_pid/3, from_subject/3, unsafe_from_name/4, subject/1, owner/1, encoder/1, decoder/1, send_error_to_string/1, send/2, 'receive'/2, drain_reply/1, call/4, receive_default/1, call_default/3, call_isolated/4, call_isolated_default/3, reply/3, call_error_to_string/1]).
-export_type([global_subject/1, send_error/0, call_error/0, call_message/0, isolated_outcome/1]).

-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 global_subject(FWW) :: {global_subject,
        gleam@erlang@process:subject(bitstring()),
        fun((FWW) -> {ok, bitstring()} |
            {error, distribute@codec:encode_error()}),
        fun((bitstring()) -> {ok, FWW} |
            {error, distribute@codec:decode_error()})}.

-type send_error() :: {send_encode_failed, distribute@codec:encode_error()} |
    {payload_too_large, integer()}.

-type call_error() :: timeout |
    target_down |
    {call_encode_failed, distribute@codec:encode_error()} |
    {call_decode_failed, distribute@codec:decode_error()} |
    {call_payload_too_large, integer()}.

-type call_message() :: {got_reply, bitstring()} | target_died.

-type isolated_outcome(FWX) :: {isolated_ready,
        {ok, FWX} | {error, call_error()}} |
    isolated_proxy_down.

-file("src/distribute/global.gleam", 46).
?DOC(" New subject owned by the current process, with a unique tag.\n").
-spec new(
    fun((FXA) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, FXA} | {error, distribute@codec:decode_error()})
) -> global_subject(FXA).
new(Encoder, Decoder) ->
    Subject = gleam@erlang@process:new_subject(),
    {global_subject, Subject, Encoder, Decoder}.

-file("src/distribute/global.gleam", 83).
?DOC(
    " **Send-only.** A `GlobalSubject` built via `from_pid` must only be\n"
    " used with `send`. Calling `receive`, `call`, or `call_isolated` on\n"
    " it is a logic error: the shared `Nil` tag collides with the actor's\n"
    " own mailbox selectors and produces silent message interleaving.\n"
    " For any bidirectional path, register the actor and use\n"
    " `registry.lookup` (which goes through the internal\n"
    " `unsafe_from_name` constructor with a verified codec pairing).\n"
    "\n"
    " ## When to use\n"
    "\n"
    " Diagnostic and probe paths only. For example, calling on a PID you\n"
    " just confirmed dead to test `TargetDown` semantics. Production\n"
    " messaging should always go through `registry.lookup`.\n"
    "\n"
    " ## What goes wrong otherwise\n"
    "\n"
    " All subjects produced by `from_pid` on the same PID share the same\n"
    " `Nil` tag. The BEAM does not distinguish them: messages sent through\n"
    " any one of them all land in the same mailbox slot, *interleaved with\n"
    " the actor's default subject*. Two `from_pid` Subjects with different\n"
    " `msg` types pointing at the same actor will silently mix on the wire.\n"
    " The compiler catches the *type* mismatch only if you use the same\n"
    " `GlobalSubject` value at both ends. This is exactly what `from_pid`\n"
    " makes hard to guarantee.\n"
    "\n"
    " The `unsafe_from_name` path used by `registry.lookup` avoids this:\n"
    " each name produces a distinct deterministic tag, so messages routed\n"
    " through the registry cannot collide with each other or with the\n"
    " actor's own selectors.\n"
).
-spec from_pid(
    gleam@erlang@process:pid_(),
    fun((FXE) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, FXE} | {error, distribute@codec:decode_error()})
) -> global_subject(FXE).
from_pid(Pid, Encoder, Decoder) ->
    Subject = distribute_ffi_utils:create_subject(Pid, gleam@dynamic:nil()),
    {global_subject, Subject, Encoder, Decoder}.

-file("src/distribute/global.gleam", 93).
?DOC(" Wrap an existing `Subject(BitArray)`, keeping its tag.\n").
-spec from_subject(
    gleam@erlang@process:subject(bitstring()),
    fun((FXJ) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, FXJ} | {error, distribute@codec:decode_error()})
) -> global_subject(FXJ).
from_subject(Subject, Encoder, Decoder) ->
    {global_subject, Subject, Encoder, Decoder}.

-file("src/distribute/global.gleam", 128).
?DOC(false).
-spec unsafe_from_name(
    binary(),
    gleam@erlang@process:pid_(),
    fun((FXN) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, FXN} | {error, distribute@codec:decode_error()})
) -> global_subject(FXN).
unsafe_from_name(Name, Pid, Encoder, Decoder) ->
    Subject = distribute_ffi_utils:create_subject(
        Pid,
        gleam_stdlib:identity(Name)
    ),
    {global_subject, Subject, Encoder, Decoder}.

-file("src/distribute/global.gleam", 142).
-spec subject(global_subject(any())) -> gleam@erlang@process:subject(bitstring()).
subject(Global) ->
    erlang:element(2, Global).

-file("src/distribute/global.gleam", 146).
-spec owner(global_subject(any())) -> {ok, gleam@erlang@process:pid_()} |
    {error, nil}.
owner(Global) ->
    gleam@erlang@process:subject_owner(erlang:element(2, Global)).

-file("src/distribute/global.gleam", 150).
-spec encoder(global_subject(FXY)) -> fun((FXY) -> {ok, bitstring()} |
    {error, distribute@codec:encode_error()}).
encoder(Global) ->
    erlang:element(3, Global).

-file("src/distribute/global.gleam", 154).
-spec decoder(global_subject(FYB)) -> fun((bitstring()) -> {ok, FYB} |
    {error, distribute@codec:decode_error()}).
decoder(Global) ->
    erlang:element(4, Global).

-file("src/distribute/global.gleam", 170).
-spec send_error_to_string(send_error()) -> binary().
send_error_to_string(Error) ->
    case Error of
        {send_encode_failed, E} ->
            <<"Send failed: "/utf8,
                (distribute@codec:encode_error_to_string(E))/binary>>;

        {payload_too_large, Size} ->
            <<<<"Payload too large: "/utf8,
                    (erlang:integer_to_binary(Size))/binary>>/binary,
                " bytes"/utf8>>
    end.

-file("src/distribute/global.gleam", 205).
?DOC(
    " Encode and send a message.\n"
    "\n"
    " Returns `Error(PayloadTooLarge(size))` when the encoded payload exceeds\n"
    " `config.get().max_payload_size_bytes`. The message is never enqueued.\n"
    "\n"
    " ## Fire-and-forget semantics\n"
    "\n"
    " `Ok(Nil)` means the BEAM accepted the message for local\n"
    " dispatch. It does **not** mean the target received it. The\n"
    " receiver may have died after a successful `lookup`, the\n"
    " inter-node connection may drop before the payload is on the\n"
    " wire, the receiving node may panic before its mailbox is read.\n"
    " Successful `send` is the BEAM contract for `erlang:send/2`:\n"
    " \"queued for delivery\", not \"delivered\".\n"
    "\n"
    " If you need delivery confirmation, use `call` (synchronous,\n"
    " monitor-backed: `Error(TargetDown)` fires immediately when the\n"
    " PID is dead, `Error(Timeout)` fires when the reply does not\n"
    " arrive in time). For higher-level reliability (at-least-once,\n"
    " idempotency tokens, durable queues) build it on top of `call`\n"
    " or `send` plus your own ack protocol. The library does not\n"
    " bake retries in because the right strategy is application-\n"
    " specific.\n"
).
-spec send(global_subject(FYE), FYE) -> {ok, nil} | {error, send_error()}.
send(Global, Message) ->
    case distribute@codec:encode(erlang:element(3, Global), Message) of
        {error, E} ->
            {error, {send_encode_failed, E}};

        {ok, Binary} ->
            Cap = erlang:element(4, distribute@config:get()),
            Size = erlang:byte_size(Binary),
            case Size > Cap of
                true ->
                    distribute@telemetry:emit(
                        {payload_rejected, Size, Cap, payload_on_send}
                    ),
                    {error, {payload_too_large, Size}};

                false ->
                    gleam@erlang@process:send(erlang:element(2, Global), Binary),
                    {ok, nil}
            end
    end.

-file("src/distribute/global.gleam", 242).
?DOC(
    " Receive and decode a message. Only works on subjects you own.\n"
    "\n"
    " Returns `Error(codec.PayloadTooLarge(size))` when the received binary\n"
    " exceeds `config.get().max_payload_size_bytes`. Decode is skipped.\n"
    "\n"
    " Negative `timeout_ms` is clamped to `0` (poll-once-and-return)\n"
    " rather than propagated to Erlang's `receive after Timeout` clause,\n"
    " which would raise `timeout_value` and crash the caller. A bug at\n"
    " the call site that produces a negative number surfaces as an\n"
    " immediate `DecodeTimeout` instead of a process exit.\n"
).
-spec 'receive'(global_subject(FYI), integer()) -> {ok, FYI} |
    {error, distribute@codec:decode_error()}.
'receive'(Global, Timeout_ms) ->
    case gleam@erlang@process:'receive'(
        erlang:element(2, Global),
        distribute_ffi_utils:clamp_timeout(Timeout_ms)
    ) of
        {error, nil} ->
            {error, decode_timeout};

        {ok, Binary} ->
            Cap = erlang:element(4, distribute@config:get()),
            Size = erlang:byte_size(Binary),
            case Size > Cap of
                true ->
                    distribute@telemetry:emit(
                        {payload_rejected, Size, Cap, payload_on_receive}
                    ),
                    {error, {payload_too_large, Size}};

                false ->
                    case distribute@codec:decode(
                        erlang:element(4, Global),
                        Binary
                    ) of
                        {ok, Value} ->
                            {ok, Value};

                        {error, E} ->
                            distribute@telemetry:emit(
                                {decode_failed,
                                    distribute@codec:decode_error_to_string(E),
                                    decode_on_receive}
                            ),
                            {error, E}
                    end
            end
    end.

-file("src/distribute/global.gleam", 696).
-spec drain_reply_bounded(
    gleam@erlang@process:selector(nil),
    integer(),
    integer()
) -> nil.
drain_reply_bounded(Drain_selector, Remaining, Deadline) ->
    case (Remaining =< 0) orelse (distribute_ffi_utils:monotonic_ms() >= Deadline) of
        true ->
            nil;

        false ->
            case gleam_erlang_ffi:select(Drain_selector, 0) of
                {ok, nil} ->
                    drain_reply_bounded(Drain_selector, Remaining - 1, Deadline);

                {error, nil} ->
                    nil
            end
    end.

-file("src/distribute/global.gleam", 688).
?DOC(false).
-spec drain_reply(gleam@erlang@process:subject(bitstring())) -> nil.
drain_reply(Reply_subject) ->
    Deadline = distribute_ffi_utils:monotonic_ms() + 5,
    Drain_selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        gleam@erlang@process:select_map(_pipe, Reply_subject, fun(_) -> nil end)
    end,
    drain_reply_bounded(Drain_selector, 2048, Deadline).

-file("src/distribute/global.gleam", 356).
?DOC(
    " Synchronous request/response with monitor-based `TargetDown` detection.\n"
    "\n"
    " See also: `call_default/3` (uses configured timeout),\n"
    " `call_isolated/4` (mailbox-safe variant for long-running callers).\n"
    "\n"
    " Flow:\n"
    " 1. Resolve target PID and set a monitor. Dead target → `TargetDown`\n"
    " 2. Encode request. Failure → `CallEncodeFailed`\n"
    " 3. Size-check request. Oversized → `CallPayloadTooLarge`\n"
    " 4. Send request\n"
    " 5. Wait for reply OR monitor Down. Timeout → `Timeout`, Down → `TargetDown`\n"
    " 6. Size-check response. Oversized → `CallPayloadTooLarge`\n"
    " 7. Decode response. Failure → `CallDecodeFailed`\n"
    "\n"
    " ## Late-reply caveat (read carefully)\n"
    "\n"
    " `drain_reply` clears reply messages already in the caller's mailbox\n"
    " at the moment the timeout fires, but the BEAM has no API (without\n"
    " `erlang:alias/0`-aware Subjects, which would require bypassing\n"
    " `gleam_erlang`'s Subject layout) to drop messages that arrive\n"
    " *after* `call` returns. A late reply lands in the caller's mailbox\n"
    " tagged with the orphan reply Subject and **stays there forever**.\n"
    " no selector ever matches it again.\n"
    "\n"
    " **`gleam/otp/actor` is NOT immune.** Its receive loop is built on\n"
    " `process.selector_receive`, so unmatched messages are not dropped:\n"
    " they accumulate in the actor's mailbox. The BEAM then pays the\n"
    " \"selective receive penalty\". Every subsequent `selector_receive`\n"
    " scans past every orphan first. A long-running actor that issues\n"
    " thousands of timed-out `call`s will gradually grind to a halt and\n"
    " can OOM.\n"
    "\n"
    " The only safe mitigations:\n"
    "\n"
    " - **Short-lived callers** (CLI tools, scripts, request handlers\n"
    "   that exit after returning their response): orphan messages die\n"
    "   with the process. `call` directly is fine.\n"
    " - **High-volume callers, including long-running OTP actors that\n"
    "   issue many `call`s**: isolate every `call` in a short-lived\n"
    "   proxy process. The proxy makes the call, forwards the result to\n"
    "   the real caller via a one-shot Subject, then exits. Its\n"
    "   mailbox (and any orphan reply) is reaped. See the \"Isolated\n"
    "   call\" recipe in `docs/recipes.md`. This is the only design that\n"
    "   bounds the caller's mailbox under sustained timeouts.\n"
).
-spec call(
    global_subject(FYM),
    fun((gleam@erlang@process:subject(bitstring())) -> FYM),
    fun((bitstring()) -> {ok, FYP} | {error, distribute@codec:decode_error()}),
    integer()
) -> {ok, FYP} | {error, call_error()}.
call(Target, Make_request, Response_decoder, Timeout_ms) ->
    Clamped_timeout_ms = distribute_ffi_utils:clamp_timeout(Timeout_ms),
    case gleam@erlang@process:subject_owner(erlang:element(2, Target)) of
        {error, nil} ->
            distribute@telemetry:emit(call_target_down),
            {error, target_down};

        {ok, Target_pid} ->
            Mon = gleam@erlang@process:monitor(Target_pid),
            Reply_subject = gleam@erlang@process:new_subject(),
            Request = Make_request(Reply_subject),
            case distribute@codec:encode(erlang:element(3, Target), Request) of
                {error, E} ->
                    gleam@erlang@process:demonitor_process(Mon),
                    {error, {call_encode_failed, E}};

                {ok, Req_bits} ->
                    Cap = erlang:element(4, distribute@config:get()),
                    Req_size = erlang:byte_size(Req_bits),
                    case Req_size > Cap of
                        true ->
                            gleam@erlang@process:demonitor_process(Mon),
                            distribute@telemetry:emit(
                                {payload_rejected,
                                    Req_size,
                                    Cap,
                                    payload_on_call_request}
                            ),
                            {error, {call_payload_too_large, Req_size}};

                        false ->
                            gleam@erlang@process:send(
                                erlang:element(2, Target),
                                Req_bits
                            ),
                            Selector = begin
                                _pipe = gleam_erlang_ffi:new_selector(),
                                _pipe@1 = gleam@erlang@process:select_map(
                                    _pipe,
                                    Reply_subject,
                                    fun(Field@0) -> {got_reply, Field@0} end
                                ),
                                gleam@erlang@process:select_specific_monitor(
                                    _pipe@1,
                                    Mon,
                                    fun(_) -> target_died end
                                )
                            end,
                            case gleam_erlang_ffi:select(
                                Selector,
                                Clamped_timeout_ms
                            ) of
                                {error, nil} ->
                                    gleam@erlang@process:demonitor_process(Mon),
                                    drain_reply(Reply_subject),
                                    distribute@telemetry:emit(
                                        {call_timed_out, Clamped_timeout_ms}
                                    ),
                                    {error, timeout};

                                {ok, target_died} ->
                                    drain_reply(Reply_subject),
                                    distribute@telemetry:emit(call_target_down),
                                    {error, target_down};

                                {ok, {got_reply, Resp_bits}} ->
                                    gleam@erlang@process:demonitor_process(Mon),
                                    Resp_size = erlang:byte_size(Resp_bits),
                                    case Resp_size > Cap of
                                        true ->
                                            distribute@telemetry:emit(
                                                {payload_rejected,
                                                    Resp_size,
                                                    Cap,
                                                    payload_on_call_response}
                                            ),
                                            {error,
                                                {call_payload_too_large,
                                                    Resp_size}};

                                        false ->
                                            case distribute@codec:decode(
                                                Response_decoder,
                                                Resp_bits
                                            ) of
                                                {ok, Value} ->
                                                    {ok, Value};

                                                {error, E@1} ->
                                                    distribute@telemetry:emit(
                                                        {decode_failed,
                                                            distribute@codec:decode_error_to_string(
                                                                E@1
                                                            ),
                                                            decode_on_call_reply}
                                                    ),
                                                    {error,
                                                        {call_decode_failed,
                                                            E@1}}
                                            end
                                    end
                            end
                    end
            end
    end.

-file("src/distribute/global.gleam", 446).
?DOC(
    " Like `receive`, but uses `config.get().default_call_timeout_ms` as the\n"
    " timeout. Receive and call share the same \"wait for a message\" semantics,\n"
    " so they share the configured default rather than introducing a third\n"
    " timeout knob.\n"
).
-spec receive_default(global_subject(FYT)) -> {ok, FYT} |
    {error, distribute@codec:decode_error()}.
receive_default(Global) ->
    'receive'(Global, erlang:element(2, distribute@config:get())).

-file("src/distribute/global.gleam", 453).
?DOC(" Like `call`, but uses `config.get().default_call_timeout_ms` as the timeout.\n").
-spec call_default(
    global_subject(FYX),
    fun((gleam@erlang@process:subject(bitstring())) -> FYX),
    fun((bitstring()) -> {ok, FZA} | {error, distribute@codec:decode_error()})
) -> {ok, FZA} | {error, call_error()}.
call_default(Target, Make_request, Response_decoder) ->
    call(
        Target,
        Make_request,
        Response_decoder,
        erlang:element(2, distribute@config:get())
    ).

-file("src/distribute/global.gleam", 468).
-spec isolated_proxy_shutdown_grace_ms() -> integer().
isolated_proxy_shutdown_grace_ms() ->
    erlang:element(7, distribute@config:get()).

-file("src/distribute/global.gleam", 621).
?DOC(
    " Consume a late result tuple raced past our caller-side timeout.\n"
    " The proxy only ever sends one message, so a 0-timeout pull suffices.\n"
).
-spec drain_isolated_result(
    gleam@erlang@process:subject({ok, any()} | {error, call_error()})
) -> nil.
drain_isolated_result(Result_subject) ->
    Drain = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        gleam@erlang@process:select_map(
            _pipe,
            Result_subject,
            fun(_) -> nil end
        )
    end,
    _ = gleam_erlang_ffi:select(Drain, 0),
    nil.

-file("src/distribute/global.gleam", 519).
-spec call_isolated(
    global_subject(FZL),
    fun((gleam@erlang@process:subject(bitstring())) -> FZL),
    fun((bitstring()) -> {ok, FZO} | {error, distribute@codec:decode_error()}),
    integer()
) -> {ok, FZO} | {error, call_error()}.
call_isolated(Target, Make_request, Response_decoder, Timeout_ms) ->
    Clamped_timeout_ms = distribute_ffi_utils:clamp_timeout(Timeout_ms),
    Result_subject = gleam@erlang@process:new_subject(),
    Proxy_pid = proc_lib:spawn(
        fun() ->
            gleam@erlang@process:send(
                Result_subject,
                call(Target, Make_request, Response_decoder, Timeout_ms)
            )
        end
    ),
    Proxy_mon = gleam@erlang@process:monitor(Proxy_pid),
    Selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        _pipe@1 = gleam@erlang@process:select_map(
            _pipe,
            Result_subject,
            fun(Field@0) -> {isolated_ready, Field@0} end
        ),
        gleam@erlang@process:select_specific_monitor(
            _pipe@1,
            Proxy_mon,
            fun(_) -> isolated_proxy_down end
        )
    end,
    case gleam_erlang_ffi:select(Selector, Clamped_timeout_ms + 100) of
        {ok, {isolated_ready, Result}} ->
            gleam@erlang@process:demonitor_process(Proxy_mon),
            Result;

        {ok, isolated_proxy_down} ->
            distribute@telemetry:emit(call_proxy_crashed),
            {error, timeout};

        {error, nil} ->
            gleam@erlang@process:kill(Proxy_pid),
            Down_selector = begin
                _pipe@2 = gleam_erlang_ffi:new_selector(),
                gleam@erlang@process:select_specific_monitor(
                    _pipe@2,
                    Proxy_mon,
                    fun(_) -> nil end
                )
            end,
            case gleam_erlang_ffi:select(
                Down_selector,
                isolated_proxy_shutdown_grace_ms()
            ) of
                {ok, nil} ->
                    nil;

                {error, nil} ->
                    gleam@erlang@process:demonitor_process(Proxy_mon)
            end,
            drain_isolated_result(Result_subject),
            distribute@telemetry:emit({call_timed_out, Clamped_timeout_ms}),
            {error, timeout}
    end.

-file("src/distribute/global.gleam", 506).
?DOC(
    " Mailbox-safe variant of `call`: each invocation runs the actual\n"
    " `call` inside a short-lived proxy process. The caller waits on a\n"
    " one-shot result Subject; when the proxy exits, its mailbox and any\n"
    " orphan late-reply tagged with the proxy-owned reply Subject are\n"
    " reaped by the BEAM at zero cost.\n"
    "\n"
    " Use this from long-running callers that issue many `call`s and\n"
    " would otherwise accumulate orphan messages under sustained\n"
    " timeouts: see the \"Late-reply caveat\" on `call/4`. For one-shot\n"
    " callers (CLI, scripts) and short-lived processes the plain\n"
    " `call/4` is cheaper.\n"
    "\n"
    " The proxy is **spawned unlinked** and tracked via `process.monitor`.\n"
    " An earlier draft used `process.spawn` (linked); a linked proxy that\n"
    " somehow exits abnormally would propagate the exit signal back to the\n"
    " caller, defeating the isolation guarantee. With unlinked + monitor,\n"
    " the caller observes either the result (`Ok(...)`), an unexpected\n"
    " proxy `DOWN` (returned as `Error(Timeout)`), or its own receive\n"
    " timeout buffer expiring. The inner `call` could not produce a reply.\n"
    " On caller-side timeout, `call_isolated` now kills the proxy, waits for\n"
    " its `DOWN`, and only then drains the result Subject; this closes the\n"
    " late-send race where a still-running proxy could otherwise pollute the\n"
    " caller mailbox after return. In every path the caller stays alive.\n"
    "\n"
    " Costs: one extra process spawn + one cross-process hop per call.\n"
    " On the BEAM this is microseconds.\n"
    " Like `call_isolated`, with `config.get().default_call_timeout_ms`.\n"
).
-spec call_isolated_default(
    global_subject(FZE),
    fun((gleam@erlang@process:subject(bitstring())) -> FZE),
    fun((bitstring()) -> {ok, FZH} | {error, distribute@codec:decode_error()})
) -> {ok, FZH} | {error, call_error()}.
call_isolated_default(Target, Make_request, Response_decoder) ->
    call_isolated(
        Target,
        Make_request,
        Response_decoder,
        erlang:element(2, distribute@config:get())
    ).

-file("src/distribute/global.gleam", 635).
?DOC(
    " Send a response through a reply subject. Used by the handler to answer a `call`.\n"
    "\n"
    " Returns `Error(PayloadTooLarge(size))` when the encoded payload exceeds\n"
    " `config.get().max_payload_size_bytes`. The message is never sent.\n"
).
-spec reply(
    gleam@erlang@process:subject(bitstring()),
    FZX,
    fun((FZX) -> {ok, bitstring()} | {error, distribute@codec:encode_error()})
) -> {ok, nil} | {error, send_error()}.
reply(Reply_to, Response, Encoder) ->
    case distribute@codec:encode(Encoder, Response) of
        {ok, Bits} ->
            Cap = erlang:element(4, distribute@config:get()),
            Size = erlang:byte_size(Bits),
            case Size > Cap of
                true ->
                    distribute@telemetry:emit(
                        {payload_rejected, Size, Cap, payload_on_reply}
                    ),
                    {error, {payload_too_large, Size}};

                false ->
                    gleam@erlang@process:send(Reply_to, Bits),
                    {ok, nil}
            end;

        {error, E} ->
            {error, {send_encode_failed, E}}
    end.

-file("src/distribute/global.gleam", 663).
-spec call_error_to_string(call_error()) -> binary().
call_error_to_string(Error) ->
    case Error of
        timeout ->
            <<"Call timed out"/utf8>>;

        target_down ->
            <<"Target process is down"/utf8>>;

        {call_encode_failed, E} ->
            <<"Call encode failed: "/utf8,
                (distribute@codec:encode_error_to_string(E))/binary>>;

        {call_decode_failed, E@1} ->
            <<"Call decode failed: "/utf8,
                (distribute@codec:decode_error_to_string(E@1))/binary>>;

        {call_payload_too_large, Size} ->
            <<<<"Call payload too large: "/utf8,
                    (erlang:integer_to_binary(Size))/binary>>/binary,
                " bytes"/utf8>>
    end.