defmodule Aerospike.Cluster.Tender do @moduledoc false use GenServer require Logger alias Aerospike.Cluster.Node, as: ClusterNode alias Aerospike.Cluster.NodeCounters alias Aerospike.Cluster.NodeSupervisor alias Aerospike.Cluster.PartitionMap alias Aerospike.Cluster.PartitionMapMerge alias Aerospike.Cluster.PartitionMapWriter alias Aerospike.Cluster.TableOwner alias Aerospike.Cluster.TendHistogram alias Aerospike.Error alias Aerospike.Policy alias Aerospike.RuntimeMetrics alias Aerospike.Telemetry @default_tend_interval_ms 1_000 @default_failure_threshold 5 @default_pool_size 10 # Default breaker thresholds. `:circuit_open_threshold` mirrors the # Tender's own `failure_threshold` semantics by default — once roughly # as many transport failures pile up as the tender would tolerate # before demoting the node, stop sending new work. `:max_concurrent_ops` # defaults to `pool_size * 10`: generous enough that ordinary # steady-state traffic never trips the cap, but tight enough that a # wedged pool cannot queue a thundering herd. Both are overridable # through `Aerospike.start_link/1`. @default_circuit_open_threshold 10 @default_max_concurrent_multiplier 10 @type seed :: {String.t(), :inet.port_number()} @type namespace :: String.t() @typedoc """ Start options: * `:name` — registered name (required). * `:transport` — module implementing `Aerospike.Cluster.NodeTransport` (required). * `:seeds` — list of `{host, port}` tuples (required, non-empty). * `:namespaces` — list of configured namespaces the cluster must serve before `:ready` flips to `true` (required, non-empty). * `:tables` — `%{owners: atom(), node_gens: atom(), meta: atom(), txn_tracking: atom()}` map of ETS table names, as returned by `Aerospike.Cluster.TableOwner.tables/1` (required). The Tender does not create tables; it reads and writes the tables created by the TableOwner so the partition map survives a Tender restart. * `:connect_opts` — keyword forwarded verbatim to `transport.connect/3`. Default `[]`. * `:failure_threshold` — consecutive refresh failures before a node is dropped. Default `#{@default_failure_threshold}`. * `:tend_interval_ms` — period between automatic tend cycles. Default `#{@default_tend_interval_ms}`. * `:tend_trigger` — `:timer` (default) or `:manual`. In `:manual` mode no timer is started and tests drive cycles with `tend_now/1`. * `:node_supervisor` — registered name (atom) or pid of the `Aerospike.Cluster.NodeSupervisor` used to start per-node connection pools. When absent, pool lifecycle is skipped entirely — the Tender runs with info sockets only, which is the mode the cluster-state invariant tests rely on. * `:pool_size` — pool workers per node. Default `#{@default_pool_size}`. Ignored when `:node_supervisor` is absent. * `:idle_timeout_ms` — milliseconds a pooled worker may sit idle before NimblePool evicts it via `handle_ping/2`. Forwarded to `Aerospike.Cluster.NodeSupervisor.start_pool/2`. Defaults to the value chosen by the supervisor (stays below Aerospike's `proto-fd-idle-ms` of 60_000 ms). Ignored when `:node_supervisor` is absent. * `:max_idle_pings` — positive integer bounding how many idle workers NimblePool may drop per verification cycle. Forwarded to `Aerospike.Cluster.NodeSupervisor.start_pool/2`. Defaults to the value chosen by the supervisor. Ignored when `:node_supervisor` is absent. * `:circuit_open_threshold` — non-negative integer. The `Aerospike.Cluster.CircuitBreaker` refuses a command when the node's `:failed` counter has reached this value. The Tender zeroes the `:failed` slot on every successful tend cycle for the node so the failure window decays with cluster-state health. Default `#{@default_circuit_open_threshold}`. * `:max_concurrent_ops_per_node` — positive integer. The breaker refuses a command when `in_flight + queued` has reached this value. Default `pool_size * #{@default_max_concurrent_multiplier}`. * `:max_retries` — non-negative integer. Default retry attempts after the initial send before `Aerospike.Command.Get` gives up. `0` disables retry. Default `2`. Per-command overrides via `Aerospike.get/3` opts. * `:sleep_between_retries_ms` — non-negative integer. Fixed sleep between retry attempts. Default `0` (no backoff; jittered backoff is not implemented). * `:replica_policy` — `:master` or `:sequence`. Default `:sequence` so retries walk the replica list; set to `:master` to pin every attempt to the master replica. * `:use_compression` — boolean, default `false`. Cluster-wide opt-in for outbound AS_MSG compression. When `true`, command dispatch asks the underlying transport to compress requests above the reference threshold — but only against nodes whose `features` capability set advertises `:compression`. The per-node gating happens inside the Tender, not at the call site, so a cluster with mixed-capability nodes never sends a compressed frame to a node that cannot decode it. * `:use_services_alternate` — boolean, default `false`. Cluster-wide toggle between the `peers-clear-std` and `peers-clear-alt` info keys during peer discovery. `peers-clear-alt` surfaces the server's alternate-access endpoints (configured via `alternate-access-address`), which is the route a client on a different subnet than the server's primary NIC has to take to reach every node. The toggle applies only to peer discovery — the seed list in `:seeds` is dialled verbatim — and is static for the lifetime of the cluster, matching Go `ClientPolicy.UseServicesAlternate`. * `:auth_mode` — `:internal` (default), `:external`, or `:pki`. * `:cluster_name` — expected server `cluster-name` value. When set, seed and peer nodes whose info response does not match are rejected. * `:login_timeout_ms` — login/authenticate read deadline in milliseconds. * `:seed_only_cluster` — when `true`, the Tender connects only to configured seed addresses and skips peer discovery. * `:application_id` — client application identity sent through the server user-agent info command when the connected server supports it. * `:user` / `:password` — cluster-wide session-login credentials. When both are present, the Tender performs a full password login on every fresh info socket it opens (seed bootstrap and peer discovery), caches the returned session token per node, and forwards the token to `NodeSupervisor.start_pool/2` so pool workers authenticate via the cached token instead of paying a bcrypt round trip. Must be strings or both absent. """ @type option :: {:name, GenServer.name()} | {:transport, module()} | {:seeds, [seed(), ...]} | {:namespaces, [namespace(), ...]} | {:tables, TableOwner.tables()} | {:writer, atom() | pid()} | {:connect_opts, keyword()} | {:failure_threshold, pos_integer()} | {:tend_interval_ms, pos_integer()} | {:tend_trigger, :timer | :manual} | {:node_supervisor, atom() | pid()} | {:pool_size, pos_integer()} | {:idle_timeout_ms, pos_integer()} | {:max_idle_pings, pos_integer()} | {:circuit_open_threshold, non_neg_integer()} | {:max_concurrent_ops_per_node, pos_integer()} | {:max_retries, non_neg_integer()} | {:sleep_between_retries_ms, non_neg_integer()} | {:replica_policy, :master | :sequence} | {:use_compression, boolean()} | {:use_services_alternate, boolean()} | {:auth_mode, :internal | :external | :pki} | {:cluster_name, String.t()} | {:login_timeout_ms, pos_integer()} | {:min_connections_per_node, non_neg_integer()} | {:seed_only_cluster, boolean()} | {:application_id, String.t()} | {:user, String.t()} | {:password, String.t()} @spec start_link([option()]) :: GenServer.on_start() def start_link(opts) do name = Keyword.fetch!(opts, :name) GenServer.start_link(__MODULE__, opts, name: name) end @doc """ Runs one tend cycle synchronously. Returns once every ETS write from the cycle has been committed. """ @spec tend_now(GenServer.server()) :: :ok def tend_now(server) do GenServer.call(server, :tend_now, 30_000) end @doc """ Returns whether every configured namespace has a complete partition map. Read-side callers should prefer `Aerospike.Cluster.ready?/1` so the named cluster seam stays the public owner of published readiness. """ @spec ready?(GenServer.server()) :: boolean() def ready?(server) do GenServer.call(server, :ready?) end @doc """ Returns the names of the cluster ETS tables. Read-side callers should prefer `Aerospike.Cluster.tables/1`. This wrapper remains for Tender-owned restart/orchestration paths and compatibility callers that still need the published table names. """ @spec tables(GenServer.server()) :: %{ owners: atom(), node_gens: atom(), meta: atom(), txn_tracking: atom() } def tables(server) do GenServer.call(server, :tables) end @doc """ Returns the pid of the `Aerospike.Cluster.NodePool` serving `node_name`. Returns `{:error, :unknown_node}` if the node is not in the cluster's current view, is currently in the `:inactive` lifecycle state (pool has been stopped pending recovery), or if the Tender was started without a `:node_supervisor` (i.e. cluster-state-only mode used by some tests). Command modules use this to check out a pooled connection without re-opening a socket per call. """ @spec pool_pid(GenServer.server(), String.t()) :: {:ok, pid()} | {:error, :unknown_node} def pool_pid(server, node_name) when is_binary(node_name) do GenServer.call(server, {:pool_pid, node_name}) end @doc """ Returns the `Aerospike.Cluster.NodeCounters` reference allocated for `node_name`, or `{:error, :unknown_node}` if the node is not in the cluster's current view or is currently `:inactive`. The circuit breaker reads counter slots through this reference on the hot path; this call pays a GenServer hop only once per node, not per command. Returns `{:error, :unknown_node}` when the Tender was started without a `:node_supervisor` (no pool lifecycle is tracked). """ @spec node_counters(GenServer.server(), String.t()) :: {:ok, NodeCounters.t()} | {:error, :unknown_node} def node_counters(server, node_name) when is_binary(node_name) do GenServer.call(server, {:node_counters, node_name}) end @doc """ Returns the `Aerospike.Cluster.TendHistogram` reference for `node_name`. The histogram accumulates per-node partition-map-refresh latency samples (one sample per tend cycle that fetches `replicas` for the node). Readers call `Aerospike.Cluster.TendHistogram.percentile/2` or `count/1` directly against the returned reference; the GenServer hop is one-shot per node, not per query. Returns `{:error, :unknown_node}` if the node is not in the cluster's current view, is currently `:inactive`, or when the Tender was started without a `:node_supervisor` (cluster-state-only mode still allocates histograms — see the head of `allocate_histogram/1`). """ @spec tend_histogram(GenServer.server(), String.t()) :: {:ok, TendHistogram.t()} | {:error, :unknown_node} def tend_histogram(server, node_name) when is_binary(node_name) do GenServer.call(server, {:tend_histogram, node_name}) end @typedoc """ Everything a command path needs to dispatch one attempt against a node: * `:pool` — pid of the node's `Aerospike.Cluster.NodePool`. * `:counters` — the node's `Aerospike.Cluster.NodeCounters` reference. * `:breaker` — breaker thresholds the caller passes to `Aerospike.Cluster.CircuitBreaker.allow?/2`. * `:use_compression` — whether the command path should ask the transport to compress the request for this attempt. Computed as `cluster_use_compression and node_supports_compression?` at handle time, so a single cluster flag composes with the node's live `features` set without per-call policy state. * `:supports_compression` — whether the node advertised the compression feature, independent of the cluster default. * `:host`, `:port`, and `:connect_opts` — direct-connect details the stream path uses to open a dedicated socket for long-lived reads. """ @type node_handle :: %{ pool: pid(), counters: NodeCounters.t(), breaker: %{ circuit_open_threshold: non_neg_integer(), max_concurrent_ops_per_node: pos_integer() }, use_compression: boolean(), supports_compression: boolean(), host: String.t(), port: :inet.port_number(), connect_opts: keyword() } @doc """ Returns the pool pid, counters reference, breaker thresholds, and direct-connect details for `node_name` in one GenServer hop. Replaces the sequence of `pool_pid/2` + `node_counters/2` that the command path used previously; commands now pay one round trip per op instead of two. Returns `{:error, :unknown_node}` under the same conditions as `pool_pid/2` and `node_counters/2` (unknown node, `:inactive` node, or cluster-state-only mode with no pool). """ @spec node_handle(GenServer.server(), String.t()) :: {:ok, node_handle()} | {:error, :unknown_node} def node_handle(server, node_name) when is_binary(node_name) do GenServer.call(server, {:node_handle, node_name}) end @doc """ Returns a snapshot of every known node's lifecycle state. The map is keyed by node name and exposes the fields the retry and circuit-breaker layers consume: * `:status` — `:active | :inactive`. * `:failures` — consecutive refresh-failure counter. * `:recoveries` — number of times the node has flipped from `:inactive` back to `:active` across the Tender's lifetime. * `:last_tend_at` — monotonic milliseconds of the most recent tend stage that touched this node, or `nil` if none has run. * `:last_tend_result` — `:ok | :error | nil` classification of the outcome of that stage. * `:generation_seen` — most recent `partition-generation` value observed for the node, or `nil`. * `:counters` — the node's `Aerospike.Cluster.NodeCounters` reference, or `nil` when the Tender was started without a `:node_supervisor` (cluster-state-only mode). * `:tend_histogram` — the node's `Aerospike.Cluster.TendHistogram` reference, allocated at registration and nilled on lifecycle demotion. Callers that only want percentiles should prefer `tend_histogram/2`; the field is exposed here so a diagnostic caller can inspect the raw slot counts. * `:features` — `MapSet` of capability tokens captured from the node's `features` info-key reply at registration. Recognised tokens (e.g. `:compression`, `:pipelining`) are atoms; unrecognised tokens are preserved as `{:unknown, raw_string}` tuples. An empty set means either a probe failure or a server that advertises no capabilities the client knows about. This call is intended for tests and diagnostics; hot-path readers use ETS (`owners`, `node_gens`, `meta`) or `node_counters/2` instead. """ @spec nodes_status(GenServer.server()) :: %{optional(String.t()) => map()} def nodes_status(server) do GenServer.call(server, :nodes_status) end @doc """ Returns the transport module this Tender dispatches I/O through. Command modules pair this with `pool_pid/2`: checkout a pool worker, then call `transport.command/2` on the checked-out connection. The value is fixed at `start_link/1` time and does not change across tend cycles. """ @spec transport(GenServer.server()) :: module() def transport(server) do GenServer.call(server, :transport) end @doc false @spec auth_credentials(GenServer.server()) :: %{ user: String.t() | nil, password: String.t() | nil } def auth_credentials(server) do GenServer.call(server, :auth_credentials) end @doc false @spec rotate_auth_credentials(GenServer.server(), String.t(), String.t()) :: :ok def rotate_auth_credentials(server, user, password) when is_binary(user) and is_binary(password) do GenServer.call(server, {:rotate_auth_credentials, user, password}) end @impl GenServer def init(opts) do name = Keyword.fetch!(opts, :name) transport = Keyword.fetch!(opts, :transport) seeds = Keyword.fetch!(opts, :seeds) namespaces = Keyword.fetch!(opts, :namespaces) tables = Keyword.fetch!(opts, :tables) writer = Keyword.get(opts, :writer, PartitionMapWriter.via(name)) seeds != [] or raise ArgumentError, "Aerospike.Cluster.Tender: :seeds must be non-empty" namespaces != [] or raise ArgumentError, "Aerospike.Cluster.Tender: :namespaces must be non-empty" %{owners: owners_tab, node_gens: node_gens_tab, meta: meta_tab} = tables pool_size = Keyword.get(opts, :pool_size, @default_pool_size) breaker_opts = %{ circuit_open_threshold: Keyword.get(opts, :circuit_open_threshold, @default_circuit_open_threshold), max_concurrent_ops_per_node: Keyword.get( opts, :max_concurrent_ops_per_node, pool_size * @default_max_concurrent_multiplier ) } %Policy.ClusterDefaults{retry: retry_opts} = Policy.cluster_defaults(opts) :ok = PartitionMapWriter.publish_retry_policy(writer, retry_opts) :ok = PartitionMapWriter.publish_active_nodes(writer, []) :ok = RuntimeMetrics.init(name, pool_size: pool_size, tend_interval_ms: Keyword.get(opts, :tend_interval_ms, @default_tend_interval_ms) ) user = Keyword.get(opts, :user) password = Keyword.get(opts, :password) validate_auth_pair!(user, password) state = %{ name: name, transport: transport, connect_opts: normalized_connect_opts(opts), seeds: seeds, namespaces: namespaces, failure_threshold: Keyword.get(opts, :failure_threshold, @default_failure_threshold), tend_interval_ms: Keyword.get(opts, :tend_interval_ms, @default_tend_interval_ms), tend_trigger: Keyword.get(opts, :tend_trigger, :timer), node_supervisor: Keyword.get(opts, :node_supervisor), pool_size: pool_size, min_connections_per_node: Keyword.get(opts, :min_connections_per_node, pool_size), idle_timeout_ms: Keyword.get(opts, :idle_timeout_ms), max_idle_pings: Keyword.get(opts, :max_idle_pings), breaker_opts: breaker_opts, retry_opts: retry_opts, use_compression: Keyword.get(opts, :use_compression, false), use_services_alternate: Keyword.get(opts, :use_services_alternate, false), auth_mode: Keyword.get(opts, :auth_mode, :internal), cluster_name: Keyword.get(opts, :cluster_name), seed_only_cluster: Keyword.get(opts, :seed_only_cluster, false), application_id: Keyword.get(opts, :application_id), user: user, password: password, tables: tables, owners_tab: owners_tab, node_gens_tab: node_gens_tab, meta_tab: meta_tab, writer: writer, nodes: %{}, peers_refresh_needed?: false, ready?: read_ready(meta_tab) } cleanup_orphan_pools(state) {:ok, maybe_schedule_tend(state)} end defp validate_auth_pair!(nil, nil), do: :ok defp validate_auth_pair!(user, password) when is_binary(user) and is_binary(password), do: :ok defp validate_auth_pair!(_, _), do: raise(ArgumentError, message: "Aerospike.Cluster.Tender: :user and :password must both be strings or both be absent" ) @impl GenServer def handle_call(:tend_now, _from, state) do {:reply, :ok, run_tend(state)} end def handle_call(:ready?, _from, state) do {:reply, state.ready?, state} end def handle_call(:tables, _from, state) do {:reply, state.tables, state} end def handle_call({:pool_pid, node_name}, _from, state) do case Map.fetch(state.nodes, node_name) do {:ok, %{status: :active, pool_pid: pid}} when is_pid(pid) -> {:reply, {:ok, pid}, state} _ -> {:reply, {:error, :unknown_node}, state} end end def handle_call({:node_counters, node_name}, _from, state) do case Map.fetch(state.nodes, node_name) do {:ok, %{status: :active, counters: ref}} when not is_nil(ref) -> {:reply, {:ok, ref}, state} _ -> {:reply, {:error, :unknown_node}, state} end end def handle_call({:tend_histogram, node_name}, _from, state) do case Map.fetch(state.nodes, node_name) do {:ok, %{status: :active, tend_histogram: ref}} when not is_nil(ref) -> {:reply, {:ok, ref}, state} _ -> {:reply, {:error, :unknown_node}, state} end end def handle_call({:node_handle, node_name}, _from, state) do case Map.fetch(state.nodes, node_name) do {:ok, %{ status: :active, pool_pid: pool, counters: counters, node: %ClusterNode{host: host, port: port, session: session, features: features} }} when is_pid(pool) and not is_nil(counters) -> handle = %{ pool: pool, counters: counters, breaker: state.breaker_opts, use_compression: state.use_compression and MapSet.member?(features, :compression), supports_compression: MapSet.member?(features, :compression), host: host, port: port, connect_opts: Keyword.put(pool_connect_opts(state, session), :node_name, node_name) } {:reply, {:ok, handle}, state} _ -> {:reply, {:error, :unknown_node}, state} end end def handle_call(:transport, _from, state) do {:reply, state.transport, state} end def handle_call(:auth_credentials, _from, state) do {:reply, %{user: state.user, password: state.password}, state} end def handle_call({:rotate_auth_credentials, user, password}, _from, state) do {:reply, :ok, %{state | user: user, password: password}} end def handle_call(:nodes_status, _from, state) do snapshot = Map.new(state.nodes, fn {name, entry} -> {name, %{ status: entry.status, failures: entry.failures, recoveries: entry.recoveries, last_tend_at: entry.last_tend_at, last_tend_result: entry.last_tend_result, generation_seen: entry.node.generation_seen, counters: entry.counters, tend_histogram: entry.tend_histogram, features: entry.node.features }} end) {:reply, snapshot, state} end @impl GenServer def handle_info(:tend, state) do state = run_tend(state) {:noreply, maybe_schedule_tend(state)} end @impl GenServer def terminate(_reason, state) do Enum.each(state.nodes, fn {_, entry} -> ClusterNode.close(state.transport, entry.node.conn) end) :ok end ## Tend cycle defp run_tend(state) do new_state = :telemetry.span(Telemetry.tend_cycle_span(), %{}, fn -> next_state = %{state | peers_refresh_needed?: false} |> bootstrap_seeds() |> refresh_nodes() |> maybe_discover_peers() |> maybe_refresh_partition_maps() |> publish_active_nodes() |> recompute_ready() {next_state, %{}} end) RuntimeMetrics.record_tend(state.name, :ok) new_state catch kind, reason -> RuntimeMetrics.record_tend(state.name, :error) :erlang.raise(kind, reason, __STACKTRACE__) end # Calls `discover_peers/1` only when any event this cycle flagged that a # peer-list refresh is warranted. Matching the C, Go, and Java clients, # the trigger fires when at least one active node's `peers-generation` # changed this cycle (including the "freshly-seen" sentinel case where # the cached generation is `nil`), or when a new node was registered # mid-cycle from the seed or peer-discovery paths. defp maybe_discover_peers(%{peers_refresh_needed?: false} = state), do: state defp maybe_discover_peers(state), do: discover_peers(state) # Only fetches `replicas` once every `:active` node that produced a # `cluster-stable` hash this cycle agrees on the same hash. Disagreement # means the cluster is mid-transition; writing partition-map entries on # top of that transition risks mixing replica lists from different # cluster views, which is exactly the poisoning the guard prevents. # # Wrapped in a partition-map-refresh span so operators can separate # the cost of fetching partition-generation (already inside # `refresh_nodes/1`) from the cost of fetching and decoding the full # `replicas` payload, which dominates wall-clock time in steady state. defp maybe_refresh_partition_maps(state) do :telemetry.span(Telemetry.partition_map_refresh_span(), %{}, fn -> new_state = case PartitionMapMerge.verify_cluster_stable(cluster_stable_contributors(state)) do {:ok, _hash_or_empty} -> refresh_partition_maps(state) {:error, :disagreement, per_node_hashes} -> Logger.warning( "Aerospike.Cluster.Tender: cluster-stable disagreement; skipping partition-map refresh: " <> inspect(per_node_hashes) ) state end {new_state, %{}} end) end # Builds the `%{node_name => hash}` map `PartitionMapMerge.verify_cluster_stable/1` # consumes. Inactive nodes are excluded by construction (they cannot serve # traffic), as are active nodes whose info call failed this cycle (they # carry a nil hash after `clear_cluster_stable/1` zeroed it at cycle start). defp cluster_stable_contributors(state) do state.nodes |> Enum.filter(fn {_, entry} -> entry.status == :active and is_binary(entry.node.cluster_stable) end) |> Map.new(fn {name, entry} -> {name, entry.node.cluster_stable} end) end # Re-runs seed bootstrap for every configured seed not currently covered # by an `:active` node at the same direct-connect host:port. This keeps # configured seeds as a recovery path even when the cluster is only # partially degraded: if one restarted seed comes back with a new node id # while other nodes remain healthy, the Tender does not have to wait for # peer discovery to notice it. defp bootstrap_seeds(state) do Enum.reduce(state.seeds, state, fn {host, port}, acc -> if seed_bootstrap_needed?(acc, host, port) do bootstrap_seed(acc, host, port) else acc end end) end defp seed_bootstrap_needed?(state, host, port) do not Enum.any?(state.nodes, fn {_name, %{status: :active, node: %ClusterNode{host: ^host, port: ^port}}} -> true _ -> false end) end # Dials `{host, port}` and registers the resulting Node struct, unless # the server reports a node name already present in `state.nodes`. The # seed list is dialled verbatim from `connect_opts`, so # `service-clear-alt` does not belong here — peer discovery is the only # place the alternate-services toggle takes effect. defp bootstrap_seed(state, host, port) do case ClusterNode.seed( state.transport, host, port, state.connect_opts, auth_opts(state) ) do {:ok, node} -> case Map.fetch(state.nodes, node.name) do {:ok, %{status: :active}} -> ClusterNode.close(state.transport, node.conn) state {:ok, %{status: status}} when status in [:inactive] -> state |> drop_node(node.name) |> drop_stale_seed_nodes(host, port) |> register_new_node(node, :bootstrap) :error -> state |> drop_stale_seed_nodes(host, port) |> register_new_node(node, :bootstrap) end {:error, %Error{} = err} -> Logger.warning( "Aerospike.Cluster.Tender: seed #{host}:#{port} bootstrap failed: #{err.message}" ) state {:error, :no_node_info} -> Logger.warning("Aerospike.Cluster.Tender: seed #{host}:#{port} missing 'node' info") state end end defp drop_stale_seed_nodes(state, host, port) do state.nodes |> Enum.reduce(state, fn {name, %{status: status, node: %ClusterNode{host: ^host, port: ^port}}}, acc when status != :active -> drop_node(acc, name) _, acc -> acc end) end defp normalized_connect_opts(opts) do opts |> Keyword.get(:connect_opts, []) |> put_connect_opt(:login_timeout_ms, Keyword.get(opts, :login_timeout_ms)) |> put_connect_opt(:auth_mode, Keyword.get(opts, :auth_mode)) end defp put_connect_opt(connect_opts, _key, nil), do: connect_opts defp put_connect_opt(connect_opts, key, value), do: Keyword.put(connect_opts, key, value) defp auth_opts(state) do [ user: state.user, password: state.password, auth_mode: state.auth_mode, login_timeout_ms: Keyword.get(state.connect_opts, :login_timeout_ms), cluster_name: state.cluster_name, application_id: state.application_id ] end defp register_new_node(state, %ClusterNode{} = node, reason) when reason in [:bootstrap, :peer_discovery] do # Allocate counters *before* starting the pool so the pool's # callbacks see the reference from the first init_worker call. # In cluster-state-only mode (`:node_supervisor` absent) no pool # runs so no writer ever touches a counters ref; skipping the # allocation there keeps `:counters` consistently nil for that # mode and matches the "counters exist iff a pool exists" # invariant used by the breaker. counters = allocate_counters(state) # The tend-latency histogram is always allocated (cluster-state-only # mode still runs tend cycles and fetches partition maps, so the # sampling call site runs regardless of whether a pool exists). tend_histogram = TendHistogram.new() case ensure_pool(state, node, counters) do {:ok, pool_pid} -> entry = %{ node: node, status: :active, failures: 0, recoveries: 0, last_tend_at: nil, last_tend_result: nil, pool_pid: pool_pid, counters: counters, tend_histogram: tend_histogram } emit_transition(node.name, :unknown, :active, reason) RuntimeMetrics.record_node_added(state.name, node.name) %{ state | nodes: Map.put(state.nodes, node.name, entry), peers_refresh_needed?: true } :error -> ClusterNode.close(state.transport, node.conn) state end end # Refreshes every known node's `partition-generation` and `cluster-stable` # values in a single combined info call per node. The cycle starts by # clearing the previous cycle's `cluster_stable` hash on every node so # that `verify_cluster_stable/1` only ever reads hashes captured in the # current cycle (a node that fails this cycle's info call cannot # contribute a stale hash to the agreement check). defp refresh_nodes(state) do state = clear_cluster_stable(state) Enum.reduce(state.nodes, state, fn {name, _}, acc -> case Map.fetch(acc.nodes, name) do {:ok, entry} -> refresh_node(acc, entry) :error -> acc end end) end defp clear_cluster_stable(state) do nodes = Map.new(state.nodes, fn {name, entry} -> {name, %{entry | node: ClusterNode.clear_cluster_stable(entry.node)}} end) %{state | nodes: nodes} end defp refresh_node(state, entry) do case ClusterNode.refresh(entry.node, state.transport) do {:ok, updated_node, %{partition_generation: gen, peers_generation_changed?: peers_changed?}} -> state = put_node_struct(state, updated_node) state = maybe_flag_peers_refresh(state, peers_changed?) store_node_gen_if_changed(state, updated_node.name, gen) register_success(state, updated_node.name) {:error, %Error{} = err} -> Logger.debug(fn -> "Aerospike.Cluster.Tender: #{entry.node.name} refresh-node info failed: #{err.message}" end) register_failure(state, entry.node.name) {:error, :malformed_reply} -> register_failure(state, entry.node.name) end end defp maybe_flag_peers_refresh(state, true), do: %{state | peers_refresh_needed?: true} defp maybe_flag_peers_refresh(state, false), do: state # Replaces the `%ClusterNode{}` embedded in `state.nodes[name]` without # touching any lifecycle field. Used by every stage that calls a Node # function and expects the updated observables to land back in state. defp put_node_struct(state, %ClusterNode{name: name} = node) do case Map.fetch(state.nodes, name) do {:ok, entry} -> %{state | nodes: Map.put(state.nodes, name, %{entry | node: node})} :error -> state end end # Records a successful info reply against the node record. Clears the # `failures` counter; if the node was `:inactive`, flips it back to # `:active` and increments `:recoveries` so the recovery count survives # future transitions. Also stamps `last_tend_at`/`last_tend_result`. # # Zeroes the per-node `:failed` counter as the tender-side "failure # window decay" for the circuit breaker. The info socket just answered # cleanly, which is the Tender's strongest statement about node health — # any pool-path transport failures recorded before this cycle are stale # relative to that statement. The breaker in # `Aerospike.Cluster.CircuitBreaker.allow?/2` re-reads `:failed` on every # command so the reset is observed immediately by the next caller. defp register_success(state, name) do now = monotonic_ms() update_entry(state, name, fn entry -> {status, recoveries} = case entry.status do :active -> {:active, entry.recoveries} :inactive -> emit_transition(name, :inactive, :active, :recovery) {:active, entry.recoveries + 1} end reset_failed_counter(entry.counters) %{ entry | failures: 0, status: status, recoveries: recoveries, last_tend_at: now, last_tend_result: :ok } end) end defp reset_failed_counter(nil), do: :ok defp reset_failed_counter(ref), do: NodeCounters.reset_failed(ref) defp register_peers_success(state, name) do now = monotonic_ms() update_entry(state, name, fn entry -> {status, recoveries} = case entry.status do :active -> {:active, entry.recoveries} :inactive -> emit_transition(name, :inactive, :active, :recovery) {:active, entry.recoveries + 1} end %{ entry | failures: 0, status: status, recoveries: recoveries, last_tend_at: now, last_tend_result: :ok } end) end defp register_replicas_success(state, name) do now = monotonic_ms() update_entry(state, name, fn entry -> %{entry | failures: 0, last_tend_at: now, last_tend_result: :ok} end) end defp store_node_gen_if_changed(state, name, gen) do case PartitionMap.get_node_gen(state.node_gens_tab, name) do {:ok, ^gen} -> :ok _ -> PartitionMapWriter.put_node_gen(state.writer, name, gen) end end # Bumps the per-node failure counter and drives lifecycle transitions: # # * `:active` ⇒ `:inactive` — on reaching `failure_threshold`. Stops # the pool, clears the node's `node_gens` / `owners` rows, but # leaves the node in `state.nodes` with its `:conn` intact so the # next tend cycle can probe it via `partition-generation` and flip # it back to `:active`. Resets `failures` to 0 so the grace-window # accounting is distinct from the active-state breach budget. # # * `:inactive` ⇒ removed — any failure while `:inactive` removes # the node entry entirely (`drop_node/2`). The pool was already # stopped and the node's ETS rows cleared by the earlier # `:active` ⇒ `:inactive` transition; this path only drops the # in-memory entry and closes `:conn`. A single grace cycle is # enough: the only thing the Tender does for an `:inactive` node # is probe `partition-generation` in `refresh_nodes/1`, so "any # failure" and "second consecutive threshold breach" collapse to # the same event. defp register_failure(state, name) do now = monotonic_ms() state = update_entry(state, name, fn entry -> %{ entry | failures: entry.failures + 1, last_tend_at: now, last_tend_result: :error } end) case Map.fetch(state.nodes, name) do {:ok, %{status: :inactive}} -> drop_node(state, name) {:ok, %{failures: failures, status: :active}} when failures >= state.failure_threshold -> mark_inactive(state, name) _ -> state end end # Flips `name` from `:active` to `:inactive`. Stops the pool so inflight # checkouts see `{:error, :unknown_node}` on the next attempt, clears # the node's `owners`/`node_gens` rows so the Router stops routing to # it, resets the `failures` counter so the next tend cycle can either # recover or fall through to a full drop, and drops the node's # counters / tend-histogram references — the pool is gone, the breaker # cannot sensibly read stale slots, and a fresh recovery will allocate # new references via `register_new_node/3`. The `:atomics` term is # GC'd once no process retains it. defp mark_inactive(state, name) do case Map.fetch(state.nodes, name) do {:ok, entry} -> drop_pool(state, entry) PartitionMapWriter.delete_node_gen(state.writer, name) PartitionMapWriter.drop_node(state.writer, name) cleared_node = %ClusterNode{ entry.node | session: nil, generation_seen: nil, applied_gen: nil, peers_generation_seen: nil } updated = %{ entry | node: cleared_node, status: :inactive, failures: 0, pool_pid: nil, counters: nil, tend_histogram: nil } emit_transition(name, :active, :inactive, :failure_threshold) %{state | nodes: Map.put(state.nodes, name, updated)} :error -> state end end # Fully removes `name` from `state.nodes` and closes its info socket. # Any pool has already been stopped by the earlier `mark_inactive/2` # transition; the ETS rows are already absent for the same reason. # Counters were cleared at the inactive transition, so dropping the # node-map entry here is enough — the GC reclaims the :counters ref # once no process retains it. defp drop_node(state, name) do case Map.pop(state.nodes, name) do {nil, _} -> state {entry, nodes} -> drop_pool(state, entry) ClusterNode.close(state.transport, entry.node.conn) PartitionMapWriter.delete_node_gen(state.writer, name) PartitionMapWriter.drop_node(state.writer, name) RuntimeMetrics.record_node_removed(state.name, name) emit_transition(name, :inactive, :unknown, :dropped) %{state | nodes: nodes} end end defp update_entry(state, name, fun) do case Map.fetch(state.nodes, name) do {:ok, entry} -> %{state | nodes: Map.put(state.nodes, name, fun.(entry))} :error -> state end end ## Peer discovery # Iterates over every known `:active` node (ordered by node name) asking # for the configured peer-discovery info key. Every parseable reply # contributes to the peer set — we do not stop at the first one. This # avoids the "first-by-term-order" failure mode where a stale or broken # lead node masks real topology. Nodes flipped to `:inactive` earlier in # the same cycle are skipped; their grace-cycle recovery probe happens # in `refresh_nodes/1` of the *next* cycle. defp discover_peers(%{seed_only_cluster: true} = state), do: state defp discover_peers(state) do ordered = state.nodes |> sorted_entries() |> Enum.filter(&(&1.status == :active)) {state, peers_by_name} = Enum.reduce(ordered, {state, %{}}, &collect_peers/2) Enum.reduce(peers_by_name, state, fn {_name, peer}, acc -> ensure_peer_connected(acc, peer) end) end defp collect_peers(entry, {state, peers_acc}) do case ClusterNode.refresh_peers(entry.node, state.transport, use_services_alternate: state.use_services_alternate ) do {:ok, _node, peers} -> {register_peers_success(state, entry.node.name), merge_peers(peers_acc, peers)} {:error, %Error{} = err} -> Logger.debug(fn -> key = ClusterNode.peer_info_key(state.use_services_alternate) "Aerospike.Cluster.Tender: #{entry.node.name} #{key} failed: #{err.message}" end) {register_failure(state, entry.node.name), peers_acc} {:error, :malformed_reply} -> {register_failure(state, entry.node.name), peers_acc} end end defp merge_peers(acc, peers) do Enum.reduce(peers, acc, fn peer, m -> Map.put_new(m, peer.node_name, peer) end) end defp ensure_peer_connected(state, %{node_name: name, host: host, port: port}) do case Map.has_key?(state.nodes, name) do true -> state false -> connect_and_register_peer(state, name, host, port) end end defp connect_and_register_peer(state, name, host, port) do case state.transport.connect(host, port, state.connect_opts) do {:ok, conn} -> login_and_register_peer(state, name, host, port, conn) {:error, %Error{} = err} -> Logger.warning( "Aerospike.Cluster.Tender: could not connect to peer #{name} at #{host}:#{port}: #{err.message}" ) state end end # Peers learn their `name` from the parent node's `peers-clear-std` reply # rather than from a fresh `node` info probe, matching Go's # `node.go:dialNode` behaviour. Only the `features` key is fetched here; # `ClusterNode.fetch_features/3` collapses a probe failure to an empty # MapSet so the peer still registers. defp login_and_register_peer(state, name, host, port, conn) do case ClusterNode.login(state.transport, conn, auth_opts(state)) do {:ok, session} -> case ClusterNode.fetch_peer_features(state.transport, conn, name, auth_opts(state)) do {:ok, features} -> peer_node = %ClusterNode{ name: name, host: host, port: port, conn: conn, session: session, features: features, generation_seen: nil, applied_gen: nil, cluster_stable: nil, peers_generation_seen: nil } register_new_node(state, peer_node, :peer_discovery) {:error, %Error{} = err} -> Logger.warning( "Aerospike.Cluster.Tender: peer #{name} at #{host}:#{port} validation failed: #{err.message}" ) ClusterNode.close(state.transport, conn) state end {:error, %Error{} = err} -> Logger.warning( "Aerospike.Cluster.Tender: peer #{name} at #{host}:#{port} login failed: #{err.message}" ) ClusterNode.close(state.transport, conn) state end end ## Partition map refresh defp refresh_partition_maps(state) do state.nodes |> sorted_entries() |> Enum.filter(&(&1.status == :active)) |> Enum.reduce(state, fn entry, acc -> case Map.fetch(acc.nodes, entry.node.name) do {:ok, %{status: :active} = current} -> maybe_refresh_partition_map(acc, current) _ -> acc end end) end # Short-circuits the `replicas` info call when the node's last observed # `partition-generation` matches the one whose replicas reply was last # successfully applied. A newly-discovered peer has `applied_gen: nil` # (and typically `generation_seen: nil` since its `partition-generation` # has not been fetched yet) — both conditions force a fetch so the first # cycle always installs the node's ownership entries. defp maybe_refresh_partition_map(state, entry) do if partition_map_fetch_needed?(entry.node) do fetch_and_apply_replicas(state, entry) else state end end defp partition_map_fetch_needed?(%ClusterNode{generation_seen: nil}), do: true defp partition_map_fetch_needed?(%ClusterNode{applied_gen: nil}), do: true defp partition_map_fetch_needed?(%ClusterNode{generation_seen: same, applied_gen: same}), do: false defp partition_map_fetch_needed?(%ClusterNode{}), do: true defp fetch_and_apply_replicas(state, entry) do # Measure the full replicas fetch + apply for this node. Per-node # latency is the useful operator signal (a single slow node shows up # here before it does in the cluster-wide tend_cycle span), so the # sample lands in the node's own histogram rather than a cluster # aggregate. start_native = System.monotonic_time() result = case ClusterNode.refresh_partitions(entry.node, state.transport) do {:ok, _node, segments} -> apply_replicas(state, entry, segments) {:error, %Error{} = err} -> Logger.debug(fn -> "Aerospike.Cluster.Tender: #{entry.node.name} replicas failed: #{err.message}" end) register_failure(state, entry.node.name) {:error, :malformed_reply} -> register_failure(state, entry.node.name) end record_tend_sample(entry.tend_histogram, System.monotonic_time() - start_native) result end defp record_tend_sample(nil, _duration_native), do: :ok defp record_tend_sample(ref, duration_native), do: TendHistogram.record(ref, duration_native) # Advances `applied_gen` to the node's `generation_seen` only when every # segment is accepted by the regime guard. If any segment is rejected as # stale the applied generation stays at its prior value so the next tend # cycle refetches `replicas` and retries the merge. defp apply_replicas(state, entry, segments) do applied? = PartitionMapWriter.apply_segments( state.writer, entry.node.name, segments, state.namespaces ) state = register_replicas_success(state, entry.node.name) if applied? do RuntimeMetrics.record_partition_map_update(state.name) mark_partition_map_applied(state, entry.node.name, entry.node.generation_seen) else state end end defp mark_partition_map_applied(state, name, gen) do update_entry(state, name, fn entry -> %{entry | node: ClusterNode.mark_partition_map_applied(entry.node, gen)} end) end ## Ready flag defp publish_active_nodes(state) do active_nodes = state.nodes |> Enum.flat_map(fn {name, %{status: :active}} -> [name] _ -> [] end) :ok = PartitionMapWriter.publish_active_nodes(state.writer, active_nodes) state end defp recompute_ready(state) do ready? = PartitionMapWriter.recompute_ready(state.writer, state.namespaces) %{state | ready?: ready?} end ## Pool lifecycle # Starts a NodePool for `node_name` under the configured NodeSupervisor. # Returns `{:ok, pool_pid}` on success. Returns `:error` on pool-start # failure: a warning is logged, the node is skipped for this cycle, and # the caller is expected to *not* register the node. Pool-start failure # never bumps the per-node failure counter because the node is never added # to `state.nodes`. When the Tender runs without a `:node_supervisor` # (cluster-state-only mode), this returns a sentinel `{:ok, nil}` so # bootstrap/peer discovery still register the node — that path is used by # the invariant tests. defp allocate_counters(%{node_supervisor: nil}), do: nil defp allocate_counters(_state), do: NodeCounters.new() defp ensure_pool(%{node_supervisor: nil}, _node, _counters), do: {:ok, nil} defp ensure_pool(state, %ClusterNode{} = node, counters) do opts = [ node_name: node.name, transport: state.transport, host: node.host, port: node.port, connect_opts: pool_connect_opts(state, node.session), pool_size: state.pool_size, min_connections_per_node: state.min_connections_per_node, counters: counters, cluster_name: state.name, features: node.features ] |> maybe_put_pool_opt(:idle_timeout_ms, state.idle_timeout_ms) |> maybe_put_pool_opt(:max_idle_pings, state.max_idle_pings) case NodeSupervisor.start_pool(state.node_supervisor, opts) do {:ok, pool_pid} -> {:ok, pool_pid} {:error, reason} -> Logger.warning( "Aerospike.Cluster.Tender: could not start pool for #{node.name} at #{node.host}:#{node.port}: " <> "#{inspect(reason)}" ) :error end end # Builds the per-worker connect opts for a pool: # # * When no creds are configured → verbatim `state.connect_opts`. # * When creds are configured but the info socket did not produce a # session token (server has security disabled, or PKI-anonymous) → # forward `:user` + `:password` so workers can run a full password # login on their own socket. # * When a session token is cached → forward `:user` + `:session_token` # so workers run the cheap AUTHENTICATE path; the password stays in # case the server rejects the token with `:expired_session` and the # worker needs to fall back to a full login. defp pool_connect_opts(%{auth_mode: :pki} = state, nil) do Keyword.put(state.connect_opts, :auth_mode, :pki) end defp pool_connect_opts(%{auth_mode: :pki} = state, {token, _expires_at}) when is_binary(token) do state.connect_opts |> Keyword.put(:auth_mode, :pki) |> Keyword.put(:session_token, token) end defp pool_connect_opts(%{user: nil, password: nil} = state, _session), do: state.connect_opts defp pool_connect_opts(%{user: user, password: password} = state, nil) do state.connect_opts |> Keyword.put(:auth_mode, state.auth_mode) |> Keyword.put(:user, user) |> Keyword.put(:password, password) end defp pool_connect_opts(%{user: user, password: password} = state, {token, _expires_at}) when is_binary(token) do state.connect_opts |> Keyword.put(:auth_mode, state.auth_mode) |> Keyword.put(:user, user) |> Keyword.put(:password, password) |> Keyword.put(:session_token, token) end # Stops the pool for `entry`. Absence of a pool (no supervisor or a nil # pool_pid) is a no-op. `{:error, :not_found}` from the NodeSupervisor # is logged at `:debug` — a pool that already exited is an acceptable # terminal state. defp drop_pool(%{node_supervisor: nil}, _entry), do: :ok defp drop_pool(_state, %{pool_pid: nil}), do: :ok defp drop_pool(state, %{node: %ClusterNode{name: name}, pool_pid: pool_pid}) do case NodeSupervisor.stop_pool(state.node_supervisor, pool_pid) do :ok -> :ok {:error, :not_found} -> Logger.debug(fn -> "Aerospike.Cluster.Tender: pool for #{name} already gone" end) :ok end end # On Tender (re)start under `rest_for_one`, the NodeSupervisor survives. # Pools started by a previous Tender incarnation are still children of # that supervisor but are no longer reachable via the new Tender's # `state.nodes`. Kill them before the first tend cycle so the Tender's # view remains the single source of truth. defp cleanup_orphan_pools(%{node_supervisor: nil}), do: :ok defp cleanup_orphan_pools(state) do case sup_pid(state.node_supervisor) do nil -> :ok sup -> sup |> DynamicSupervisor.which_children() |> Enum.each(fn {_id, pid, _type, _mods} when is_pid(pid) -> _ = NodeSupervisor.stop_pool(state.node_supervisor, pid) _ -> :ok end) end end defp sup_pid(name) when is_atom(name), do: Process.whereis(name) defp sup_pid(pid) when is_pid(pid), do: pid # Only forward pool opts the caller set explicitly — absent values let # `NodeSupervisor.start_pool/2` pick its own defaults (which stay # below Aerospike's `proto-fd-idle-ms` for `:idle_timeout_ms`). defp maybe_put_pool_opt(opts, _key, nil), do: opts defp maybe_put_pool_opt(opts, key, value), do: Keyword.put(opts, key, value) ## Helpers defp sorted_entries(nodes) do nodes |> Map.values() |> Enum.sort_by(& &1.node.name) end defp maybe_schedule_tend(%{tend_trigger: :manual} = state), do: state defp maybe_schedule_tend(state) do Process.send_after(self(), :tend, state.tend_interval_ms) state end defp read_ready(meta_tab) do case :ets.lookup(meta_tab, :ready) do [{:ready, ready?}] -> ready? [] -> false end end defp monotonic_ms, do: System.monotonic_time(:millisecond) # Instant telemetry event fired whenever a node's lifecycle status # changes. The `:reason` enum is fixed at the call sites — see the # `Aerospike.Telemetry` moduledoc for the contract. Logs stay alongside # (operators still see the existing Logger output); this is purely # additive. defp emit_transition(name, from, to, reason) do :telemetry.execute( Telemetry.node_transition(), %{system_time: System.system_time()}, %{node_name: name, from: from, to: to, reason: reason} ) end end