-module(distribute@actor). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/distribute/actor.gleam"). -export([start_observed/5, start_registered_observed/5, start/4, start_default/3, start_registered/4, start_registered_default/3, child_spec/4, child_spec_default/3, start_registered_error_to_string/1, start_supervised/4, start_supervised_default/3, pool/5, pool_default/4, start_resource_owner/3]). -export_type([start_registered_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. ?MODULEDOC( " Named actor lifecycle: start, register, supervise, pool.\n" "\n" " ## Resource cleanup and the `{terminate, Reason}` gap\n" "\n" " `gleam/otp/actor` 1.x does **not** implement OTP's\n" " `{terminate, Reason}` system message\n" " (see `gleam_otp_external.erl`; there is an explicit `TODO`).\n" " That has two consequences for any actor that owns external\n" " resources (file handles, ETS tables, ports, locks, sockets):\n" "\n" " 1. **Handler-controlled exits run user code first.** If your\n" " handler returns `receiver.Stop` or `receiver.StopAbnormal`,\n" " you have full control: release every resource you own *before*\n" " returning the stop variant. The actor will then exit and the\n" " BEAM frees its mailbox.\n" " 2. **External terminations skip the actor.** When the actor is\n" " `process.kill`-ed (uncatchable), or when the supervisor sends\n" " `exit(Pid, shutdown)`, gleam/otp/actor does not currently\n" " convert it into a callback. Resources owned only on the\n" " actor's heap are released by the BEAM (memory), but any\n" " *external* handle (a file, a TCP connection, an ETS table you\n" " forgot to make public) leaks.\n" "\n" " **The OTP-pure mitigation** is the \"linked resource owner\"\n" " pattern: spawn a tiny dedicated process to own the external\n" " resource, monitor the actor, and run a `close` callback when the\n" " actor dies for any reason. The library ships\n" " `actor.start_resource_owner/3` as a ready-made helper for this\n" " pattern. See `docs/recipes.md` for the underlying recipe.\n" "\n" " When upstream gleam/otp ships termination callbacks (tracking\n" " issue: gleam-lang/otp#126) the helper becomes redundant for the\n" " actor case. Its API is shaped so the migration is a one-line\n" " change in user code and a deprecation here, no semantics shift.\n" ). -type start_registered_error() :: {actor_start_failed, gleam@otp@actor:start_error()} | {global_register_failed, distribute@registry:register_error()}. -file("src/distribute/actor.gleam", 84). ?DOC( " Two-phase orphan termination, mirroring `supervisor:terminate_child/2`.\n" "\n" " 1. Unlink so the brutal phase (if reached) does not propagate to the\n" " caller via the default actor link.\n" " 2. Send `exit(Pid, shutdown)`. Non-trapping actors die immediately\n" " with reason `shutdown`; the monitor we set fires within the grace\n" " window and we return cleanly. Trapping actors receive\n" " `{'EXIT', From, shutdown}` as a mailbox message.\n" " 3. If the orphan is still alive after `orphan_shutdown_grace_ms`,\n" " escalate to `process.kill` (uncatchable).\n" "\n" " ## Gleam OTP caveat\n" "\n" " `gleam/otp/actor` (as of 1.x) does not yet implement the\n" " `{terminate, Reason}` system message (see TODO in\n" " `gleam_otp_external.erl`), so a true OTP-style graceful shutdown via\n" " `sys:terminate/2` or `gen:stop/3` is currently a no-op for those\n" " actors. The system message is logged as \"unexpected\" and the actor\n" " keeps running until the kill phase. We use the simpler\n" " `exit(Pid, shutdown)` path: non-trapping actors die immediately,\n" " trapping actors fall through to brutal kill within the grace window.\n" " When upstream `gleam/otp` ships terminate support, switching to\n" " `gen:stop` here will give trapping actors a real `terminate` callback\n" " invocation without changing this function's contract.\n" "\n" " Used by `start_registered`, `start_registered_observed`, and `child_spec`\n" " when global registration fails after a successful actor init.\n" ). -spec terminate_orphan_gracefully(gleam@erlang@process:pid_()) -> nil. terminate_orphan_gracefully(Pid) -> gleam@erlang@process:unlink(Pid), Mon = gleam@erlang@process:monitor(Pid), _ = distribute_ffi_utils:exit_shutdown(Pid), Selector = begin _pipe = gleam_erlang_ffi:new_selector(), gleam@erlang@process:select_specific_monitor( _pipe, Mon, fun(_) -> nil end ) end, case gleam_erlang_ffi:select(Selector, 100) of {ok, nil} -> nil; {error, nil} -> gleam@erlang@process:kill(Pid), gleam@erlang@process:demonitor_process(Mon), distribute@telemetry:emit({orphan_kill_escalated, Pid}) end. -file("src/distribute/actor.gleam", 110). ?DOC( " Start a named actor, with a callback for decode errors.\n" "\n" " Useful for logging or metering malformed messages across nodes\n" " (e.g. during rolling deploys with mismatched codec versions).\n" ). -spec start_observed( distribute@registry:typed_name(HBH), HBJ, fun((HBH, HBJ) -> distribute@receiver:handler_step(HBJ)), integer(), fun((distribute@codec:decode_error()) -> nil) ) -> {ok, distribute@global:global_subject(HBH)} | {error, gleam@otp@actor:start_error()}. start_observed( Typed_name, Initial_state, Handler, Init_timeout_ms, On_decode_error ) -> case distribute@receiver:start_distributed_worker_observed( distribute@registry:typed_name_to_string(Typed_name), Initial_state, distribute@registry:typed_name_encoder(Typed_name), distribute@registry:typed_name_decoder(Typed_name), Handler, On_decode_error, Init_timeout_ms ) of {ok, Started} -> {ok, erlang:element(3, Started)}; {error, Err} -> {error, Err} end. -file("src/distribute/actor.gleam", 134). ?DOC(" Start a named actor and register it globally, with a callback for decode errors.\n"). -spec start_registered_observed( distribute@registry:typed_name(HBO), HBQ, fun((HBO, HBQ) -> distribute@receiver:handler_step(HBQ)), integer(), fun((distribute@codec:decode_error()) -> nil) ) -> {ok, distribute@global:global_subject(HBO)} | {error, start_registered_error()}. start_registered_observed( Typed_name, Initial_state, Handler, Init_timeout_ms, On_decode_error ) -> Name = distribute@registry:typed_name_to_string(Typed_name), case distribute@receiver:start_distributed_worker_observed( Name, Initial_state, distribute@registry:typed_name_encoder(Typed_name), distribute@registry:typed_name_decoder(Typed_name), Handler, On_decode_error, Init_timeout_ms ) of {ok, Started} -> case distribute@registry:register_global( Typed_name, erlang:element(3, Started) ) of {ok, _} -> {ok, erlang:element(3, Started)}; {error, Err} -> terminate_orphan_gracefully(erlang:element(2, Started)), {error, {global_register_failed, Err}} end; {error, Err@1} -> {error, {actor_start_failed, Err@1}} end. -file("src/distribute/actor.gleam", 167). ?DOC(" Start a named actor.\n"). -spec start( distribute@registry:typed_name(HBV), HBX, fun((HBV, HBX) -> distribute@receiver:handler_step(HBX)), integer() ) -> {ok, distribute@global:global_subject(HBV)} | {error, gleam@otp@actor:start_error()}. start(Typed_name, Initial_state, Handler, Init_timeout_ms) -> case distribute@receiver:start_distributed_worker( distribute@registry:typed_name_to_string(Typed_name), Initial_state, distribute@registry:typed_name_encoder(Typed_name), distribute@registry:typed_name_decoder(Typed_name), Handler, Init_timeout_ms ) of {ok, Started} -> {ok, erlang:element(3, Started)}; {error, Err} -> {error, Err} end. -file("src/distribute/actor.gleam", 189). ?DOC(" Like `start`, but uses `config.get().default_init_timeout_ms` as the init timeout.\n"). -spec start_default( distribute@registry:typed_name(HCC), HCE, fun((HCC, HCE) -> distribute@receiver:handler_step(HCE)) ) -> {ok, distribute@global:global_subject(HCC)} | {error, gleam@otp@actor:start_error()}. start_default(Typed_name, Initial_state, Handler) -> start( Typed_name, Initial_state, Handler, erlang:element(3, distribute@config:get()) ). -file("src/distribute/actor.gleam", 306). ?DOC( " Start an actor and register it globally. Kills the actor if\n" " registration fails.\n" "\n" " See also: `start_registered_default/3` (configured init timeout),\n" " `start_registered_observed/5` (decode-error hook),\n" " `start_supervised/4` (auto-restart on crash).\n" ). -spec start_registered( distribute@registry:typed_name(HDC), HDE, fun((HDC, HDE) -> distribute@receiver:handler_step(HDE)), integer() ) -> {ok, distribute@global:global_subject(HDC)} | {error, start_registered_error()}. start_registered(Typed_name, Initial_state, Handler, Init_timeout_ms) -> Name = distribute@registry:typed_name_to_string(Typed_name), case distribute@receiver:start_distributed_worker( Name, Initial_state, distribute@registry:typed_name_encoder(Typed_name), distribute@registry:typed_name_decoder(Typed_name), Handler, Init_timeout_ms ) of {ok, Started} -> case distribute@registry:register_global( Typed_name, erlang:element(3, Started) ) of {ok, _} -> {ok, erlang:element(3, Started)}; {error, Err} -> terminate_orphan_gracefully(erlang:element(2, Started)), {error, {global_register_failed, Err}} end; {error, Err@1} -> {error, {actor_start_failed, Err@1}} end. -file("src/distribute/actor.gleam", 203). ?DOC(" Like `start_registered`, but uses `config.get().default_init_timeout_ms` as the init timeout.\n"). -spec start_registered_default( distribute@registry:typed_name(HCJ), HCL, fun((HCJ, HCL) -> distribute@receiver:handler_step(HCL)) ) -> {ok, distribute@global:global_subject(HCJ)} | {error, start_registered_error()}. start_registered_default(Typed_name, Initial_state, Handler) -> start_registered( Typed_name, Initial_state, Handler, erlang:element(3, distribute@config:get()) ). -file("src/distribute/actor.gleam", 231). ?DOC(" OTP child spec for a named actor that auto-registers on (re)start.\n"). -spec child_spec( distribute@registry:typed_name(HCW), HCY, fun((HCW, HCY) -> distribute@receiver:handler_step(HCY)), integer() ) -> gleam@otp@supervision:child_specification(distribute@global:global_subject(HCW)). child_spec(Typed_name, Initial_state, Handler, Init_timeout_ms) -> Name = distribute@registry:typed_name_to_string(Typed_name), gleam@otp@supervision:worker( fun() -> case distribute@receiver:start_distributed_worker( Name, Initial_state, distribute@registry:typed_name_encoder(Typed_name), distribute@registry:typed_name_decoder(Typed_name), Handler, Init_timeout_ms ) of {ok, Started} -> case distribute@registry:register_global( Typed_name, erlang:element(3, Started) ) of {ok, _} -> {ok, Started}; {error, _} -> terminate_orphan_gracefully( erlang:element(2, Started) ), {error, {init_failed, <<"registration_failed"/utf8>>}} end; {error, Err} -> {error, Err} end end ). -file("src/distribute/actor.gleam", 217). ?DOC(" Like `child_spec`, with `config.get().default_init_timeout_ms`.\n"). -spec child_spec_default( distribute@registry:typed_name(HCQ), HCS, fun((HCQ, HCS) -> distribute@receiver:handler_step(HCS)) ) -> gleam@otp@supervision:child_specification(distribute@global:global_subject(HCQ)). child_spec_default(Typed_name, Initial_state, Handler) -> child_spec( Typed_name, Initial_state, Handler, erlang:element(3, distribute@config:get()) ). -file("src/distribute/actor.gleam", 289). ?DOC( " Render a `StartRegisteredError` as a human-readable string. Mirrors\n" " the `*_error_to_string` formatters published by the other error\n" " modules so observability paths can `io.println` an error directly\n" " without pattern-matching at every call site.\n" ). -spec start_registered_error_to_string(start_registered_error()) -> binary(). start_registered_error_to_string(Err) -> case Err of {actor_start_failed, init_timeout} -> <<"Actor init timed out"/utf8>>; {actor_start_failed, {init_failed, Reason}} -> <<"Actor init failed: "/utf8, Reason/binary>>; {actor_start_failed, {init_exited, _}} -> <<"Actor init process exited"/utf8>>; {global_register_failed, Re} -> <<"Global registration failed: "/utf8, (distribute@registry:register_error_to_string(Re))/binary>> end. -file("src/distribute/actor.gleam", 354). ?DOC( " Start a supervised actor that auto-registers on (re)start.\n" "\n" " If registration fails, the worker crashes and the supervisor retries,\n" " which is the correct OTP pattern for transient registration failures.\n" ). -spec start_supervised( distribute@registry:typed_name(HDP), HDR, fun((HDP, HDR) -> distribute@receiver:handler_step(HDR)), integer() ) -> {ok, gleam@erlang@process:pid_()} | {error, gleam@otp@actor:start_error()}. start_supervised(Typed_name, Initial_state, Handler, Init_timeout_ms) -> Spec = child_spec(Typed_name, Initial_state, Handler, Init_timeout_ms), Builder = begin _pipe = gleam@otp@static_supervisor:new(one_for_one), gleam@otp@static_supervisor:add(_pipe, Spec) end, case gleam@otp@static_supervisor:start(Builder) of {ok, Sup} -> {ok, erlang:element(2, Sup)}; {error, Err} -> {error, Err} end. -file("src/distribute/actor.gleam", 337). ?DOC(" Like `start_supervised`, with `config.get().default_init_timeout_ms`.\n"). -spec start_supervised_default( distribute@registry:typed_name(HDJ), HDL, fun((HDJ, HDL) -> distribute@receiver:handler_step(HDL)) ) -> {ok, gleam@erlang@process:pid_()} | {error, gleam@otp@actor:start_error()}. start_supervised_default(Typed_name, Initial_state, Handler) -> start_supervised( Typed_name, Initial_state, Handler, erlang:element(3, distribute@config:get()) ). -file("src/distribute/actor.gleam", 575). -spec do_index_list(integer(), list(integer())) -> list(integer()). do_index_list(I, Acc) -> case I < 1 of true -> Acc; false -> do_index_list(I - 1, [I | Acc]) end. -file("src/distribute/actor.gleam", 571). ?DOC( " Build `[1, 2, ..., n]` for `n >= 1`. Replacement for `list.range`,\n" " which was removed in `gleam_stdlib` 1.0.\n" ). -spec index_list(integer()) -> list(integer()). index_list(N) -> do_index_list(N, []). -file("src/distribute/actor.gleam", 404). ?DOC( " Start N supervised actors, registered as `name_1` .. `name_N`.\n" "\n" " Returns `Error(actor.InitFailed(...))` if `size < 1`. `list.range`\n" " would otherwise produce a degenerate list of weird worker names\n" " (e.g. `name_0`, `name_-1`) and silently start a useless supervisor.\n" "\n" " > [!WARNING]\n" " > **Cascading Failure Risk**:\n" " > `:global` and fixed pools do not mix perfectly. If a single worker\n" " > in the pool fails its global registration (e.g. because `name_4` is\n" " > legitimately held by a node on the other side of a network split),\n" " > the worker crashes. The pool supervisor will try to restart it. If the\n" " > conflict persists, the supervisor exhausts its `MaxR` restart intensity\n" " > and **crashes the entire pool**. This means a collision on 1 worker\n" " > brings down all N workers, causing the parent supervisor to restart the\n" " > whole pool. This architectural limit will be solved natively when the\n" " > `syn` registry backend is introduced in v4.2.0 (via `syn` process groups).\n" ). -spec pool( distribute@registry:typed_name(HEB), integer(), HED, fun((HEB, HED) -> distribute@receiver:handler_step(HED)), integer() ) -> {ok, gleam@erlang@process:pid_()} | {error, gleam@otp@actor:start_error()}. pool(Typed_name, Size, Initial_state, Handler, Init_timeout_ms) -> case Size < 1 of true -> {error, {init_failed, <<"pool size must be >= 1, got "/utf8, (erlang:integer_to_binary(Size))/binary>>}}; false -> Specs = begin _pipe = index_list(Size), gleam@list:map( _pipe, fun(I) -> Worker_tn = distribute@registry:pool_member( Typed_name, I ), child_spec( Worker_tn, Initial_state, Handler, Init_timeout_ms ) end ) end, Builder = gleam@list:fold( Specs, gleam@otp@static_supervisor:new(one_for_one), fun(Acc, Spec) -> gleam@otp@static_supervisor:add(Acc, Spec) end ), case gleam@otp@static_supervisor:start(Builder) of {ok, Sup} -> {ok, erlang:element(2, Sup)}; {error, Err} -> {error, Err} end end. -file("src/distribute/actor.gleam", 372). ?DOC(" Like `pool`, with `config.get().default_init_timeout_ms`.\n"). -spec pool_default( distribute@registry:typed_name(HDV), integer(), HDX, fun((HDV, HDX) -> distribute@receiver:handler_step(HDX)) ) -> {ok, gleam@erlang@process:pid_()} | {error, gleam@otp@actor:start_error()}. pool_default(Typed_name, Size, Initial_state, Handler) -> pool( Typed_name, Size, Initial_state, Handler, erlang:element(3, distribute@config:get()) ). -file("src/distribute/actor.gleam", 447). ?DOC( " How long the owner waits between liveness re-checks while watching\n" " `lifetime`. Picked to be long enough that an idle owner is essentially\n" " free, short enough that a `DOWN` we somehow missed (cosmic ray,\n" " runtime-tracing tool dropping the message) is detected within seconds.\n" ). -spec resource_owner_poll_ms() -> integer(). resource_owner_poll_ms() -> erlang:element(8, distribute@config:get()). -file("src/distribute/actor.gleam", 551). ?DOC( " Block the owner until the lifetime PID is dead. Long-polls the\n" " `DOWN` selector and re-checks `process.is_alive` on each timeout.\n" " Without the re-check, a buggy or instrumented runtime that loses a\n" " `DOWN` message would leave us waiting forever; with it, we converge\n" " on the truth within `resource_owner_poll_ms`.\n" ). -spec await_lifetime_death( gleam@erlang@process:selector(nil), gleam@erlang@process:pid_() ) -> nil. await_lifetime_death(Selector, Lifetime) -> case gleam_erlang_ffi:select(Selector, resource_owner_poll_ms()) of {ok, nil} -> nil; {error, nil} -> case erlang:is_process_alive(Lifetime) of false -> nil; true -> await_lifetime_death(Selector, Lifetime) end end. -file("src/distribute/actor.gleam", 528). ?DOC( " Spawn a dedicated process that owns an external resource and runs\n" " `close(resource)` when `lifetime` dies for any reason. Workaround\n" " for the missing `{terminate, Reason}` callback in\n" " `gleam/otp/actor` 1.x: external resources (database connections,\n" " distributed locks, NIF handles, OS pipes) cannot rely on the\n" " actor's exit to free them, but a dedicated observer process can.\n" "\n" " Returns the owner's PID, in case you need to stop the owner\n" " independently of the lifetime PID.\n" "\n" " ## Why this shape\n" "\n" " The resource is opened **inside the owner**, so it is BEAM-process-\n" " owned by the owner. That matters for resources whose lifetime is\n" " tied to a specific process at the runtime level (ETS tables, ports,\n" " `gen_tcp` controlling process). \n" "\n" " The owner is **linked** to the caller (which is usually the actor's \n" " `init` function). It also traps exits (`process.trap_exits(True)`). \n" " This guarantees a **fail-fast** behaviour: if `open()` crashes before \n" " it returns, the owner dies, and because it is linked, it pulls the \n" " actor down with it immediately. This prevents the \"partial failure\" \n" " scenario where the actor survives but its critical resource failed \n" " to initialize.\n" "\n" " The owner uses a `process.monitor` on `lifetime`. Monitor delivery\n" " is asynchronous but reliable: the BEAM guarantees a `DOWN` for\n" " every monitored PID, and a monitor on an already-dead PID fires\n" " immediately. That makes the helper safe to call right after the\n" " actor has started: even if there is a microsecond gap before the\n" " monitor goes up, the BEAM resolves it correctly.\n" "\n" " ## Usage\n" "\n" " ```gleam\n" " import distribute\n" " import distribute/actor as dist_actor\n" " import distribute/global\n" "\n" " let assert Ok(gs) =\n" " distribute.start_registered(name, init, handler)\n" " let assert Ok(actor_pid) = global.owner(gs)\n" " let _owner = dist_actor.start_resource_owner(\n" " fn() { open_postgres_pool() },\n" " fn(pool) { close_postgres_pool(pool) },\n" " actor_pid,\n" " )\n" " ```\n" "\n" " If the actor needs to *use* the resource, have `open` build a\n" " `process.Subject` the owner serves and pass it through your\n" " actor's `init`.\n" "\n" " ## Failure modes\n" "\n" " - `open` raises: the owner dies before the resource is opened. Because\n" " the owner is linked to the caller (the actor), the actor receives an\n" " exit signal and crashes immediately. This fail-fast behaviour avoids\n" " partial failures where the actor runs without its resource.\n" " - `close` raises: the owner dies after the panic, no further\n" " cleanup runs. The resource may be in a partially-closed state.\n" " - `lifetime` is already dead at call time: monitor fires `DOWN`\n" " immediately, `close` runs on the next scheduler tick.\n" " - The owner is `process.kill`-ed: cleanup is uncatchable, BEAM\n" " reaps the resource via process death (only effective for\n" " BEAM-process-tied resources). External handles leak. This is\n" " the same uncatchable-kill caveat OTP itself has.\n" "\n" " ## Future\n" "\n" " When `gleam/otp/actor` ships native termination callbacks\n" " (gleam-lang/otp#126), this helper becomes redundant for the\n" " actor case. The contract will not shift: callers can replace the\n" " `actor.start_resource_owner(open, close, pid)` line with the\n" " upstream API and delete the helper call site. The function will\n" " stay (deprecated) for the more general \"tie cleanup to any PID\"\n" " pattern.\n" ). -spec start_resource_owner( fun(() -> HEH), fun((HEH) -> nil), gleam@erlang@process:pid_() ) -> gleam@erlang@process:pid_(). start_resource_owner(Open, Close, Lifetime) -> proc_lib:spawn_link( fun() -> gleam_erlang_ffi:trap_exits(true), Resource = Open(), Mon = gleam@erlang@process:monitor(Lifetime), Down_selector = begin _pipe = gleam_erlang_ffi:new_selector(), gleam@erlang@process:select_specific_monitor( _pipe, Mon, fun(_) -> nil end ) end, await_lifetime_death(Down_selector, Lifetime), Close(Resource), nil end ).