defmodule Aerospike.Cluster.NodeSupervisor do @moduledoc false @default_pool_size 10 # Aerospike's default `proto-fd-idle-ms` is 60_000 ms: the server # drops client sockets that sit idle longer than that. Evict at 55_000 # so the client closes first and the next checkout sees a fresh # worker instead of a server-initiated RST. @default_idle_timeout_ms 55_000 # Cap how many idle workers NimblePool drops per ping cycle so a big # pool cannot lose every worker in one tick. 2 matches the guidance # in NimblePool's docs and keeps at least some capacity hot through # a quiet period. @default_max_idle_pings 2 @doc false def child_spec(opts) when is_list(opts) do name = Keyword.fetch!(opts, :name) %{ id: {__MODULE__, name}, start: {__MODULE__, :start_link, [opts]}, type: :supervisor, restart: :permanent, shutdown: :infinity } end @doc """ Starts the per-cluster `DynamicSupervisor`. Options: * `:name` — cluster name atom. The supervisor is registered under `sup_name/1`. """ @spec start_link(keyword()) :: Supervisor.on_start() def start_link(opts) when is_list(opts) do name = Keyword.fetch!(opts, :name) is_atom(name) or raise ArgumentError, "Aerospike.Cluster.NodeSupervisor: :name must be an atom" DynamicSupervisor.start_link(strategy: :one_for_one, name: sup_name(name)) end @doc """ Returns the registered name atom used for the cluster's NodeSupervisor. """ @spec sup_name(atom()) :: atom() def sup_name(name) when is_atom(name), do: :"#{name}_node_sup" @doc """ Starts an `Aerospike.Cluster.NodePool` child under this supervisor. Options: * `:node_name` — string identifier of the Aerospike node (required). * `:transport` — module implementing `Aerospike.Cluster.NodeTransport` (required). * `:host` — host the pool workers connect to (required). * `:port` — TCP port the pool workers connect to (required). * `:connect_opts` — keyword list passed as the third argument to `transport.connect/3` (required). * `:pool_size` — positive integer, defaults to `#{@default_pool_size}`. * `:min_connections_per_node` — non-negative integer floor requested by startup policy. `0` starts workers lazily; positive values keep the existing eager pool behavior, which maintains at least that many reusable workers. * `:idle_timeout_ms` — milliseconds a worker may sit idle before NimblePool evicts it via `handle_ping/2`. Defaults to `#{@default_idle_timeout_ms}` to stay under Aerospike's default `proto-fd-idle-ms` of 60_000 ms. * `:max_idle_pings` — positive integer bounding how many idle workers NimblePool may drop per verification cycle. Defaults to `#{@default_max_idle_pings}`. * `:counters` — optional `Aerospike.Cluster.NodeCounters.t()` reference owned by the `Aerospike.Cluster.Tender`. When present, pool callbacks increment `:in_flight` on checkout and decrement on checkin / cancellation, and bump `:failed` when the caller signals a transport-class failure via `{:close, :failure}`. Omitting this key leaves counter writes as no-ops, which is the intended behaviour for tests and cluster-state-only modes. * `:cluster_name` — optional cluster identity atom used by internal runtime metrics hooks. When absent, connection metrics are skipped. * `:features` — optional `MapSet` of capability tokens captured from the node's `features` info-key reply at registration. Pool consumers (capability-gated send paths) read this to decide whether to enable optional features per node. Defaults to an empty `MapSet`. """ @spec start_pool(pid() | atom(), keyword()) :: DynamicSupervisor.on_start_child() def start_pool(supervisor, opts) when is_list(opts) do node_name = Keyword.fetch!(opts, :node_name) transport = Keyword.fetch!(opts, :transport) host = Keyword.fetch!(opts, :host) port = Keyword.fetch!(opts, :port) connect_opts = Keyword.fetch!(opts, :connect_opts) pool_size = Keyword.get(opts, :pool_size, @default_pool_size) min_connections_per_node = Keyword.get(opts, :min_connections_per_node, pool_size) idle_timeout_ms = Keyword.get(opts, :idle_timeout_ms, @default_idle_timeout_ms) max_idle_pings = Keyword.get(opts, :max_idle_pings, @default_max_idle_pings) counters = Keyword.get(opts, :counters) cluster_name = Keyword.get(opts, :cluster_name) features = Keyword.get(opts, :features, MapSet.new()) worker_opts = [ transport: transport, host: host, port: port, connect_opts: connect_opts, node_name: node_name, counters: counters, cluster_name: cluster_name, features: features ] lazy? = min_connections_per_node == 0 child = %{ id: {:node_pool, node_name}, start: {NimblePool, :start_link, [ [ worker: {Aerospike.Cluster.NodePool, worker_opts}, pool_size: pool_size, lazy: lazy?, worker_idle_timeout: idle_timeout_ms, max_idle_pings: max_idle_pings ] ]}, restart: :temporary, shutdown: 5_000 } case sup_pid(supervisor) do nil -> {:error, :not_found} sup -> DynamicSupervisor.start_child(sup, child) end end @doc """ Terminates the pool child identified by `pool_pid`. Returns `:ok` on success, `{:error, :not_found}` when the pid is not a child of this supervisor (including when the supervisor itself is not reachable by name). """ @spec stop_pool(pid() | atom(), pid()) :: :ok | {:error, :not_found} def stop_pool(supervisor, pool_pid) when is_pid(pool_pid) do case sup_pid(supervisor) do nil -> {:error, :not_found} sup -> case DynamicSupervisor.terminate_child(sup, pool_pid) do :ok -> :ok {:error, _} -> {:error, :not_found} end end end defp sup_pid(supervisor) when is_atom(supervisor), do: Process.whereis(supervisor) defp sup_pid(pid) when is_pid(pid), do: pid end