distribute

-module(distribute@conflict).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/conflict.gleam").
-export([lowest_pid_wins/0, highest_pid_wins/0, keep_local/0, kill_both/0, node_priority/1]).
-export_type([conflict_outcome/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(
    " Conflict resolvers for `:global` registrations across split-brain\n"
    " healing.\n"
    "\n"
    " ## Why this module exists\n"
    "\n"
    " Erlang's `:global` registry is the backbone of `distribute`'s\n"
    " `register_global` path. It is also the part of the BEAM with the\n"
    " most teeth on a partitioned cluster: when a network split heals,\n"
    " `:global` discovers any name that was claimed on both halves and\n"
    " resolves the conflict by sending an **uncatchable**\n"
    " `exit(loser, kill)` to one of the two PIDs, picked by a\n"
    " resolver function.\n"
    "\n"
    " The default resolver (`:global.random_notify_name/3`) is a coin\n"
    " flip. For workloads where the choice matters. A leader that\n"
    " must always sit on a specific node, a router whose state lives\n"
    " on a particular shard, a singleton whose oldest instance is\n"
    " Authoritative. Coin-flip resolution turns split-brain healing\n"
    " into a non-deterministic outage.\n"
    "\n"
    " `register_global_with_resolver` (in `distribute/registry`) lets\n"
    " you swap that default for a function you control. This module\n"
    " supplies the `Resolver` type and a small set of pure built-in\n"
    " resolvers covering the most common policies. Users compose or\n"
    " write their own.\n"
    "\n"
    " ## Operational guard rails\n"
    "\n"
    " The resolver is invoked by `:global` *inside the global_name_server\n"
    " process*. A slow or panicking resolver stalls every other\n"
    " `:global` operation cluster-wide for the duration. The library\n"
    " wraps every user resolver in an FFI shim that:\n"
    "\n"
    " - Spawns a short-lived worker to run the user fn (so a panic\n"
    "   does not propagate into global_name_server).\n"
    " - Imposes a hard deadline from config (default **1 000 ms**, see\n"
    "   `conflict_ffi:default_resolver_timeout_ms/0`).\n"
    " - On timeout or crash, applies a deterministic fallback (lowest\n"
    "   term-ordered Pid wins) so the cluster never wedges.\n"
    " - Emits `telemetry.ConflictResolved` on every resolution and\n"
    "   `telemetry.ConflictResolverFailed` whenever the user fn\n"
    "   misbehaved and the fallback fired.\n"
    "\n"
    " Watch the latter event in production: a steady stream of\n"
    " `ConflictResolverFailed` means your resolver itself is broken.\n"
    " A steady stream of `ConflictResolved` (without failures) means\n"
    " the cluster is flapping. Partitions are healing repeatedly\n"
    " And the application can't fix it on its own.\n"
    "\n"
    " ## Data-loss honesty box\n"
    "\n"
    " The \"lowest term-ordered Pid wins\" fallback that fires after a\n"
    " resolver crash or timeout is **deterministic**, not **state-\n"
    " aware**. If your resolver was supposed to keep the side with\n"
    " the most recent state (an oldest-wins, an authoritative-leader,\n"
    " a longest-lived counter) and it fails for any reason, the\n"
    " fallback can pick the wrong PID and the cluster loses whatever\n"
    " state the loser was holding.\n"
    "\n"
    " Two pragmatic choices:\n"
    "\n"
    " - **Stateless / idempotent actors** (routers, dispatchers,\n"
    "   stateless coordinators): the fallback is fine, the cluster\n"
    "   converges to *some* PID, the loss is bounded.\n"
    " - **Stateful actors** (counters, caches, leaders with quorum\n"
    "   state): use `kill_both()` as the *primary* resolver, watch\n"
    "   `ConflictResolved(_, None)` in telemetry, and trigger an\n"
    "   application-level recovery path (re-elect, re-bootstrap from\n"
    "   durable storage). Better to admit \"no winner\" than silently\n"
    "   pick the side without the state.\n"
    "\n"
    " The library cannot know which category a given actor falls\n"
    " into, so the default fallback prioritises convergence over\n"
    " state preservation. This is the right call for the BEAM's\n"
    " \"let it crash\" baseline; it is the wrong call for systems\n"
    " where state divergence costs money. Pick consciously.\n"
    "\n"
    " ## What this does NOT do\n"
    "\n"
    " Conflict resolution at the registry level cannot reconcile the\n"
    " **state** of two diverged actors. The loser dies, the winner\n"
    " keeps its state, anything pending on the loser is gone.\n"
    " Reconciling state is the next layer up (snapshot the loser\n"
    " before kill, merge into the winner) and intentionally not in\n"
    " scope here. The plan to ship that as a higher-level\n"
    " \"merge_resolver\" composes on top of this module without changing\n"
    " the `Resolver` type.\n"
).

-type conflict_outcome() :: {keep, gleam@erlang@process:pid_()} | kill_both.

-file("src/distribute/conflict.gleam", 136).
?DOC(
    " Pick the PID with the lowest Erlang term order. Deterministic,\n"
    " no I/O, no node-locality bias. Usable as a tiebreaker even\n"
    " when no semantic policy fits the workload.\n"
    "\n"
    " Term ordering on Pids is stable across the cluster but not\n"
    " human-meaningful: it is essentially \"which Pid was created\n"
    " first on the lower-numbered node\". Use this when you need\n"
    " reproducibility, not when you need a specific side to win.\n"
).
-spec lowest_pid_wins() -> fun((binary(), gleam@erlang@process:pid_(), gleam@erlang@process:pid_()) -> conflict_outcome()).
lowest_pid_wins() ->
    fun(_, Pid1, Pid2) -> case erlang:'<'(Pid1, Pid2) of
            true ->
                {keep, Pid1};

            false ->
                {keep, Pid2}
        end end.

-file("src/distribute/conflict.gleam", 148).
?DOC(
    " Pick the PID with the highest Erlang term order. Mirror image\n"
    " of `lowest_pid_wins`. Mostly useful when paired with another\n"
    " strategy in a fallback chain.\n"
).
-spec highest_pid_wins() -> fun((binary(), gleam@erlang@process:pid_(), gleam@erlang@process:pid_()) -> conflict_outcome()).
highest_pid_wins() ->
    fun(_, Pid1, Pid2) -> case erlang:'<'(Pid1, Pid2) of
            true ->
                {keep, Pid2};

            false ->
                {keep, Pid1}
        end end.

-file("src/distribute/conflict.gleam", 166).
?DOC(
    " Always keep the PID running on the local node, if either does.\n"
    " Falls back to `lowest_pid_wins` when neither (or both) PID is\n"
    " Local. The resolver runs on the global_name_server worker,\n"
    " which is co-located with one of the two PIDs only if that\n"
    " side is the lock-holder.\n"
    "\n"
    " Useful in deployments where one node is the canonical \"primary\"\n"
    " and remote duplicates from a partition heal should always\n"
    " defer.\n"
).
-spec keep_local() -> fun((binary(), gleam@erlang@process:pid_(), gleam@erlang@process:pid_()) -> conflict_outcome()).
keep_local() ->
    fun(Name, Pid1, Pid2) ->
        P1_local = conflict_ffi_helpers:pid_node_is_local(Pid1),
        P2_local = conflict_ffi_helpers:pid_node_is_local(Pid2),
        case {P1_local, P2_local} of
            {true, false} ->
                {keep, Pid1};

            {false, true} ->
                {keep, Pid2};

            {true, true} ->
                (lowest_pid_wins())(Name, Pid1, Pid2);

            {false, false} ->
                (lowest_pid_wins())(Name, Pid1, Pid2)
        end
    end.

-file("src/distribute/conflict.gleam", 196).
?DOC(
    " Refuse to pick. Both PIDs die, the name is removed from\n"
    " `:global`, and the next `lookup` returns `Error(Nil)` until\n"
    " one side re-registers.\n"
    "\n"
    " Use this for **stateful** actors where killing the wrong PID\n"
    " would lose authoritative state and the application has its\n"
    " own reconciliation path (re-elect a leader, re-bootstrap from\n"
    " durable storage). The other built-in resolvers\n"
    " (`lowest_pid_wins`, `keep_local`, `node_priority`) all preserve\n"
    " *one* PID at the cost of potentially picking the wrong one;\n"
    " `kill_both` is the safer choice when \"wrong choice\" is worse\n"
    " than \"no choice\".\n"
    "\n"
    " Operational pairing: subscribe to `cluster_monitor` and watch\n"
    " `telemetry.ConflictResolved(_, None)` events; on each, trigger\n"
    " the application-specific recovery (e.g. re-register from a\n"
    " snapshot, kick off a leader-election round).\n"
).
-spec kill_both() -> fun((binary(), gleam@erlang@process:pid_(), gleam@erlang@process:pid_()) -> conflict_outcome()).
kill_both() ->
    fun(_, _, _) -> kill_both end.

-file("src/distribute/conflict.gleam", 247).
-spec rank_loop(list(binary()), binary(), integer(), integer()) -> integer().
rank_loop(Order, Node, Idx, Len) ->
    case Order of
        [] ->
            Len + 1;

        [Head | Rest] ->
            case Head =:= Node of
                true ->
                    Idx;

                false ->
                    rank_loop(Rest, Node, Idx + 1, Len + 1)
            end
    end.

-file("src/distribute/conflict.gleam", 240).
?DOC(
    " Rank of a node name in the priority list. Index 0 = highest\n"
    " priority. Names not in the list rank as `length + 1` so any\n"
    " listed node beats any unlisted node.\n"
).
-spec node_rank(list(binary()), binary()) -> integer().
node_rank(Order, Node) ->
    case Order of
        [] ->
            0;

        _ ->
            rank_loop(Order, Node, 0, 0)
    end.

-file("src/distribute/conflict.gleam", 209).
?DOC(
    " Pick by static node priority. The resolver checks each PID's\n"
    " node against the supplied list (highest priority first) and\n"
    " keeps the one whose node appears earlier. PIDs on nodes not in\n"
    " the list lose to PIDs on nodes that are; ties are broken by\n"
    " `lowest_pid_wins`.\n"
    "\n"
    " Use this in deployments with a clear hierarchy. e.g.\n"
    " `[\"primary@host\", \"secondary@host\"]`, where you can name\n"
    " the authoritative side ahead of time.\n"
).
-spec node_priority(list(binary())) -> fun((binary(), gleam@erlang@process:pid_(), gleam@erlang@process:pid_()) -> conflict_outcome()).
node_priority(Order) ->
    fun(Name, Pid1, Pid2) ->
        P1_rank = node_rank(Order, conflict_ffi_helpers:pid_node_name(Pid1)),
        P2_rank = node_rank(Order, conflict_ffi_helpers:pid_node_name(Pid2)),
        case P1_rank =:= P2_rank of
            true ->
                (lowest_pid_wins())(Name, Pid1, Pid2);

            false ->
                case P1_rank < P2_rank of
                    true ->
                        {keep, Pid1};

                    false ->
                        {keep, Pid2}
                end
        end
    end.